alien_core/

sync.rs

//! Sync protocol types for agent ↔ manager communication.
//!
//! The agent periodically calls `POST /v1/sync` with a `SyncRequest` and
//! receives a `SyncResponse` containing the target deployment state.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

use crate::{
    DeploymentConfig, DeploymentState, ObservedInventoryBatch, ReleaseInfo, ResourceHeartbeat,
};

/// State of an Operator capability as observed inside the environment.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "kebab-case")]
pub enum OperatorCapabilityState {
    /// The Operator has the permission or local facility needed for the capability.
    Granted,
    /// The environment explicitly denied the capability.
    Denied,
    /// The capability does not apply in this environment.
    Unavailable,
}

/// Report-only Operator capability status.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct OperatorCapabilityReport {
    /// Stable capability key, such as `k8s-workloads` or `logs`.
    pub key: String,
    /// Whether the capability is currently usable.
    pub state: OperatorCapabilityState,
    /// Optional human-readable detail from the Operator.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
}

/// A single operation the Operator currently has loaded. Opaque identifiers
/// only — no tier, description, or other plugin business logic crosses into
/// this public crate.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ReportedOperation {
    /// Name of the plugin that owns this operation.
    pub plugin: String,
    /// Version of the plugin that owns this operation.
    pub plugin_version: String,
    /// Name of the operation within the plugin.
    pub name: String,
}

/// Report-only summary of the operations the Operator has loaded, used to
/// confirm a bundle sync actually took effect.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct OperationsReport {
    /// Hash of the enabled-plugin bundle set the Operator currently has
    /// loaded. Compared against the platform's target hash to detect drift.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub loaded_bundle_hash: Option<String>,
    /// Operations currently loaded and executable by the Operator.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub operations: Vec<ReportedOperation>,
}

/// Origin of the exact Operator image running this process.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "kebab-case")]
pub enum OperatorImageSource {
    /// The image came from an immutable generated package.
    Package,
    /// The image was configured directly by the installer.
    Configured,
}

/// Exact immutable Operator image identity observed by the running process.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct OperatorImageReport {
    /// How the installer selected this image.
    pub source: OperatorImageSource,
    /// Package ID when `source` is `package`; otherwise `null`.
    pub package_id: Option<String>,
    /// Package version when `source` is `package`; otherwise `null`.
    pub package_version: Option<String>,
    /// Exact OCI image reference in `repository@sha256:digest` form.
    pub image: String,
    /// Exact lowercase OCI digest in `sha256:digest` form.
    pub digest: String,
}

impl OperatorImageReport {
    /// Validate that this report identifies one immutable image and that its
    /// package fields agree with `source`.
    pub fn validate(&self) -> Result<(), &'static str> {
        let Some(encoded_digest) = self.digest.strip_prefix("sha256:") else {
            return Err("operator image digest must start with sha256:");
        };
        if encoded_digest.len() != 64
            || !encoded_digest
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        {
            return Err("operator image digest must contain 64 lowercase hex characters");
        }

        let Some((repository, image_digest)) = self.image.rsplit_once('@') else {
            return Err("operator image must use repository@sha256:digest form");
        };
        if repository.is_empty()
            || repository.chars().any(char::is_whitespace)
            || repository.contains('@')
            || image_digest != self.digest
        {
            return Err("operator image must contain one repository and the reported digest");
        }

        let package_id = self.package_id.as_deref().map(str::trim);
        let package_version = self.package_version.as_deref().map(str::trim);
        match self.source {
            OperatorImageSource::Package
                if package_id.is_some_and(|value| !value.is_empty())
                    && package_version.is_some_and(|value| !value.is_empty()) =>
            {
                Ok(())
            }
            OperatorImageSource::Package => {
                Err("package operator images require package ID and version")
            }
            OperatorImageSource::Configured
                if self.package_id.is_none() && self.package_version.is_none() =>
            {
                Ok(())
            }
            OperatorImageSource::Configured => {
                Err("configured operator images must not include package identity")
            }
        }
    }
}

/// One bundle the Operator needs to download to reach `targetBundleHash`.
/// The manager mints a short-lived presigned GET URL per bundle — the
/// Operator never holds real cloud storage credentials, mirroring the OCI
/// registry proxy's credential-injection pattern.
#[derive(Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct OperationsBundleDownload {
    /// Plugin name this bundle provides.
    pub plugin: String,
    /// Plugin version this bundle provides.
    pub plugin_version: String,
    /// Presigned URL to GET the bundle ZIP from. Short-lived.
    pub url: String,
    /// Environment the plugin process runs with: its settings, which may hold
    /// secrets. Store only in encrypted state, and never log the values.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub env: BTreeMap<String, String>,
}

impl std::fmt::Debug for OperationsBundleDownload {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OperationsBundleDownload")
            .field("plugin", &self.plugin)
            .field("plugin_version", &self.plugin_version)
            .field("env", &self.env.keys().collect::<Vec<_>>())
            .finish_non_exhaustive()
    }
}

/// Target operations-bundle set for the Operator to converge its loaded
/// plugin registry toward, independent of any release/config target — a
/// plugin can be enabled with no release change, so this is not nested under
/// `TargetDeployment`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct TargetOperationsBundleSet {
    /// Hash identifying this exact enabled-plugin set. Compare against
    /// `OperationsReport.loadedBundleHash` to detect drift.
    pub hash: String,
    /// Presigned downloads for every bundle in the target set. The Operator
    /// fetches only the ones it doesn't already have loaded at the right
    /// version; already-loaded bundles are harmless to re-download.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub bundles: Vec<OperationsBundleDownload>,
}

/// One release-independent container the Operator should run in its namespace.
/// The manager sends the complete set on every sync. An empty set removes
/// containers previously owned by this deployment.
#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct TargetDynamicContainer {
    pub name: String,
    pub generation: u64,
    pub image: String,
    pub cpu: String,
    pub memory: String,
    pub replicas: u32,
    pub ports: Vec<u16>,
    pub deleted: bool,
    #[serde(default)]
    pub env: BTreeMap<String, String>,
    #[serde(default)]
    pub secret_env: BTreeMap<String, String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub health_check: Option<DynamicContainerHealthCheck>,
    /// Stop an installed workload when its release no longer admits the image.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub suspended_reason: Option<String>,
}

// Never include secret values in sync diagnostics.
impl std::fmt::Debug for TargetDynamicContainer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TargetDynamicContainer")
            .field("name", &self.name)
            .field("generation", &self.generation)
            .field("image", &self.image)
            .field("replicas", &self.replicas)
            .finish_non_exhaustive()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct DynamicContainerHealthCheck {
    pub path: String,
    pub port: u16,
}

/// What the Operator observed after applying one target generation.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct DynamicContainerReport {
    pub name: String,
    pub generation: u64,
    pub status: DynamicContainerStatus,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "lowercase")]
pub enum DynamicContainerStatus {
    Pending,
    Running,
    Failing,
    Stopped,
}

/// Request sent by the agent to the manager during periodic sync.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SyncRequest {
    /// The deployment ID this agent is managing.
    pub deployment_id: String,
    /// Stable identity for this Operator process. Used to fence update work.
    #[serde(default)]
    pub session: String,
    /// Signals that this Operator persists and echoes execution claims.
    #[serde(default)]
    pub supports_execution_claims: bool,
    /// Signals that this Operator understands container tunnels. Older
    /// Operators reject stacks that declare one, so the manager leaves
    /// tunnels out of their targets.
    #[serde(default)]
    pub supports_tunnels: bool,
    /// Exact update claim returned by the previous sync response.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub execution_claim: Option<SyncExecutionClaim>,
    /// Current deployment state as seen by the agent.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub current_state: Option<DeploymentState>,
    /// Managed Alien resource status samples emitted by the Operator's deployment step.
    #[serde(
        default,
        rename = "resourceHeartbeats",
        skip_serializing_if = "Vec::is_empty"
    )]
    pub heartbeats: Vec<ResourceHeartbeat>,
    /// Observed raw-resource inventory batches successfully read by the Operator.
    #[serde(
        default,
        rename = "observedInventoryBatches",
        skip_serializing_if = "Vec::is_empty"
    )]
    pub observed_inventory_batches: Vec<ObservedInventoryBatch>,
    /// Report-only capabilities observed by the Operator.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub capabilities: Vec<OperatorCapabilityReport>,
    /// Version of the Operator binary reporting this sync.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operator_version: Option<String>,
    /// Report-only summary of the operations the Operator currently has
    /// loaded. Absent means the Operator does not yet support reporting
    /// this (older Operator versions), not that it has no operations.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operations_report: Option<OperationsReport>,
    /// Operations an Operator installed on its own declares through its
    /// environment, without setting values. Absent when its deployment's
    /// release declares them instead.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operations_config: Option<crate::OperationsConfig>,
}

/// Where the Operator read the application identity from.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub enum ObservedApplicationSource {
    /// Labels and pod statuses of the observed Kubernetes workloads.
    Kubernetes,
}

/// A container image running in one observed application workload.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ObservedApplicationImage {
    /// Inventory identity of the workload, matching the `rawIdentity` of its
    /// observed resource sample (for example `apps/v1:Deployment:shop:api`).
    pub workload: String,
    /// Container name within the workload.
    pub container: String,
    /// Image reference reported by the container runtime.
    pub image: String,
    /// Registry manifest digest in `sha256:<hex>` form, when the runtime
    /// reports one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub digest: Option<String>,
}

/// Application release the Operator observes running in its environment.
///
/// This identifies the customer's application, not the Operator: the
/// Operator's own image is reported separately as [`OperatorImageReport`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ObservedApplicationReport {
    /// Where this identity was read from.
    pub source: ObservedApplicationSource,
    /// Helm chart name from the workloads' `helm.sh/chart` label. Present only
    /// when every labelled workload names the same chart.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub chart_name: Option<String>,
    /// Helm chart version from the same label.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub chart_version: Option<String>,
    /// Distinct container images running in the observed workloads.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub images: Vec<ObservedApplicationImage>,
    /// Whether every workload kind could be listed. When `false`, the chart
    /// and images describe only the workloads the Operator could read.
    pub complete: bool,
    /// When the workloads were read.
    pub observed_at: DateTime<Utc>,
}

/// Extensible wire input for sync metadata that is not part of the
/// long-standing [`SyncRequest`] struct-literal contract.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SyncInput {
    #[serde(flatten)]
    request: SyncRequest,
    /// Exact immutable image identity injected by the installer. Absent for
    /// older installations that do not carry an image receipt.
    #[serde(skip_serializing_if = "Option::is_none")]
    operator_image: Option<OperatorImageReport>,
    /// Application release observed in the environment. Absent when the
    /// Operator observed none or predates this report.
    #[serde(skip_serializing_if = "Option::is_none")]
    application: Option<ObservedApplicationReport>,
    /// Absent for older Operators. An empty report means no containers remain.
    #[serde(skip_serializing_if = "Option::is_none")]
    dynamic_containers: Option<Vec<DynamicContainerReport>>,
}

impl SyncInput {
    /// Start building a sync payload while preserving the stable
    /// [`SyncRequest`] literal surface for downstream callers.
    pub fn builder(request: SyncRequest) -> SyncInputBuilder {
        SyncInputBuilder {
            request,
            operator_image: None,
            application: None,
            dynamic_containers: None,
        }
    }
}

/// Builder for optional sync receipts. New optional wire metadata belongs
/// here so adding it does not break downstream [`SyncRequest`] literals.
pub struct SyncInputBuilder {
    request: SyncRequest,
    operator_image: Option<OperatorImageReport>,
    application: Option<ObservedApplicationReport>,
    dynamic_containers: Option<Vec<DynamicContainerReport>>,
}

impl SyncInputBuilder {
    /// Attach the immutable image receipt for the running Operator.
    pub fn operator_image(mut self, operator_image: OperatorImageReport) -> Self {
        self.operator_image = Some(operator_image);
        self
    }

    /// Attach the application release observed in the environment.
    pub fn application(mut self, application: ObservedApplicationReport) -> Self {
        self.application = Some(application);
        self
    }

    pub fn dynamic_containers(mut self, reports: Vec<DynamicContainerReport>) -> Self {
        self.dynamic_containers = Some(reports);
        self
    }

    /// Finish the serializable sync payload.
    pub fn build(self) -> SyncInput {
        SyncInput {
            request: self.request,
            operator_image: self.operator_image,
            application: self.application,
            dynamic_containers: self.dynamic_containers,
        }
    }
}

/// Response from the manager to the agent sync request.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SyncResponse {
    /// Exact operation attempt associated with `target`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub execution_claim: Option<SyncExecutionClaim>,
    /// Authoritative deployment state from the manager.
    ///
    /// Pull agents use this to hydrate local state when attaching to an
    /// already-imported deployment. Absent means the agent's local state is
    /// already authoritative or no state has been established yet.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub current_state: Option<DeploymentState>,
    /// Target deployment the agent should converge toward.
    /// None means no changes needed or this is an observe-only deployment.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub target: Option<TargetDeployment>,
    /// Public URL for the commands API (e.g. `https://manager.example.com/v1`).
    /// Operators and app-owned receivers use this to lease pending commands;
    /// Workers themselves receive pushes.
    /// When absent, the agent falls back to its sync URL.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub commands_url: Option<String>,
    /// Target operations-bundle set the Operator should converge its loaded
    /// plugin registry toward. None means no enabled plugin set has ever
    /// been established for this project (nothing to sync), not that the
    /// Operator is up to date — compare `hash` against the Operator's own
    /// loaded hash to decide whether to download anything.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target_operations_bundle_set: Option<TargetOperationsBundleSet>,
    /// Complete release-independent target set. None means the manager does
    /// not support this protocol; Some(empty) means remove owned containers.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target_dynamic_containers: Option<Vec<TargetDynamicContainer>>,
    /// Base URL the Operator opens tunnel connections to. None means the
    /// manager does not accept tunnels, so the Operator never dials.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tunnel_url: Option<String>,
    /// Operator image the manager wants this Operator to run. Operators that
    /// manage their own workload update to it; None means no opinion.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target_operator_image: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct SyncExecutionClaim {
    pub operation_id: String,
    pub attempt_id: String,
}

/// Target deployment state for the agent to converge toward.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TargetDeployment {
    /// Release information (ID, version, stack definition).
    pub release_info: ReleaseInfo,
    /// Full deployment configuration (settings, env vars, etc.).
    pub config: DeploymentConfig,
}

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

    #[test]
    fn test_sync_request_serialization() {
        let req = SyncRequest {
            deployment_id: "dep_abc123".to_string(),
            session: "operator-test".to_string(),
            supports_execution_claims: true,
            supports_tunnels: true,
            execution_claim: None,
            current_state: None,
            heartbeats: Vec::new(),
            observed_inventory_batches: Vec::new(),
            capabilities: Vec::new(),
            operator_version: None,
            operations_report: None,
            operations_config: None,
        };

        let json = serde_json::to_value(&req).unwrap();
        assert_eq!(json["deploymentId"], "dep_abc123");
        assert_eq!(json["supportsExecutionClaims"], true);
        // current_state is None → should be omitted
        assert!(json.get("currentState").is_none());
        assert!(json.get("resourceHeartbeats").is_none());
        assert!(json.get("capabilities").is_none());
        assert!(json.get("operatorVersion").is_none());
        assert!(json.get("operatorImage").is_none());
        assert!(json.get("operationsReport").is_none());
    }

    #[test]
    fn test_sync_request_deserialization() {
        let json = r#"{"deploymentId": "dep_xyz"}"#;
        let req: SyncRequest = serde_json::from_str(json).unwrap();
        assert_eq!(req.deployment_id, "dep_xyz");
        assert!(req.session.is_empty());
        assert!(!req.supports_execution_claims);
        assert!(req.current_state.is_none());
        assert!(req.heartbeats.is_empty());
        assert!(req.observed_inventory_batches.is_empty());
        assert!(req.capabilities.is_empty());
        assert!(req.operator_version.is_none());
        assert!(req.operations_report.is_none());
    }

    #[test]
    fn test_sync_response_empty() {
        let resp = SyncResponse {
            execution_claim: None,
            current_state: None,
            target: None,
            commands_url: None,
            target_operations_bundle_set: None,
            target_dynamic_containers: None,
            tunnel_url: None,
            target_operator_image: None,
        };
        let json = serde_json::to_value(&resp).unwrap();
        // target is None → should be omitted
        assert!(json.get("target").is_none());
        assert!(json.get("currentState").is_none());
        assert!(json.get("targetOperationsBundleSet").is_none());
    }

    #[test]
    fn test_sync_response_roundtrip_no_target() {
        let resp = SyncResponse {
            execution_claim: None,
            current_state: None,
            target: None,
            commands_url: None,
            target_operations_bundle_set: None,
            target_dynamic_containers: None,
            tunnel_url: None,
            target_operator_image: None,
        };
        let serialized = serde_json::to_string(&resp).unwrap();
        let deserialized: SyncResponse = serde_json::from_str(&serialized).unwrap();
        assert!(deserialized.target.is_none());
        assert!(deserialized.current_state.is_none());
    }

    #[test]
    fn test_sync_request_with_camel_case() {
        // Verify camelCase renaming works correctly
        let json = r#"{"deploymentId": "dep_1", "currentState": null}"#;
        let req: SyncRequest = serde_json::from_str(json).unwrap();
        assert_eq!(req.deployment_id, "dep_1");
        assert!(req.current_state.is_none());
        assert!(req.heartbeats.is_empty());
        assert!(req.capabilities.is_empty());

        // snake_case should NOT work
        let json = r#"{"deployment_id": "dep_1"}"#;
        assert!(serde_json::from_str::<SyncRequest>(json).is_err());
    }

    #[test]
    fn test_sync_request_heartbeats_roundtrip() {
        let json = serde_json::json!({
            "deploymentId": "dep_1",
            "resourceHeartbeats": [{
                "deploymentId": "dep_1",
                "resourceId": "api",
                "resourceType": "container",
                "controllerPlatform": "kubernetes",
                "backend": "kubernetes",
                "observedAt": "2026-01-01T00:00:00Z",
                "data": {
                    "resourceType": "container",
                    "data": {
                        "backend": "kubernetes",
                        "status": {
                            "health": "healthy",
                            "lifecycle": "running",
                            "message": null,
                            "stale": false,
                            "partial": false,
                            "collectionIssues": []
                        },
                        "namespace": "default",
                        "name": "api",
                        "workloadKind": "deployment",
                        "replicas": { "desired": 1, "current": 1, "ready": 1, "available": 1, "updated": null, "misscheduled": null },
                        "restarts": 0,
                        "cpu": null,
                        "memory": null,
                        "workload": null,
                        "pods": [],
                        "instances": [],
                        "events": []
                    }
                },
                "raw": []
            }]
        });

        let req: SyncRequest = serde_json::from_value(json).unwrap();
        assert_eq!(req.heartbeats.len(), 1);
        assert_eq!(req.heartbeats[0].resource_id, "api");
        assert!(req.capabilities.is_empty());

        let serialized = serde_json::to_value(&req).unwrap();
        assert_eq!(serialized["resourceHeartbeats"][0]["resourceId"], "api");
    }

    #[test]
    fn test_sync_response_observe_only_state_roundtrip() {
        let state = DeploymentState {
            status: crate::DeploymentStatus::Running,
            platform: crate::Platform::Kubernetes,
            current_release: None,
            target_release: None,
            stack_state: None,
            error: None,
            environment_info: None,
            runtime_metadata: None,
            retry_requested: false,
            protocol_version: crate::DEPLOYMENT_PROTOCOL_VERSION,
        };
        assert!(!state.has_desired());

        let resp = SyncResponse {
            execution_claim: None,
            current_state: Some(state),
            target: None,
            commands_url: None,
            target_operations_bundle_set: None,
            target_dynamic_containers: None,
            tunnel_url: None,
            target_operator_image: None,
        };

        let serialized = serde_json::to_string(&resp).unwrap();
        let deserialized: SyncResponse = serde_json::from_str(&serialized).unwrap();
        let current_state = deserialized.current_state.unwrap();

        assert_eq!(current_state.status, crate::DeploymentStatus::Running);
        assert!(!current_state.has_desired());
        assert!(deserialized.target.is_none());
    }

    #[test]
    fn test_sync_request_operations_report_absent_by_default() {
        // Simulates an old Operator that doesn't know about operationsReport:
        // the field must default to None, never be required.
        let json = r#"{"deploymentId": "dep_old_operator", "operatorVersion": "0.9.0"}"#;
        let req: SyncRequest = serde_json::from_str(json).unwrap();
        assert_eq!(req.operator_version.as_deref(), Some("0.9.0"));
        assert!(req.operations_report.is_none());
    }

    #[test]
    fn test_sync_request_operator_image_roundtrip() {
        let digest = format!("sha256:{}", "a".repeat(64));
        let report = OperatorImageReport {
            source: OperatorImageSource::Package,
            package_id: Some("pkg_operator".to_string()),
            package_version: Some("1.2.3".to_string()),
            image: format!("registry.example.com/operator@{digest}"),
            digest,
        };
        report.validate().expect("valid package image report");

        let value = serde_json::to_value(&report).unwrap();
        assert_eq!(value["source"], "package");
        assert_eq!(value["packageId"], "pkg_operator");
        assert_eq!(value["packageVersion"], "1.2.3");

        let decoded: OperatorImageReport = serde_json::from_value(value).unwrap();
        assert_eq!(decoded, report);
    }

    #[test]
    fn sync_input_adds_operator_image_without_expanding_sync_request() {
        let digest = format!("sha256:{}", "a".repeat(64));
        let request = SyncRequest {
            deployment_id: "dep_1".to_string(),
            session: String::new(),
            supports_execution_claims: false,
            supports_tunnels: false,
            execution_claim: None,
            current_state: None,
            heartbeats: Vec::new(),
            observed_inventory_batches: Vec::new(),
            capabilities: Vec::new(),
            operator_version: None,
            operations_report: None,
            operations_config: None,
        };
        let input = SyncInput::builder(request)
            .operator_image(OperatorImageReport {
                source: OperatorImageSource::Configured,
                package_id: None,
                package_version: None,
                image: format!("registry.example.com/operator@{digest}"),
                digest,
            })
            .build();

        let value = serde_json::to_value(input).unwrap();
        assert_eq!(value["deploymentId"], "dep_1");
        assert_eq!(value["operatorImage"]["source"], "configured");
    }

    #[test]
    fn sync_input_omits_absent_operator_image() {
        let request = SyncRequest {
            deployment_id: "dep_1".to_string(),
            session: String::new(),
            supports_execution_claims: false,
            supports_tunnels: false,
            execution_claim: None,
            current_state: None,
            heartbeats: Vec::new(),
            observed_inventory_batches: Vec::new(),
            capabilities: Vec::new(),
            operator_version: None,
            operations_report: None,
            operations_config: None,
        };

        let value = serde_json::to_value(SyncInput::builder(request).build()).unwrap();
        assert!(value.get("operatorImage").is_none());
    }

    #[test]
    fn operator_image_report_rejects_mutable_or_inconsistent_identity() {
        let digest = format!("sha256:{}", "a".repeat(64));
        let mutable = OperatorImageReport {
            source: OperatorImageSource::Configured,
            package_id: None,
            package_version: None,
            image: "registry.example.com/operator:latest".to_string(),
            digest: digest.clone(),
        };
        assert_eq!(
            mutable.validate(),
            Err("operator image must use repository@sha256:digest form")
        );

        let mismatched = OperatorImageReport {
            source: OperatorImageSource::Configured,
            package_id: None,
            package_version: None,
            image: format!("registry.example.com/operator@sha256:{}", "b".repeat(64)),
            digest,
        };
        assert_eq!(
            mismatched.validate(),
            Err("operator image must contain one repository and the reported digest")
        );
    }

    #[test]
    fn operator_image_report_enforces_source_specific_package_fields() {
        let digest = format!("sha256:{}", "a".repeat(64));
        let image = format!("registry.example.com/operator@{digest}");
        let missing_package = OperatorImageReport {
            source: OperatorImageSource::Package,
            package_id: None,
            package_version: None,
            image: image.clone(),
            digest: digest.clone(),
        };
        assert_eq!(
            missing_package.validate(),
            Err("package operator images require package ID and version")
        );

        let configured_with_package = OperatorImageReport {
            source: OperatorImageSource::Configured,
            package_id: Some("pkg_operator".to_string()),
            package_version: Some("1.2.3".to_string()),
            image,
            digest,
        };
        assert_eq!(
            configured_with_package.validate(),
            Err("configured operator images must not include package identity")
        );
    }

    #[test]
    fn test_sync_request_operations_report_roundtrip() {
        let req = SyncRequest {
            deployment_id: "dep_1".to_string(),
            session: String::new(),
            supports_execution_claims: false,
            supports_tunnels: false,
            execution_claim: None,
            current_state: None,
            heartbeats: Vec::new(),
            observed_inventory_batches: Vec::new(),
            capabilities: Vec::new(),
            operator_version: Some("1.2.3".to_string()),
            operations_report: Some(OperationsReport {
                loaded_bundle_hash: Some("builtin:s3@1.0.0:key|".to_string()),
                operations: vec![ReportedOperation {
                    plugin: "s3".to_string(),
                    plugin_version: "1.0.0".to_string(),
                    name: "list-buckets".to_string(),
                }],
            }),
            operations_config: None,
        };

        let json = serde_json::to_value(&req).unwrap();
        assert_eq!(
            json["operationsReport"]["loadedBundleHash"],
            "builtin:s3@1.0.0:key|"
        );
        assert_eq!(json["operationsReport"]["operations"][0]["plugin"], "s3");
        assert_eq!(
            json["operationsReport"]["operations"][0]["pluginVersion"],
            "1.0.0"
        );
        assert_eq!(
            json["operationsReport"]["operations"][0]["name"],
            "list-buckets"
        );

        let deserialized: SyncRequest = serde_json::from_value(json).unwrap();
        assert_eq!(
            deserialized
                .operations_report
                .as_ref()
                .unwrap()
                .loaded_bundle_hash,
            req.operations_report.as_ref().unwrap().loaded_bundle_hash
        );
        assert_eq!(
            deserialized.operations_report.unwrap().operations,
            req.operations_report.unwrap().operations
        );
    }

    #[test]
    fn test_sync_response_target_operations_bundle_set_roundtrip() {
        let resp = SyncResponse {
            execution_claim: None,
            current_state: None,
            target: None,
            commands_url: None,
            target_operations_bundle_set: Some(TargetOperationsBundleSet {
                hash: "builtin:s3@1.0.0:key|".to_string(),
                bundles: vec![OperationsBundleDownload {
                    plugin: "s3".to_string(),
                    plugin_version: "1.0.0".to_string(),
                    url: "https://storage.example.com/bundle.zip?sig=abc".to_string(),
                    env: BTreeMap::new(),
                }],
            }),
            target_dynamic_containers: None,
            tunnel_url: None,
            target_operator_image: None,
        };

        let json = serde_json::to_value(&resp).unwrap();
        assert_eq!(
            json["targetOperationsBundleSet"]["hash"],
            "builtin:s3@1.0.0:key|"
        );
        assert_eq!(
            json["targetOperationsBundleSet"]["bundles"][0]["plugin"],
            "s3"
        );

        let deserialized: SyncResponse = serde_json::from_value(json).unwrap();
        assert_eq!(
            deserialized.target_operations_bundle_set,
            resp.target_operations_bundle_set
        );
    }

    #[test]
    fn test_sync_response_target_operations_bundle_set_absent_by_default() {
        let json = serde_json::json!({});
        let resp: SyncResponse = serde_json::from_value(json).unwrap();
        assert!(resp.target_operations_bundle_set.is_none());
    }

    #[test]
    fn dynamic_container_sync_preserves_empty_target_and_hides_secrets_in_debug() {
        let old_manager_response: SyncResponse = serde_json::from_str("{}").unwrap();
        assert!(old_manager_response.target_dynamic_containers.is_none());

        let empty_target = SyncResponse {
            execution_claim: None,
            current_state: None,
            target: None,
            commands_url: None,
            target_operations_bundle_set: None,
            target_dynamic_containers: Some(vec![]),
            tunnel_url: None,
            target_operator_image: None,
        };
        let json = serde_json::to_value(&empty_target).unwrap();
        assert_eq!(json["targetDynamicContainers"], serde_json::json!([]));

        let target = TargetDynamicContainer {
            name: "api".to_string(),
            generation: 2,
            image: "example.com/api@sha256:abc".to_string(),
            cpu: "0.5".to_string(),
            memory: "512Mi".to_string(),
            replicas: 1,
            ports: vec![8080],
            deleted: false,
            env: BTreeMap::new(),
            secret_env: BTreeMap::from([("TOKEN".to_string(), "private-value".to_string())]),
            health_check: None,
            suspended_reason: None,
        };
        assert!(!format!("{target:?}").contains("private-value"));
        let roundtrip: TargetDynamicContainer =
            serde_json::from_value(serde_json::to_value(&target).unwrap()).unwrap();
        assert_eq!(roundtrip, target);
    }
}