hatchery-core 0.2.5

//! Shared hatchery types and utilities: the single crate node, edge (C2) and
//! core (server) all depend on. It depends on none of them.
//!
//! Contents: inventory/control wire contract, multiplex frames and envelopes,
//! nested restream crypto and key types (`nested`), replay buffer and node
//! journal.

pub use gate4agent_node_protocol::{
    AdapterContractRevision, AdapterFamily, AdapterId, AgentId, ArchitectureId, CapabilityId,
    ClientCompatibilityOffer, HostDescriptor, HostDirectoryEntry, HostDirectoryListing,
    ManagedWorktreeCleanupFailure, ManagedWorktreeGitScope, ManagedWorktreeLeaseId,
    ManagedWorktreeLeaseSnapshot,
    ManagedWorktreeLeaseState, ManagedWorktreeRetention, ManagedWorktreeSpawnReceipt,
    ManagedWorktreeSpawnRequest, ManagedWorktreeSpawnRequestV2, ManagedWorktreeProfileSummary,
    WorktreeProfileInventory,
    WorktreeServiceMode,
    provider_id_is_legacy, NodeCursor, NodeEvent, NodeFailure, NodeId, NodeIncarnationId, NodeRequest,
    NodeResponse, OpaqueHostPath, OperatingSystemId, PathEncoding, PathSemantics, PathStyle,
    ProviderAdapterContractSupport, ProviderContractRevision, ProviderContractSupport,
    ProviderRuntimeContractId, ProviderRuntimeMode, ProviderRuntimeStatus,
    ProviderRuntimeStatuses, ProviderRuntimeVersion,
    LaunchInventory, ResolvedBundleReceipt, ResolvedEnvironmentProfileReceipt,
    ResolvedSpawnReceipt, SpawnProfileSummary,
    ContextPackBytesRead, ContextPackLineageReceipt, ResolvedContextPackReceipt, SpawnContextDigest,
    ResolvedSpawnSpec, SpawnBundleDigest, SpawnBundleId, SpawnBundleRevision,
    SpawnContextId, SpawnDeadlineMs, SpawnEnvironmentProfileId,
    SpawnEnvironmentProfileRevision, SpawnNetworkAllowlistId, SpawnBrowserProfileId,
    SpawnFieldProvenance, SpawnIdempotencyKey, SpawnOverride,
    SpawnOverrides, SpawnProfileDefaults, SpawnProfileId, SpawnProfileRevision, SpawnPrompt,
    SpawnPromptMetadata, SpawnRequiredCapabilities, SpawnResolutionProvenance, SpawnSpec,
    SpawnTarget,
    WorktreeProfileId, WorktreeProfileRevision,
    NODE_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY,
    NODE_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY,
    NODE_CHILD_ENVIRONMENT_PROFILE_CAPABILITY,
    NODE_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY,
    NODE_HISTORY_CONTEXT_PACK_CAPABILITY,
    NODE_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY,
    NODE_NATIVE_SESSION_CATALOG_CAPABILITY,
    NODE_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY,
    NODE_NATIVE_SESSION_INDEX_CAPABILITY,
    NODE_NATIVE_SESSION_PREVIEW_CAPABILITY,
    NODE_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY,
    NODE_WORKSPACE_ENTRY_CREATE_CAPABILITY,
    NODE_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY,
    CAPABILITY_HOST_DIRECTORY_BROWSE_V1,
    NODE_PROVIDER_ID_OPEN_CAPABILITY, NODE_SESSION_TASK_CORRELATION_CAPABILITY,
    NODE_SPAWN_PROFILE_REVISION_CAPABILITY, NODE_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY,
    NODE_TERMINAL_FRAME_EVENTS_CAPABILITY,
    NODE_AGENT_STREAM_EVENTS_CAPABILITY,
    NODE_ACP_CONTROL_CAPABILITY,
    NODE_WORKTREE_SELECTION_CAPABILITY,
    GitDiff, GitHistoryPage, RepositoryPath, WorkspaceFileContent, WorkspaceFileRead,
    HistoryCandidateSummary, NativeSessionCatalogEntry, NativeSessionCatalogPage,
    NativeSessionCatalogRoute, NativeSessionCatalogScope, NativeSessionCatalogSummary,
    NativeSessionCatalogWindow, NativeSessionExternalGroup, NativeSessionExternalGroupKind,
    NativeSessionPreview, NativeSessionSelection, SessionRecordPreview,
    AgentProgressAttentionKindV1, AgentProgressAttentionV1, AgentProgressCurrentV1,
    AgentProgressEventKindV1, AgentProgressUsageV1, AgentProgressV1, SessionAgentProgress,
    SessionTaskBindingV1, SessionTaskTargetV1, TaskId,
    DeliveryBlobChunkHexV1, DeliveryBlobDigestV1, DeliveryBlobReceiptV1,
    DeliveryBundleManifestV2, DeliveryCommitReceiptV1, DeliveryManifestDigestV2,
    DeliveryStageId,
    NODE_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY,
    HarnessMcpActivationDigest, HarnessMcpCallId, HarnessMcpContentTypeV1,
    HarnessMcpOpaquePayloadV1, HarnessMcpRejectReasonV1,
    HarnessMcpReplyChunkHexV1, HarnessMcpReservationId,
    ResolvedHarnessMcpProxyReceiptV1,
    NODE_HARNESS_MCP_READ_PROXY_CAPABILITY,
    ProtocolNegotiationError, ProtocolRange,
};
pub use gate4agent_build_stamp::BUILD_STAMP;
mod control_detail;
pub use control_detail::control_telemetry_detail;

mod multiplex;
pub use multiplex::*;
pub mod inner;
#[cfg(feature = "nested-crypto")]
pub mod nested;
pub mod buffer;
pub mod journal;

/// Nested restream crypto: outer edge layer, inner Noise layer, key types.
#[cfg(feature = "nested-crypto")]
pub mod crypto {
    pub use crate::nested::*;
}
#[cfg(feature = "nested-crypto")]
pub use nested::{PublicKeyBytes, RestreamerKeys};
pub use buffer::{BufferLimits, BufferedEvent, EnqueueOutcome, ReplayBuffer, StreamKey};
pub use journal::{
    recover_gap, FileJournal, JournalEntry, JournalError, LocalFileJournal, MemoryJournal,
    NodeJournal,
};
use gate4agent_node_protocol::{
    AgentStreamChunkV1, ManagedSessionRecord, ManagedSessionState, SessionHistorySummaryV1,
    NodeSnapshot, SessionAddress, SessionMode, SessionRecordId, WorkspaceId,
};
use gate4agent_types::{
    AgentInstanceId, OperationId, PreparedInputKind, ProviderActivity, PtyScreenState,
    SessionGeneration, SessionStatus, TerminalFrame, TerminalSize, TransportKind,
};
use serde::de::{SeqAccess, Visitor};
use serde::{Deserialize, Deserializer, Serialize};
use std::collections::BTreeMap;
use std::fmt;

pub const C2_API_VERSION: u16 = 2;
pub const DEFAULT_C2_API_LISTEN: &str = "127.0.0.1:18320";
pub const C2_COMPATIBILITY_METADATA_CAPABILITY: &str = "compatibility.metadata";
pub const C2_OPAQUE_UNIX_PATH_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_OPAQUE_UNIX_PATH_CAPABILITY;
pub const C2_REPOSITORY_PATH_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_REPOSITORY_PATH_CAPABILITY;
pub const C2_WORKSPACE_FILE_READ_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_WORKSPACE_FILE_READ_CAPABILITY;
pub const C2_WORKSPACE_FILE_WRITE_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_WORKSPACE_FILE_WRITE_CAPABILITY;
pub const C2_WORKSPACE_ENTRY_CREATE_CAPABILITY: &str =
    NODE_WORKSPACE_ENTRY_CREATE_CAPABILITY;
pub const C2_GIT_READ_CAPABILITY: &str = gate4agent_node_protocol::NODE_GIT_READ_CAPABILITY;
pub const C2_PROVIDER_CONTRACT_MANIFEST_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_PROVIDER_CONTRACT_MANIFEST_CAPABILITY;
pub const C2_PROVIDER_RUNTIME_STATUS_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_PROVIDER_RUNTIME_STATUS_CAPABILITY;
pub const C2_PROVIDER_ID_OPEN_CAPABILITY: &str = NODE_PROVIDER_ID_OPEN_CAPABILITY;
pub const C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY: &str =
    NODE_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY;
pub const C2_SPAWN_PROFILE_REVISION_CAPABILITY: &str =
    NODE_SPAWN_PROFILE_REVISION_CAPABILITY;
pub const C2_TERMINAL_FRAME_EVENTS_CAPABILITY: &str = NODE_TERMINAL_FRAME_EVENTS_CAPABILITY;
/// The outbound agent-content stream -- `C2NodeEvent::AgentStream` -- gated
/// the same way `C2_TERMINAL_FRAME_EVENTS_CAPABILITY` gates
/// `C2NodeEvent::TerminalFrame`: a peer that never negotiated it must never
/// receive it.
pub const C2_AGENT_STREAM_EVENTS_CAPABILITY: &str = NODE_AGENT_STREAM_EVENTS_CAPABILITY;
/// The four ACP control verbs -- `ResolveInteraction`, `SetSessionMode`,
/// `SetSessionConfigOption`, `SetSessionModel` -- gated the same way every
/// other capability-scoped path is: a peer that never negotiated it must
/// never have its requests admitted.
pub const C2_ACP_CONTROL_CAPABILITY: &str = NODE_ACP_CONTROL_CAPABILITY;
pub const C2_WORKTREE_SELECTION_CAPABILITY: &str = NODE_WORKTREE_SELECTION_CAPABILITY;
pub const C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY: &str =
    NODE_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY;
pub const C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY: &str =
    NODE_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY;
pub const C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY: &str =
    NODE_CHILD_ENVIRONMENT_PROFILE_CAPABILITY;
pub const C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY: &str =
    NODE_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY;
pub const C2_HISTORY_CONTEXT_PACK_CAPABILITY: &str = NODE_HISTORY_CONTEXT_PACK_CAPABILITY;
pub const C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY: &str =
    NODE_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY;
pub const C2_NATIVE_SESSION_CATALOG_CAPABILITY: &str = NODE_NATIVE_SESSION_CATALOG_CAPABILITY;
pub const C2_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY: &str =
    NODE_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY;
pub const C2_NATIVE_SESSION_PREVIEW_CAPABILITY: &str = NODE_NATIVE_SESSION_PREVIEW_CAPABILITY;
pub const C2_NATIVE_SESSION_INDEX_CAPABILITY: &str = NODE_NATIVE_SESSION_INDEX_CAPABILITY;
pub const C2_HOST_DIRECTORY_BROWSE_CAPABILITY: &str = CAPABILITY_HOST_DIRECTORY_BROWSE_V1;
pub const C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY: &str =
    NODE_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY;
pub const C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY: &str =
    NODE_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY;
pub const C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY: &str =
    gate4agent_node_protocol::NODE_AGENT_PROGRESS_SNAPSHOT_CAPABILITY;
pub const C2_SESSION_TASK_CORRELATION_CAPABILITY: &str =
    NODE_SESSION_TASK_CORRELATION_CAPABILITY;
/// The sanitized control-event detail -- `C2ControlEvent::detail` -- gated the
/// same way `C2_AGENT_STREAM_EVENTS_CAPABILITY` gates the agent stream: a peer
/// that never negotiated it must never receive it. A C2-level capability, not a
/// node one: the node always publishes its full control events and the C2
/// decides, per connection, what of them crosses.
pub const C2_CONTROL_DETAIL_CAPABILITY: &str = "control-detail-v1";
pub const C2_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY: &str =
    NODE_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY;
pub const C2_HARNESS_MCP_READ_PROXY_CAPABILITY: &str =
    NODE_HARNESS_MCP_READ_PROXY_CAPABILITY;
pub const C2_AUTH_NONCE_BYTES: usize = 32;
pub const C2_AUTH_PROOF_BYTES: usize = 32;
pub const MAX_C2_AUTH_COMPATIBILITY_CAPABILITIES: usize = 64;
pub const MAX_C2_BOUND_AUTH_TRANSCRIPT_BYTES: usize = 16 * 1024;
pub const MAX_C2_CLIENT_FRAME_BYTES: usize = 256 * 1024;
pub const MAX_C2_SERVER_FRAME_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_C2_AUTH_FRAME_BYTES: usize = 8 * 1024;
pub const MAX_C2_HELLO_FRAME_BYTES: usize = MAX_C2_SERVER_FRAME_BYTES;
pub const MAX_C2_RESPONSE_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_C2_NODES: usize = 64;
pub const MAX_C2_ENDPOINT_BYTES: usize = 1024;
pub const MAX_C2_WORKSPACES_PER_NODE: usize = 32;
pub const MAX_C2_SESSIONS_PER_NODE: usize = 128;
pub const MAX_C2_MANAGED_SESSIONS_PER_NODE: usize = 128;
pub const MAX_C2_MANAGED_WORKTREES_PER_NODE: usize =
    gate4agent_node_protocol::MAX_MANAGED_WORKTREE_LEASES;
pub const MAX_C2_GAPS_PER_NODE: usize = 64;
pub const MAX_C2_ROOT_BYTES: usize = 1024;
pub const MAX_C2_SESSION_DISPLAY_NAME_BYTES: usize =
    gate4agent_node_protocol::MAX_SESSION_DISPLAY_NAME_BYTES;

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct C2RequestId(pub u64);

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct NodeRoute {
    pub node_id: NodeId,
    pub expected_incarnation_id: NodeIncarnationId,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct RoutedNodeRequest {
    pub route: NodeRoute,
    pub request: NodeRequest,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct RoutedNodeResponse {
    pub node_id: NodeId,
    pub incarnation_id: NodeIncarnationId,
    pub response: Result<C2NodeResponse, C2NodeFailure>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct RoutedNodeEvent {
    pub node_id: NodeId,
    pub cursor: NodeCursor,
    pub event: C2NodeEvent,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2NodeFailure {
    pub code: gate4agent_node_protocol::NodeFailureCode,
    pub message: String,
}

impl From<&NodeFailure> for C2NodeFailure {
    fn from(failure: &NodeFailure) -> Self {
        use gate4agent_node_protocol::NodeFailureCode;
        let message = match failure.code {
            NodeFailureCode::InvalidRequest => "invalid request",
            NodeFailureCode::UnsupportedCapability => "required capability unavailable",
            NodeFailureCode::SpawnProfileRevisionMismatch => {
                "spawn profile revision mismatch"
            }
            NodeFailureCode::HarnessMcpUnavailable => "harness MCP proxy unavailable",
            NodeFailureCode::ReservationNotFound => "harness MCP reservation unavailable",
            NodeFailureCode::ReservationConflict => "harness MCP reservation conflict",
            NodeFailureCode::ReservationExpired => "harness MCP reservation expired",
            NodeFailureCode::BindingMismatch => "harness MCP binding mismatch",
            NodeFailureCode::NotActivated => "harness MCP reservation not activated",
            NodeFailureCode::CallNotFound => "harness MCP call unavailable",
            NodeFailureCode::ChunkOutOfOrder => "harness MCP reply chunk out of order",
            NodeFailureCode::ResponseTooLarge => "harness MCP response too large",
            NodeFailureCode::DeliveryManifestInvalid => "delivery manifest invalid",
            NodeFailureCode::UnknownDeliveryStage => "delivery stage unavailable",
            NodeFailureCode::DeliveryStageConflict => "delivery stage conflict",
            NodeFailureCode::DeliveryBlobUnexpected => "delivery blob unexpected",
            NodeFailureCode::DeliveryChunkOutOfOrder => "delivery chunk out of order",
            NodeFailureCode::DeliveryBlobDigestMismatch => "delivery blob digest mismatch",
            NodeFailureCode::DeliveryBundleDigestMismatch => "delivery bundle digest mismatch",
            NodeFailureCode::DeliveryStageIncomplete => "delivery stage incomplete",
            NodeFailureCode::DeliveryStageStorageFailed => "delivery stage storage failed",
            NodeFailureCode::Unauthorized => "authentication rejected",
            NodeFailureCode::ObserverReadOnly => "operator access required",
            NodeFailureCode::ControllerBusy => "controller busy",
            NodeFailureCode::ControllerRequired => "controller required",
            NodeFailureCode::UnknownWorkspace => "workspace unavailable",
            NodeFailureCode::HostDirectoryInvalid => "host directory invalid",
            NodeFailureCode::HostDirectoryReadFailed => "host directory read failed",
            NodeFailureCode::HostDirectoryReadTimedOut => "host directory read timed out",
            NodeFailureCode::StandaloneWorkspaceRecoveryRequired => {
                "standalone workspace recovery required"
            }
            NodeFailureCode::InvalidRepositoryPath => "repository path invalid",
            NodeFailureCode::RepositoryFileNotFound => "repository file unavailable",
            NodeFailureCode::RepositoryFileNotRegular => "repository path is not a regular file",
            NodeFailureCode::RepositoryPathUnsafe => "repository path is unsafe",
            NodeFailureCode::RepositoryFileReadTimedOut => "repository file read timed out",
            NodeFailureCode::RepositoryFileReadFailed => "repository file read failed",
            NodeFailureCode::RepositoryFileWriteTimedOut => "repository file write timed out",
            NodeFailureCode::RepositoryFileWriteFailed => "repository file write failed",
            NodeFailureCode::RepositoryFileRevisionConflict => "repository file changed since it was opened",
            NodeFailureCode::RepositoryEntryAlreadyExists => "repository entry already exists",
            NodeFailureCode::RepositoryParentNotFound => "repository parent directory unavailable",
            NodeFailureCode::RepositoryParentNotDirectory => "repository parent path is not a directory",
            NodeFailureCode::RepositoryEntryCreateTimedOut => "repository entry creation timed out",
            NodeFailureCode::RepositoryEntryCreateFailed => "repository entry creation failed",
            NodeFailureCode::GitReadTimedOut => "git read timed out",
            NodeFailureCode::GitReadFailed => "git read failed",
            NodeFailureCode::InvalidWorkspaceRoot => "workspace root invalid",
            NodeFailureCode::DuplicateWorkspaceId => "workspace ID already registered",
            NodeFailureCode::DuplicateWorkspaceRoot => "workspace root already registered",
            NodeFailureCode::WorkspaceBusy => "workspace busy",
            NodeFailureCode::LastWorkspace => "last workspace protected",
            NodeFailureCode::NotGitRepository => "workspace is not a git repository",
            NodeFailureCode::WorktreeConflict => "worktree conflict",
            NodeFailureCode::WorktreeProtected => "worktree protected",
            NodeFailureCode::WorktreeDirty => "worktree dirty",
            NodeFailureCode::WorktreeLocked => "worktree locked",
            NodeFailureCode::UnknownManagedWorktreeLease => "managed worktree unavailable",
            NodeFailureCode::ManagedWorktreeBusy => "managed worktree busy",
            NodeFailureCode::ManagedWorktreeOwnershipConflict => {
                "managed worktree ownership conflict"
            }
            NodeFailureCode::ManagedWorktreeProfileRevisionMismatch => {
                "managed worktree profile revision mismatch"
            }
            NodeFailureCode::ManagedWorktreeRecoveryRequired => {
                "managed worktree recovery required"
            }
            NodeFailureCode::SpawnTargetMismatch => "spawn target mismatch",
            NodeFailureCode::UnknownSpawnProfile => "spawn profile unavailable",
            NodeFailureCode::UnknownBundle => "session bundle unavailable",
            NodeFailureCode::UnknownEnvironmentProfile => {
                "environment profile unavailable"
            }
            NodeFailureCode::UnknownNetworkAllowlist => {
                "network allowlist unavailable"
            }
            NodeFailureCode::UnsupportedNetworkAllowlistMapping => {
                "network allowlist mapping unsupported for this provider"
            }
            NodeFailureCode::BrowserStationProbeUnavailable => {
                "dig2browser station probe unavailable on this platform"
            }
            NodeFailureCode::BrowserStationUnreachable => {
                "dig2browser station unreachable"
            }
            NodeFailureCode::BrowserStationProfileBusy => {
                "dig2browser station profile busy"
            }
            NodeFailureCode::BundleBindingMismatch => {
                "session bundle binding mismatch"
            }
            NodeFailureCode::EnvironmentProfileBindingMismatch => {
                "environment profile binding mismatch"
            }
            NodeFailureCode::BundleMaterializationFailed => {
                "session bundle materialization failed"
            }
            NodeFailureCode::SpawnIdempotencyConflict => "spawn idempotency conflict",
            NodeFailureCode::SpawnIdempotencyCapacity => "spawn idempotency capacity exhausted",
            NodeFailureCode::SpawnDeadlineExceeded => "spawn deadline exceeded",
            NodeFailureCode::UnsupportedSpawnCapability => "spawn capability unavailable",
            NodeFailureCode::UnsupportedTransport => "provider does not support the requested transport",
            NodeFailureCode::TurnInFlight => "session already has a turn in flight",
            NodeFailureCode::UnknownSession => "session unavailable",
            NodeFailureCode::UnknownSessionRecord => "managed session unavailable",
            NodeFailureCode::SessionRecordNotResumable => "managed session cannot resume",
            NodeFailureCode::SessionRecordBusy => "managed session busy",
            NodeFailureCode::SessionRecordConflict => "managed session conflict",
            NodeFailureCode::SessionWorkspaceMismatch => "session workspace mismatch",
            NodeFailureCode::WorkspaceRegistrationRequired => {
                "workspace registration required"
            }
            NodeFailureCode::StaleNativeSessionCatalog => "native session catalog is stale",
            NodeFailureCode::UnknownContextPack => "context pack unavailable",
            NodeFailureCode::ContextPackBusy => "context pack busy",
            NodeFailureCode::ContextPackMaterializationFailed => {
                "context pack materialization failed"
            }
            NodeFailureCode::StaleGeneration => "stale session generation",
            NodeFailureCode::BackendBusy => "node backend busy",
            NodeFailureCode::BackendDisconnected => "node backend disconnected",
            NodeFailureCode::BackendOperationFailed => "node backend operation failed",
            NodeFailureCode::ShuttingDown => "node shutting down",
        };
        Self { code: failure.code, message: message.to_owned() }
    }
}

impl C2NodeFailure {
    pub fn requires_harness_mcp_proxy_capability(&self) -> bool {
        use gate4agent_node_protocol::NodeFailureCode;
        matches!(self.code,
            NodeFailureCode::HarnessMcpUnavailable
                | NodeFailureCode::ReservationNotFound
                | NodeFailureCode::ReservationConflict
                | NodeFailureCode::ReservationExpired
                | NodeFailureCode::BindingMismatch
                | NodeFailureCode::NotActivated
                | NodeFailureCode::CallNotFound
                | NodeFailureCode::ChunkOutOfOrder
                | NodeFailureCode::ResponseTooLarge)
    }

    pub fn requires_host_directory_browse_capability(&self) -> bool {
        use gate4agent_node_protocol::NodeFailureCode;
        matches!(
            self.code,
            NodeFailureCode::HostDirectoryInvalid
                | NodeFailureCode::HostDirectoryReadFailed
                | NodeFailureCode::HostDirectoryReadTimedOut
        )
    }

    pub fn requires_history_context_pack_capability(&self) -> bool {
        use gate4agent_node_protocol::NodeFailureCode;
        matches!(
            self.code,
            NodeFailureCode::UnknownContextPack
                | NodeFailureCode::ContextPackBusy
                | NodeFailureCode::ContextPackMaterializationFailed
        )
    }

    pub fn requires_native_session_catalog_paging_capability(&self) -> bool {
        self.code == gate4agent_node_protocol::NodeFailureCode::StaleNativeSessionCatalog
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct C2ManagedSessionRecord {
    pub record_id: SessionRecordId,
    pub display_name: String,
    pub provider: AgentId,
    pub mode: SessionMode,
    pub state: ManagedSessionState,
    pub workspace_id: WorkspaceId,
    pub active_session: Option<SessionAddress>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bundle: Option<ResolvedBundleReceipt>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub context_id: Option<SpawnContextId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub context: Option<ResolvedContextPackReceipt>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exported_context: Option<ResolvedContextPackReceipt>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub task_binding: Option<SessionTaskBindingV1>,
    pub provider_identity_present: bool,
    pub created_at_unix_ms: u64,
    pub updated_at_unix_ms: u64,
}

impl C2ManagedSessionRecord {
    pub fn context_binding_is_valid(&self) -> bool {
        match (self.context_id.as_ref(), self.context.as_ref()) {
            (None, None) => true,
            (Some(context_id), Some(context)) => {
                &context.id == context_id && context.is_valid()
            }
            (None, Some(_)) | (Some(_), None) => false,
        }
    }

    pub fn requires_history_context_pack_capability(&self) -> bool {
        self.context_id.is_some() || self.context.is_some()
    }

    pub fn exported_context_is_valid(&self) -> bool {
        self.exported_context.as_ref().map_or(true, |pack| {
            pack.is_valid() && pack.lineage.source_provider == self.provider
        })
    }

    pub fn requires_session_task_correlation_capability(&self) -> bool {
        self.task_binding.is_some()
    }

    pub fn task_binding_is_valid(&self) -> bool {
        self.task_binding.as_ref().map_or(true, |binding| {
            binding.revision > 0
                && binding.changed_at_unix_ms >= self.created_at_unix_ms
                && binding.changed_at_unix_ms <= self.updated_at_unix_ms
        })
    }
}

impl<'de> Deserialize<'de> for C2ManagedSessionRecord {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        #[derive(Deserialize)]
        struct WireRecord {
            record_id: SessionRecordId,
            display_name: String,
            provider: AgentId,
            mode: SessionMode,
            state: ManagedSessionState,
            workspace_id: WorkspaceId,
            active_session: Option<SessionAddress>,
            #[serde(default)]
            environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
            #[serde(default)]
            bundle: Option<ResolvedBundleReceipt>,
            #[serde(default)]
            context_id: Option<SpawnContextId>,
            #[serde(default)]
            context: Option<ResolvedContextPackReceipt>,
            #[serde(default)]
            exported_context: Option<ResolvedContextPackReceipt>,
            #[serde(default)]
            task_binding: Option<SessionTaskBindingV1>,
            provider_identity_present: bool,
            created_at_unix_ms: u64,
            updated_at_unix_ms: u64,
        }

        let wire = WireRecord::deserialize(deserializer)?;
        let record = Self {
            record_id: wire.record_id,
            display_name: wire.display_name,
            provider: wire.provider,
            mode: wire.mode,
            state: wire.state,
            workspace_id: wire.workspace_id,
            active_session: wire.active_session,
            environment_profile: wire.environment_profile,
            bundle: wire.bundle,
            context_id: wire.context_id,
            context: wire.context,
            exported_context: wire.exported_context,
            task_binding: wire.task_binding,
            provider_identity_present: wire.provider_identity_present,
            created_at_unix_ms: wire.created_at_unix_ms,
            updated_at_unix_ms: wire.updated_at_unix_ms,
        };
        if !record.context_binding_is_valid() {
            return Err(serde::de::Error::custom(
                "C2 managed session context id and materialization receipt are not correlated",
            ));
        }
        if !record.exported_context_is_valid() {
            return Err(serde::de::Error::custom(
                "C2 managed session exported context receipt is invalid or from a different provider",
            ));
        }
        if !record.task_binding_is_valid() {
            return Err(serde::de::Error::custom(
                "C2 managed session task binding revision or timestamp is invalid",
            ));
        }
        Ok(record)
    }
}

impl From<&ManagedSessionRecord> for C2ManagedSessionRecord {
    fn from(record: &ManagedSessionRecord) -> Self {
        Self {
            record_id: record.record_id.clone(),
            display_name: record.display_name.clone(),
            provider: record.provider.clone(),
            mode: record.mode,
            state: record.state,
            workspace_id: record.workspace_id.clone(),
            active_session: record.active_session.clone(),
            environment_profile: record.environment_profile.clone(),
            bundle: record.bundle.clone(),
            context_id: record.context_id.clone(),
            context: record.context.clone(),
            exported_context: record.exported_context.clone(),
            task_binding: record.task_binding.clone().filter(|binding| {
                binding.revision > 0
                    && binding.changed_at_unix_ms >= record.created_at_unix_ms
                    && binding.changed_at_unix_ms <= record.updated_at_unix_ms
            }),
            provider_identity_present: record.provider_session.is_some(),
            created_at_unix_ms: record.created_at_unix_ms,
            updated_at_unix_ms: record.updated_at_unix_ms,
        }
    }

}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum C2SessionStatus {
    Registered,
    Starting,
    Running,
    Stopping,
    Exited { exit_code: Option<i32> },
    Failed,
}

impl From<&SessionStatus> for C2SessionStatus {
    fn from(status: &SessionStatus) -> Self {
        match status {
            SessionStatus::Registered => Self::Registered,
            SessionStatus::Starting => Self::Starting,
            SessionStatus::Running => Self::Running,
            SessionStatus::Stopping => Self::Stopping,
            SessionStatus::Exited { exit_code } => Self::Exited { exit_code: *exit_code },
            SessionStatus::Failed { .. } => Self::Failed,
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2SessionSnapshot {
    pub instance_id: AgentInstanceId,
    pub agent_id: AgentId,
    pub transport: TransportKind,
    pub generation: SessionGeneration,
    pub status: C2SessionStatus,
    pub pending_operation: Option<OperationId>,
    pub pending_input: Option<PreparedInputKind>,
    pub process_id: Option<u32>,
    pub terminal_size: Option<TerminalSize>,
    pub terminal_frame: Option<TerminalFrame>,
    pub provider_activity: ProviderActivity,
    pub provider_interaction_pending: bool,
    pub provider_identity_present: bool,
    /// The session's CURRENT screen classification, projected from
    /// `gate4agent_types::SessionSnapshot::screen_state` the same way
    /// `provider_activity`/`provider_interaction_pending` are: as its own
    /// field rather than something a reader reconstructs from
    /// `terminal_frame`, which only carries the classification stamped at
    /// one past frame, not the session's current one. `#[serde(default)]`
    /// so a peer that predates this field decodes it as `Unknown`.
    #[serde(default)]
    pub screen_state: PtyScreenState,
}

impl From<&gate4agent_types::SessionSnapshot> for C2SessionSnapshot {
    fn from(session: &gate4agent_types::SessionSnapshot) -> Self {
        Self {
            instance_id: session.instance_id,
            agent_id: session.agent_id.clone(),
            transport: session.transport,
            generation: session.generation,
            status: C2SessionStatus::from(&session.status),
            pending_operation: session.pending_operation,
            pending_input: session.pending_input,
            process_id: session.process_id,
            terminal_size: session.terminal_size,
            terminal_frame: session.terminal_frame.clone(),
            provider_activity: session.provider.activity,
            provider_interaction_pending: !session.provider.interactions.is_empty(),
            provider_identity_present: session.provider.session.is_some(),
            screen_state: session.screen_state.clone(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2WorkspaceSnapshot {
    pub workspace_id: WorkspaceId,
    pub canonical_root: OpaqueHostPath,
    pub sessions: Vec<C2SessionSnapshot>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub worktree_service_mode: Option<WorktreeServiceMode>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub managed_worktree_profiles: Option<WorktreeProfileInventory>,
}

impl From<&gate4agent_node_protocol::WorkspaceSnapshot> for C2WorkspaceSnapshot {
    fn from(workspace: &gate4agent_node_protocol::WorkspaceSnapshot) -> Self {
        Self {
            workspace_id: workspace.workspace_id.clone(),
            canonical_root: workspace.canonical_root.clone(),
            sessions: workspace.sessions.iter().map(C2SessionSnapshot::from).collect(),
            worktree_service_mode: workspace.worktree_service_mode,
            managed_worktree_profiles: workspace.managed_worktree_profiles.clone(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2NodeSnapshot {
    pub node_id: NodeId,
    pub enabled_providers: Vec<AgentId>,
    #[serde(default, skip_serializing_if = "ProviderRuntimeStatuses::is_empty")]
    pub provider_runtime_statuses: ProviderRuntimeStatuses,
    pub workspaces: Vec<C2WorkspaceSnapshot>,
    pub session_records: Vec<C2ManagedSessionRecord>,
    #[serde(
        default,
        skip_serializing_if = "Vec::is_empty",
        deserialize_with = "deserialize_c2_agent_progress_entries"
    )]
    pub agent_progress: Vec<SessionAgentProgress>,
    #[serde(
        default,
        skip_serializing_if = "Vec::is_empty",
        deserialize_with = "deserialize_c2_managed_worktree_leases"
    )]
    pub managed_worktrees: Vec<ManagedWorktreeLeaseSnapshot>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub launch_inventory: Option<LaunchInventory>,
}

impl C2NodeSnapshot {
    pub fn requires_child_environment_profile_capability(&self) -> bool {
        self.session_records
            .iter()
            .any(|record| record.environment_profile.is_some())
            || self.launch_inventory.as_ref().is_some_and(|inventory| {
                inventory.spawn_profiles.as_ref().is_some_and(|profiles| {
                    profiles
                        .iter()
                        .any(|profile| profile.environment_profile.is_some())
                })
            })
    }

    pub fn requires_session_bundle_materialization_capability(&self) -> bool {
        self.session_records
            .iter()
            .any(|record| record.bundle.is_some())
    }

    pub fn requires_history_context_pack_capability(&self) -> bool {
        self.session_records
            .iter()
            .any(C2ManagedSessionRecord::requires_history_context_pack_capability)
    }

    pub fn requires_session_task_correlation_capability(&self) -> bool {
        self.session_records
            .iter()
            .any(C2ManagedSessionRecord::requires_session_task_correlation_capability)
    }
}

fn deserialize_c2_managed_worktree_leases<'de, D>(
    deserializer: D,
) -> Result<Vec<ManagedWorktreeLeaseSnapshot>, D::Error>
where
    D: Deserializer<'de>,
{
    struct ManagedWorktreesVisitor;

    impl<'de> Visitor<'de> for ManagedWorktreesVisitor {
        type Value = Vec<ManagedWorktreeLeaseSnapshot>;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(
                formatter,
                "at most {MAX_C2_MANAGED_WORKTREES_PER_NODE} managed worktrees",
            )
        }

        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
        where
            A: SeqAccess<'de>,
        {
            let mut leases = Vec::with_capacity(
                sequence
                    .size_hint()
                    .unwrap_or(0)
                    .min(MAX_C2_MANAGED_WORKTREES_PER_NODE),
            );
            while let Some(lease) = sequence.next_element::<ManagedWorktreeLeaseSnapshot>()? {
                if leases.len() == MAX_C2_MANAGED_WORKTREES_PER_NODE {
                    return Err(serde::de::Error::invalid_length(leases.len() + 1, &self));
                }
                if leases.iter().any(|existing: &ManagedWorktreeLeaseSnapshot| {
                    existing.lease_id == lease.lease_id
                        || existing.workspace_id == lease.workspace_id
                }) {
                    return Err(serde::de::Error::custom(
                        "C2 managed worktree snapshot contains duplicate identity",
                    ));
                }
                leases.push(lease);
            }
            Ok(leases)
        }
    }

    deserializer.deserialize_seq(ManagedWorktreesVisitor)
}

fn deserialize_c2_agent_progress_entries<'de, D>(
    deserializer: D,
) -> Result<Vec<SessionAgentProgress>, D::Error>
where
    D: Deserializer<'de>,
{
    struct AgentProgressEntriesVisitor;

    impl<'de> Visitor<'de> for AgentProgressEntriesVisitor {
        type Value = Vec<SessionAgentProgress>;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(
                formatter,
                "at most {} bounded agent progress entries",
                gate4agent_node_protocol::MAX_AGENT_PROGRESS_ENTRIES,
            )
        }

        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
        where
            A: SeqAccess<'de>,
        {
            let mut entries = Vec::with_capacity(
                sequence
                    .size_hint()
                    .unwrap_or(0)
                    .min(gate4agent_node_protocol::MAX_AGENT_PROGRESS_ENTRIES),
            );
            while let Some(entry) = sequence.next_element::<SessionAgentProgress>()? {
                if entries.len() == gate4agent_node_protocol::MAX_AGENT_PROGRESS_ENTRIES {
                    return Err(serde::de::Error::invalid_length(entries.len() + 1, &self));
                }
                if entries.iter().any(|existing: &SessionAgentProgress| {
                    existing.address == entry.address
                }) {
                    continue;
                }
                entries.push(entry);
            }
            Ok(entries)
        }
    }

    deserializer.deserialize_seq(AgentProgressEntriesVisitor)
}

impl From<&NodeSnapshot> for C2NodeSnapshot {
    fn from(snapshot: &NodeSnapshot) -> Self {
        Self {
            node_id: snapshot.node_id.clone(),
            enabled_providers: snapshot.enabled_providers.clone(),
            provider_runtime_statuses: snapshot.provider_runtime_statuses.clone(),
            workspaces: snapshot.workspaces.iter().map(C2WorkspaceSnapshot::from).collect(),
            session_records: snapshot.session_records.iter()
                .map(C2ManagedSessionRecord::from)
                .collect(),
            agent_progress: snapshot.agent_progress.clone(),
            managed_worktrees: snapshot.managed_worktrees.clone(),
            launch_inventory: snapshot.launch_inventory.clone(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2NodeEventEnvelope {
    pub sequence: u64,
    pub event: C2NodeEvent,
}

impl From<&gate4agent_node_protocol::NodeEventEnvelope> for C2NodeEventEnvelope {
    fn from(envelope: &gate4agent_node_protocol::NodeEventEnvelope) -> Self {
        Self { sequence: envelope.sequence, event: C2NodeEvent::from(&envelope.event) }
    }
}

impl C2NodeEventEnvelope {
    /// Like `From`, but a `NodeEvent::Control` keeps its sanitized telemetry
    /// detail. A C2 relay uses this and strips the detail again for every
    /// connection that did not negotiate `C2_CONTROL_DETAIL_CAPABILITY`.
    pub fn from_node_envelope_with_control_detail(
        envelope: &gate4agent_node_protocol::NodeEventEnvelope,
    ) -> Self {
        Self {
            sequence: envelope.sequence,
            event: C2NodeEvent::from_node_event_with_control_detail(&envelope.event),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2ControlEvent {
    pub sequence: u64,
    pub command_id: Option<gate4agent_types::CommandId>,
    pub instance_id: AgentInstanceId,
    pub generation: SessionGeneration,
    pub event: C2ControlEventKind,
    /// The sanitized facts behind the tag, for a client that derives its own
    /// telemetry (tool calls, token usage, plans, lifecycle) and negotiated
    /// `C2_CONTROL_DETAIL_CAPABILITY`. `None` on every default projection; see
    /// [`control_telemetry_detail`] for exactly what it keeps and clears.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub detail: Option<gate4agent_types::ControlEvent>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum C2ProviderEventKind {
    SessionStarted,
    SessionIdentityObserved,
    TurnStarted,
    WorkingObserved,
    Text,
    Thinking,
    ToolStarted,
    ToolCompleted,
    TurnCompleted,
    ContextWindowUsage,
    TurnInterrupted,
    SessionEnded,
    Error,
    Ready,
    InteractionRequested,
    InteractionResolved,
    SubagentStarted,
    SubagentStopped,
    RateLimited,
    HostRequestObserved,
    UnrecognizedNotification,
}

impl From<&gate4agent_types::ProviderEvent> for C2ProviderEventKind {
    fn from(event: &gate4agent_types::ProviderEvent) -> Self {
        use gate4agent_types::ProviderEvent;
        match event {
            ProviderEvent::SessionStarted { .. } => Self::SessionStarted,
            ProviderEvent::SessionIdentityObserved { .. } => Self::SessionIdentityObserved,
            ProviderEvent::TurnStarted { .. } => Self::TurnStarted,
            ProviderEvent::WorkingObserved => Self::WorkingObserved,
            ProviderEvent::Text { .. } => Self::Text,
            ProviderEvent::Thinking { .. } => Self::Thinking,
            ProviderEvent::ToolStarted { .. } => Self::ToolStarted,
            ProviderEvent::ToolCompleted { .. } => Self::ToolCompleted,
            ProviderEvent::TurnCompleted { .. } => Self::TurnCompleted,
            ProviderEvent::ContextWindowUsage { .. } => Self::ContextWindowUsage,
            ProviderEvent::TurnInterrupted => Self::TurnInterrupted,
            ProviderEvent::SessionEnded { .. } => Self::SessionEnded,
            ProviderEvent::Error { .. } => Self::Error,
            ProviderEvent::Ready => Self::Ready,
            ProviderEvent::InteractionRequested { .. } => Self::InteractionRequested,
            ProviderEvent::InteractionResolved { .. } => Self::InteractionResolved,
            ProviderEvent::SubagentStarted { .. } => Self::SubagentStarted,
            ProviderEvent::SubagentStopped { .. } => Self::SubagentStopped,
            ProviderEvent::RateLimited { .. } => Self::RateLimited,
            ProviderEvent::HostRequestObserved { .. } => Self::HostRequestObserved,
            ProviderEvent::UnrecognizedNotification { .. } => Self::UnrecognizedNotification,
            // ACP session/update coverage beyond text/tool/turn streaming
            // (`plan`, `available_commands_update`, `current_mode_update`,
            // `session_info_update`, `usage_update`, `config_option_
            // update`, `user_message_chunk`). `C2ProviderEventKind` is a
            // wire enum without a `#[serde(other)]` fallback; minting new
            // variants for it is a deliberate wire-contract decision for
            // whoever owns this protocol, not a side effect of parsing
            // more of ACP's own wire -- until that decision is made, these
            // fold into the same bucket as `UnrecognizedNotification`.
            ProviderEvent::UserMessage { .. }
            | ProviderEvent::Plan { .. }
            | ProviderEvent::AvailableCommandsUpdated { .. }
            | ProviderEvent::ModeChanged { .. }
            | ProviderEvent::SessionInfoUpdated { .. }
            | ProviderEvent::UsageUpdated { .. }
            | ProviderEvent::ConfigOptionsUpdated { .. } => Self::UnrecognizedNotification,
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum C2ControlEventKind {
    CommandRejected,
    Registered,
    StartRequested,
    Running,
    StopRequested,
    InputRequested,
    InputCompleted,
    InputFailed,
    ResizeRequested,
    Resized,
    ResizeFailed,
    ForegroundRefreshRequested,
    ForegroundObserved,
    ForegroundFailed,
    CapabilityProbeRequested,
    CapabilitiesProbed,
    CapabilityProbeFailed,
    HistoryRequested,
    HistoryDiscovered,
    HistoryLoaded,
    HistoryFailed,
    ResumeRequested,
    ResumeAuthorized,
    Resumed,
    ResumeDenied,
    ResumeFailed,
    TerminalStale,
    ProviderEvent { event: C2ProviderEventKind },
    ProviderGap,
    InteractionRequested,
    InteractionResolutionRequested,
    InteractionResolutionFailed,
    InteractionResolved,
    // The nine session-control tags below are the same shape as the
    // interaction trio above: this projection carries WHICH kind of thing
    // happened, never the mode/option/model id it happened to. An id is
    // the session's own vocabulary and has no business on a relay wire.
    SessionModeSetRequested,
    SessionModeSet,
    SessionModeSetFailed,
    SessionConfigOptionSetRequested,
    SessionConfigOptionSet,
    SessionConfigOptionSetFailed,
    SessionModelSetRequested,
    SessionModelSet,
    SessionModelSetFailed,
    Exited { exit_code: Option<i32>, forced: bool },
    Failed,
    Removed,
    ObservationIgnored,
}

impl From<&gate4agent_types::ControlEvent> for C2ControlEvent {
    fn from(event: &gate4agent_types::ControlEvent) -> Self {
        use gate4agent_types::ControlEventKind;
        let projected = match &event.event {
            ControlEventKind::CommandRejected { .. } => C2ControlEventKind::CommandRejected,
            ControlEventKind::Registered => C2ControlEventKind::Registered,
            ControlEventKind::StartRequested { .. } => C2ControlEventKind::StartRequested,
            ControlEventKind::Running { .. } => C2ControlEventKind::Running,
            ControlEventKind::StopRequested { .. } => C2ControlEventKind::StopRequested,
            ControlEventKind::InputRequested { .. } => C2ControlEventKind::InputRequested,
            ControlEventKind::InputCompleted { .. } => C2ControlEventKind::InputCompleted,
            ControlEventKind::InputFailed { .. } => C2ControlEventKind::InputFailed,
            ControlEventKind::ResizeRequested { .. } => C2ControlEventKind::ResizeRequested,
            ControlEventKind::Resized { .. } => C2ControlEventKind::Resized,
            ControlEventKind::ResizeFailed { .. } => C2ControlEventKind::ResizeFailed,
            ControlEventKind::ForegroundRefreshRequested { .. } => C2ControlEventKind::ForegroundRefreshRequested,
            ControlEventKind::ForegroundObserved { .. } => C2ControlEventKind::ForegroundObserved,
            ControlEventKind::ForegroundFailed { .. } => C2ControlEventKind::ForegroundFailed,
            ControlEventKind::CapabilityProbeRequested { .. } => C2ControlEventKind::CapabilityProbeRequested,
            ControlEventKind::CapabilitiesProbed { .. } => C2ControlEventKind::CapabilitiesProbed,
            ControlEventKind::CapabilityProbeFailed { .. } => C2ControlEventKind::CapabilityProbeFailed,
            ControlEventKind::HistoryRequested { .. } => C2ControlEventKind::HistoryRequested,
            ControlEventKind::HistoryDiscovered { .. } => C2ControlEventKind::HistoryDiscovered,
            ControlEventKind::HistoryLoaded { .. } => C2ControlEventKind::HistoryLoaded,
            ControlEventKind::HistoryFailed { .. } => C2ControlEventKind::HistoryFailed,
            ControlEventKind::ResumeRequested { .. } => C2ControlEventKind::ResumeRequested,
            ControlEventKind::ResumeAuthorized { .. } => C2ControlEventKind::ResumeAuthorized,
            ControlEventKind::Resumed { .. } => C2ControlEventKind::Resumed,
            ControlEventKind::ResumeDenied { .. } => C2ControlEventKind::ResumeDenied,
            ControlEventKind::ResumeFailed { .. } => C2ControlEventKind::ResumeFailed,
            ControlEventKind::TerminalStale { .. } => C2ControlEventKind::TerminalStale,
            ControlEventKind::ProviderEvent { event, .. } => C2ControlEventKind::ProviderEvent {
                event: C2ProviderEventKind::from(event),
            },
            ControlEventKind::ProviderGap { .. } => C2ControlEventKind::ProviderGap,
            ControlEventKind::InteractionRequested { .. } => C2ControlEventKind::InteractionRequested,
            ControlEventKind::InteractionResolutionRequested { .. } => C2ControlEventKind::InteractionResolutionRequested,
            ControlEventKind::InteractionResolutionFailed { .. } => C2ControlEventKind::InteractionResolutionFailed,
            ControlEventKind::InteractionResolved { .. } => C2ControlEventKind::InteractionResolved,
            ControlEventKind::SessionModeSetRequested { .. } => C2ControlEventKind::SessionModeSetRequested,
            ControlEventKind::SessionModeSet { .. } => C2ControlEventKind::SessionModeSet,
            ControlEventKind::SessionModeSetFailed { .. } => C2ControlEventKind::SessionModeSetFailed,
            ControlEventKind::SessionConfigOptionSetRequested { .. } => C2ControlEventKind::SessionConfigOptionSetRequested,
            ControlEventKind::SessionConfigOptionSet { .. } => C2ControlEventKind::SessionConfigOptionSet,
            ControlEventKind::SessionConfigOptionSetFailed { .. } => C2ControlEventKind::SessionConfigOptionSetFailed,
            ControlEventKind::SessionModelSetRequested { .. } => C2ControlEventKind::SessionModelSetRequested,
            ControlEventKind::SessionModelSet { .. } => C2ControlEventKind::SessionModelSet,
            ControlEventKind::SessionModelSetFailed { .. } => C2ControlEventKind::SessionModelSetFailed,
            ControlEventKind::Exited { exit_code, forced } => C2ControlEventKind::Exited {
                exit_code: *exit_code,
                forced: *forced,
            },
            ControlEventKind::Failed { .. } => C2ControlEventKind::Failed,
            ControlEventKind::Removed => C2ControlEventKind::Removed,
            ControlEventKind::ObservationIgnored { .. } => C2ControlEventKind::ObservationIgnored,
        };
        Self {
            sequence: event.sequence,
            command_id: event.command_id,
            instance_id: event.instance_id,
            generation: event.generation,
            event: projected,
            detail: None,
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum C2NodeEvent {
    HarnessMcpReadCall {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
        record_id: SessionRecordId,
        session: SessionAddress,
        call_id: HarnessMcpCallId,
        request: HarnessMcpOpaquePayloadV1,
        deadline_unix_ms: u64,
    },
    Control {
        address: SessionAddress,
        event: C2ControlEvent,
    },
    SessionRecordHistorySummarized {
        record_id: SessionRecordId,
        summary: SessionHistorySummaryV1,
    },
    TerminalFrame {
        address: SessionAddress,
        frame: TerminalFrame,
    },
    AgentStream {
        address: SessionAddress,
        chunk: AgentStreamChunkV1,
    },
    ControllerChanged {
        controller: Option<gate4agent_node_protocol::ControllerState>,
    },
    WorkspaceAdded {
        workspace: C2WorkspaceSnapshot,
    },
    WorkspaceRemoved { workspace_id: WorkspaceId },
    SessionRecordUpserted { record: C2ManagedSessionRecord },
    SessionRecordRemoved { record_id: SessionRecordId },
    ManagedWorktreeUpserted { lease: ManagedWorktreeLeaseSnapshot },
    ManagedWorktreeRemoved { lease_id: ManagedWorktreeLeaseId },
    ResyncRequired { oldest_available_sequence: u64 },
}

impl From<&NodeEvent> for C2NodeEvent {
    fn from(event: &NodeEvent) -> Self {
        match event {
            NodeEvent::HarnessMcpReadCall {
                reservation_id,
                activation_digest,
                record_id,
                session,
                call_id,
                request,
                deadline_unix_ms,
            } => Self::HarnessMcpReadCall {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                record_id: record_id.clone(),
                session: session.clone(),
                call_id: call_id.clone(),
                request: request.clone(),
                deadline_unix_ms: *deadline_unix_ms,
            },
            NodeEvent::Control { address, event } => Self::Control {
                address: address.clone(),
                event: C2ControlEvent::from(event),
            },
            NodeEvent::SessionRecordHistorySummarized { record_id, summary } => {
                Self::SessionRecordHistorySummarized {
                    record_id: record_id.clone(),
                    summary: *summary,
                }
            }
            NodeEvent::TerminalFrame { address, frame } => Self::TerminalFrame {
                address: address.clone(),
                frame: frame.clone(),
            },
            NodeEvent::AgentStream { address, chunk } => Self::AgentStream {
                address: address.clone(),
                chunk: chunk.clone(),
            },
            NodeEvent::ControllerChanged { controller } => Self::ControllerChanged {
                controller: controller.clone(),
            },
            NodeEvent::WorkspaceAdded { workspace } => Self::WorkspaceAdded {
                workspace: C2WorkspaceSnapshot::from(workspace),
            },
            NodeEvent::WorkspaceRemoved { workspace_id } => Self::WorkspaceRemoved {
                workspace_id: workspace_id.clone(),
            },
            NodeEvent::SessionRecordUpserted { record } => Self::SessionRecordUpserted {
                record: C2ManagedSessionRecord::from(record),
            },
            NodeEvent::SessionRecordRemoved { record_id } => Self::SessionRecordRemoved {
                record_id: record_id.clone(),
            },
            NodeEvent::ManagedWorktreeUpserted { lease } => Self::ManagedWorktreeUpserted {
                lease: lease.clone(),
            },
            NodeEvent::ManagedWorktreeRemoved { lease_id } => Self::ManagedWorktreeRemoved {
                lease_id: lease_id.clone(),
            },
            NodeEvent::ResyncRequired { oldest_available_sequence } => Self::ResyncRequired {
                oldest_available_sequence: *oldest_available_sequence,
            },
        }
    }
}

impl C2NodeEvent {
    /// Like `From`, but a `NodeEvent::Control` keeps its sanitized telemetry
    /// detail; see [`control_telemetry_detail`].
    pub fn from_node_event_with_control_detail(event: &NodeEvent) -> Self {
        match event {
            NodeEvent::Control { address, event } => Self::Control {
                address: address.clone(),
                event: C2ControlEvent {
                    detail: control_telemetry_detail(event),
                    ..C2ControlEvent::from(event)
                },
            },
            other => Self::from(other),
        }
    }

    pub fn requires_control_detail_capability(&self) -> bool {
        matches!(self, Self::Control { event, .. } if event.detail.is_some())
    }

    /// Drops the control detail, leaving the coarse tag projection.
    pub fn strip_control_detail(&mut self) {
        if let Self::Control { event, .. } = self {
            event.detail = None;
        }
    }

    pub fn requires_harness_mcp_proxy_capability(&self) -> bool {
        matches!(self, Self::HarnessMcpReadCall { .. })
    }

    pub fn harness_mcp_contract_is_valid_at(&self, now_unix_ms: u64) -> bool {
        match self {
            Self::HarnessMcpReadCall {
                reservation_id,
                activation_digest,
                record_id,
                session,
                call_id,
                request,
                deadline_unix_ms,
            } => NodeEvent::HarnessMcpReadCall {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                record_id: record_id.clone(),
                session: session.clone(),
                call_id: call_id.clone(),
                request: request.clone(),
                deadline_unix_ms: *deadline_unix_ms,
            }.harness_mcp_contract_is_valid_at(now_unix_ms),
            _ => true,
        }
    }

    pub fn requires_child_environment_profile_capability(&self) -> bool {
        matches!(self, Self::SessionRecordUpserted { record }
            if record.environment_profile.is_some())
    }

    pub fn requires_session_bundle_materialization_capability(&self) -> bool {
        matches!(self, Self::SessionRecordUpserted { record }
            if record.bundle.is_some())
    }

    pub fn requires_history_context_pack_capability(&self) -> bool {
        matches!(self, Self::SessionRecordUpserted { record }
            if record.requires_history_context_pack_capability())
    }

    pub fn requires_session_task_correlation_capability(&self) -> bool {
        matches!(self, Self::SessionRecordUpserted { record }
            if record.requires_session_task_correlation_capability())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2GitWorktreeSnapshot {
    pub path: OpaqueHostPath,
    pub head: String,
    pub branch: Option<String>,
    pub is_bare: bool,
    pub is_main: bool,
    pub locked: bool,
    pub prunable: bool,
    pub workspace_id: Option<WorkspaceId>,
}

impl From<&gate4agent_node_protocol::GitWorktreeSnapshot> for C2GitWorktreeSnapshot {
    fn from(worktree: &gate4agent_node_protocol::GitWorktreeSnapshot) -> Self {
        Self {
            path: worktree.path.clone(),
            head: worktree.head.clone(),
            branch: worktree.branch.clone(),
            is_bare: worktree.is_bare,
            is_main: worktree.is_main,
            locked: worktree.locked,
            prunable: worktree.prunable,
            workspace_id: worktree.workspace_id.clone(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2GitSnapshot {
    pub is_repository: bool,
    pub branch: Option<String>,
    pub status: Vec<gate4agent_node_protocol::GitStatusEntry>,
    pub recent_commits: Vec<gate4agent_node_protocol::GitCommitSummary>,
    pub worktrees: Vec<C2GitWorktreeSnapshot>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub managed_worktree: Option<ManagedWorktreeGitScope>,
    pub truncated: bool,
    pub diagnostic_present: bool,
}

impl From<&gate4agent_node_protocol::GitSnapshot> for C2GitSnapshot {
    fn from(git: &gate4agent_node_protocol::GitSnapshot) -> Self {
        Self {
            is_repository: git.is_repository,
            branch: git.branch.clone(),
            status: git.status.clone(),
            recent_commits: git.recent_commits.clone(),
            worktrees: git.worktrees.iter().map(C2GitWorktreeSnapshot::from).collect(),
            managed_worktree: git.managed_worktree.clone(),
            truncated: git.truncated,
            diagnostic_present: git.diagnostic.is_some(),
        }
    }
}

impl C2GitSnapshot {
    pub fn managed_worktree_is_valid_for(&self, workspace_id: &WorkspaceId) -> bool {
        self.managed_worktree.as_ref().map_or(true, |scope| {
            self.is_repository
                && self.branch.as_deref() == Some(scope.branch.as_str())
                && &scope.source_workspace_id != workspace_id
                && (u32::from(scope.active_session_count)
                    + u32::from(scope.managed_record_count))
                    > 0
        })
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct C2WorkspaceInspection {
    pub workspace_id: WorkspaceId,
    pub entries: Vec<gate4agent_node_protocol::WorkspaceEntry>,
    pub tree_truncated: bool,
    pub git: C2GitSnapshot,
    /// Additive mirror of `gate4agent_node_protocol::WorkspaceInspection::
    /// truncation` — reuses the node-protocol type directly (plain
    /// counts/bools, nothing sensitive), matching how `status`/
    /// `recent_commits` already reuse node-protocol leaf types unchanged.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub truncation: Option<gate4agent_node_protocol::WorkspaceInspectionTruncationV1>,
}

impl<'de> Deserialize<'de> for C2WorkspaceInspection {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(Deserialize)]
        struct WireInspection {
            workspace_id: WorkspaceId,
            entries: Vec<gate4agent_node_protocol::WorkspaceEntry>,
            tree_truncated: bool,
            git: C2GitSnapshot,
            #[serde(default)]
            truncation: Option<gate4agent_node_protocol::WorkspaceInspectionTruncationV1>,
        }

        let wire = WireInspection::deserialize(deserializer)?;
        if !wire.git.managed_worktree_is_valid_for(&wire.workspace_id) {
            return Err(serde::de::Error::custom(
                "managed worktree git scope is inconsistent with workspace inspection",
            ));
        }
        Ok(Self {
            workspace_id: wire.workspace_id,
            entries: wire.entries,
            tree_truncated: wire.tree_truncated,
            git: wire.git,
            truncation: wire.truncation,
        })
    }
}

impl From<&gate4agent_node_protocol::WorkspaceInspection> for C2WorkspaceInspection {
    fn from(inspection: &gate4agent_node_protocol::WorkspaceInspection) -> Self {
        let mut projected = Self {
            workspace_id: inspection.workspace_id.clone(),
            entries: inspection.entries.clone(),
            tree_truncated: inspection.tree_truncated,
            git: C2GitSnapshot::from(&inspection.git),
            truncation: inspection.truncation,
        };
        if !projected
            .git
            .managed_worktree_is_valid_for(&projected.workspace_id)
        {
            projected.git.managed_worktree = None;
        }
        projected
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum C2NodeResponse {
    Snapshot {
        event_sequence: u64,
        controller: Option<gate4agent_node_protocol::ControllerState>,
        snapshot: C2NodeSnapshot,
    },
    Resync {
        event_sequence: u64,
        oldest_available_sequence: u64,
        snapshot: C2NodeSnapshot,
        events: Vec<C2NodeEventEnvelope>,
    },
    Armed {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
        expires_at_unix_ms: u64,
    },
    Spawned {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
        receipt: ResolvedSpawnReceipt,
    },
    Activated {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
        record_id: SessionRecordId,
        session: SessionAddress,
    },
    Aborted {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
    },
    ReplyChunkAccepted {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
        record_id: SessionRecordId,
        session: SessionAddress,
        call_id: HarnessMcpCallId,
        next_offset: u32,
        completed: bool,
    },
    CallRejected {
        reservation_id: HarnessMcpReservationId,
        activation_digest: HarnessMcpActivationDigest,
        record_id: SessionRecordId,
        session: SessionAddress,
        call_id: HarnessMcpCallId,
    },
    DeliveryStageBegun {
        stage_id: DeliveryStageId,
        manifest_digest: DeliveryManifestDigestV2,
        missing_blobs: Vec<DeliveryBlobDigestV1>,
    },
    DeliveryBlobChunkAccepted {
        stage_id: DeliveryStageId,
        blob_digest: DeliveryBlobDigestV1,
        next_offset: u64,
    },
    DeliveryCommitted {
        receipt: DeliveryCommitReceiptV1,
    },
    DeliveryStageAborted {
        stage_id: DeliveryStageId,
    },
    WorkspaceInspected {
        inspection: C2WorkspaceInspection,
    },
    HostDirectoriesBrowsed {
        listing: HostDirectoryListing,
    },
    WorkspaceFileRead {
        file: WorkspaceFileRead,
    },
    WorkspaceFileWritten {
        file: WorkspaceFileRead,
    },
    WorkspaceFileCreated {
        file: WorkspaceFileRead,
    },
    WorkspaceDirectoryCreated {
        workspace_id: WorkspaceId,
        entry: gate4agent_node_protocol::WorkspaceEntry,
    },
    GitHistoryRead {
        workspace_id: WorkspaceId,
        page: GitHistoryPage,
    },
    GitDiffRead {
        workspace_id: WorkspaceId,
        diff: GitDiff,
    },
    Controller {
        controller: Option<gate4agent_node_protocol::ControllerState>,
    },
    SpawnAccepted { session: SessionAddress },
    SpawnSpecAccepted { receipt: ResolvedSpawnReceipt },
    ManagedWorktreeSpawnAccepted { receipt: ManagedWorktreeSpawnReceipt },
    ManagedWorktreeCleanup { lease: ManagedWorktreeLeaseSnapshot },
    SessionRecordUpdated { record: C2ManagedSessionRecord },
    ProviderSessionIndexed { record: C2ManagedSessionRecord },
    NativeSessionIndexed {
        selection: NativeSessionSelection,
        record: C2ManagedSessionRecord,
    },
    SessionRecordResumed {
        record: C2ManagedSessionRecord,
        session: SessionAddress,
    },
    SessionRecordForgotten { record_id: SessionRecordId },
    NativeSessionsCataloged {
        route: NativeSessionCatalogRoute,
        #[serde(deserialize_with = "deserialize_c2_native_session_catalog_entries")]
        entries: Vec<NativeSessionCatalogEntry>,
        #[serde(
            default,
            skip_serializing_if = "Option::is_none",
            deserialize_with = "deserialize_optional_c2_native_session_catalog_summary"
        )]
        summary: Option<NativeSessionCatalogSummary>,
    },
    NativeSessionsPaged {
        route: NativeSessionCatalogRoute,
        #[serde(deserialize_with = "deserialize_c2_native_session_catalog_page")]
        page: NativeSessionCatalogPage,
    },
    NativeSessionPreviewed {
        selection: NativeSessionSelection,
        #[serde(deserialize_with = "deserialize_c2_native_session_preview")]
        preview: NativeSessionPreview,
    },
    SessionRecordPreviewed {
        record_id: SessionRecordId,
        #[serde(deserialize_with = "deserialize_c2_session_record_preview")]
        preview: SessionRecordPreview,
    },
    HistoryDiscovered {
        session: SessionAddress,
        #[serde(deserialize_with = "deserialize_c2_history_candidates")]
        candidates: Vec<HistoryCandidateSummary>,
    },
    HistoryLoaded {
        session: SessionAddress,
        #[serde(deserialize_with = "deserialize_c2_history_session_id")]
        session_id: String,
        message_count: u64,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        completed_turn_count: Option<u64>,
    },
    ContextPackExported { context: ResolvedContextPackReceipt },
    ContextPackForSessionRecordExported {
        record_id: SessionRecordId,
        session: SessionAddress,
        context: ResolvedContextPackReceipt,
    },
    ContextPackForgotten { context_id: SpawnContextId },
    DurableContextPackResolved { context: ResolvedContextPackReceipt },
    ContextPackBytesRead { pack: ContextPackBytesRead },
    WorkspaceRegistered {
        workspace: C2WorkspaceSnapshot,
    },
    StandaloneWorkspaceCreated {
        workspace: C2WorkspaceSnapshot,
    },
    WorkspaceUnregistered { workspace_id: WorkspaceId },
    WorktreeCreated {
        worktree: C2GitWorktreeSnapshot,
        workspace: C2WorkspaceSnapshot,
    },
    WorktreeRemoved {
        target_root: OpaqueHostPath,
        workspace_id: Option<WorkspaceId>,
    },
    Accepted,
    ShuttingDown,
}

impl From<&NodeResponse> for C2NodeResponse {
    fn from(response: &NodeResponse) -> Self {
        match response {
            NodeResponse::Snapshot { event_sequence, controller, snapshot } => Self::Snapshot {
                event_sequence: *event_sequence,
                controller: controller.clone(),
                snapshot: C2NodeSnapshot::from(snapshot),
            },
            NodeResponse::Resync {
                event_sequence,
                oldest_available_sequence,
                snapshot,
                events,
            } => Self::Resync {
                event_sequence: *event_sequence,
                oldest_available_sequence: *oldest_available_sequence,
                snapshot: C2NodeSnapshot::from(snapshot),
                events: events.iter().map(C2NodeEventEnvelope::from).collect(),
            },
            NodeResponse::Armed {
                reservation_id,
                activation_digest,
                expires_at_unix_ms,
            } => Self::Armed {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                expires_at_unix_ms: *expires_at_unix_ms,
            },
            NodeResponse::Spawned { reservation_id, activation_digest, receipt } => {
                Self::Spawned {
                    reservation_id: reservation_id.clone(),
                    activation_digest: activation_digest.clone(),
                    receipt: receipt.clone(),
                }
            }
            NodeResponse::Activated {
                reservation_id,
                activation_digest,
                record_id,
                session,
            } => Self::Activated {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                record_id: record_id.clone(),
                session: session.clone(),
            },
            NodeResponse::Aborted { reservation_id, activation_digest } => Self::Aborted {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
            },
            NodeResponse::ReplyChunkAccepted {
                reservation_id,
                activation_digest,
                record_id,
                session,
                call_id,
                next_offset,
                completed,
            } => Self::ReplyChunkAccepted {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                record_id: record_id.clone(),
                session: session.clone(),
                call_id: call_id.clone(),
                next_offset: *next_offset,
                completed: *completed,
            },
            NodeResponse::CallRejected {
                reservation_id,
                activation_digest,
                record_id,
                session,
                call_id,
            } => Self::CallRejected {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                record_id: record_id.clone(),
                session: session.clone(),
                call_id: call_id.clone(),
            },
            NodeResponse::DeliveryStageBegun {
                stage_id,
                manifest_digest,
                missing_blobs,
            } => Self::DeliveryStageBegun {
                stage_id: stage_id.clone(),
                manifest_digest: manifest_digest.clone(),
                missing_blobs: missing_blobs.clone(),
            },
            NodeResponse::DeliveryBlobChunkAccepted {
                stage_id,
                blob_digest,
                next_offset,
            } => Self::DeliveryBlobChunkAccepted {
                stage_id: stage_id.clone(),
                blob_digest: blob_digest.clone(),
                next_offset: *next_offset,
            },
            NodeResponse::DeliveryCommitted { receipt } => Self::DeliveryCommitted {
                receipt: receipt.clone(),
            },
            NodeResponse::DeliveryStageAborted { stage_id } => Self::DeliveryStageAborted {
                stage_id: stage_id.clone(),
            },
            NodeResponse::WorkspaceInspected { inspection } => Self::WorkspaceInspected {
                inspection: C2WorkspaceInspection::from(inspection),
            },
            NodeResponse::HostDirectoriesBrowsed { listing } => Self::HostDirectoriesBrowsed {
                listing: listing.clone(),
            },
            NodeResponse::WorkspaceFileRead { file } => Self::WorkspaceFileRead {
                file: file.clone(),
            },
            NodeResponse::WorkspaceFileWritten { file } => Self::WorkspaceFileWritten {
                file: file.clone(),
            },
            NodeResponse::WorkspaceFileCreated { file } => Self::WorkspaceFileCreated {
                file: file.clone(),
            },
            NodeResponse::WorkspaceDirectoryCreated { workspace_id, entry } => {
                Self::WorkspaceDirectoryCreated {
                    workspace_id: workspace_id.clone(),
                    entry: entry.clone(),
                }
            }
            NodeResponse::GitHistoryRead { workspace_id, page } => Self::GitHistoryRead {
                workspace_id: workspace_id.clone(),
                page: page.clone(),
            },
            NodeResponse::GitDiffRead { workspace_id, diff } => Self::GitDiffRead {
                workspace_id: workspace_id.clone(),
                diff: diff.clone(),
            },
            NodeResponse::Controller { controller } => Self::Controller {
                controller: controller.clone(),
            },
            NodeResponse::SpawnAccepted { session } => Self::SpawnAccepted { session: session.clone() },
            NodeResponse::SpawnSpecAccepted { receipt } => Self::SpawnSpecAccepted {
                receipt: receipt.clone(),
            },
            NodeResponse::ManagedWorktreeSpawnAccepted { receipt } => {
                Self::ManagedWorktreeSpawnAccepted {
                    receipt: receipt.clone(),
                }
            }
            NodeResponse::ManagedWorktreeCleanup { lease } => Self::ManagedWorktreeCleanup {
                lease: lease.clone(),
            },
            NodeResponse::SessionRecordUpdated { record } => Self::SessionRecordUpdated {
                record: C2ManagedSessionRecord::from(record),
            },
            NodeResponse::ProviderSessionIndexed { record } => Self::ProviderSessionIndexed {
                record: C2ManagedSessionRecord::from(record),
            },
            NodeResponse::NativeSessionIndexed { selection, record } => {
                Self::NativeSessionIndexed {
                    selection: selection.clone(),
                    record: C2ManagedSessionRecord::from(record),
                }
            }
            NodeResponse::SessionRecordResumed { record, session } => Self::SessionRecordResumed {
                record: C2ManagedSessionRecord::from(record),
                session: session.clone(),
            },
            NodeResponse::SessionRecordForgotten { record_id } => Self::SessionRecordForgotten {
                record_id: record_id.clone(),
            },
            NodeResponse::NativeSessionsCataloged {
                route,
                entries,
                summary,
            } => Self::NativeSessionsCataloged {
                route: route.clone(),
                entries: entries.clone(),
                summary: summary.clone(),
            },
            NodeResponse::NativeSessionsPaged {
                route,
                page,
            } => Self::NativeSessionsPaged {
                route: route.clone(),
                page: page.clone(),
            },
            NodeResponse::NativeSessionPreviewed {
                selection,
                preview,
            } => Self::NativeSessionPreviewed {
                selection: selection.clone(),
                preview: preview.clone(),
            },
            NodeResponse::SessionRecordPreviewed { record_id, preview } => {
                Self::SessionRecordPreviewed {
                    record_id: record_id.clone(),
                    preview: preview.clone(),
                }
            }
            NodeResponse::HistoryDiscovered { session, candidates } => Self::HistoryDiscovered {
                session: session.clone(),
                candidates: candidates.clone(),
            },
            NodeResponse::HistoryLoaded {
                session,
                session_id,
                message_count,
                completed_turn_count,
            } => Self::HistoryLoaded {
                session: session.clone(),
                session_id: session_id.clone(),
                message_count: *message_count,
                completed_turn_count: *completed_turn_count,
            },
            NodeResponse::ContextPackExported { context } => Self::ContextPackExported {
                context: context.clone(),
            },
            NodeResponse::ContextPackForSessionRecordExported {
                record_id,
                session,
                context,
            } => Self::ContextPackForSessionRecordExported {
                record_id: record_id.clone(),
                session: session.clone(),
                context: context.clone(),
            },
            NodeResponse::ContextPackForgotten { context_id } => Self::ContextPackForgotten {
                context_id: context_id.clone(),
            },
            NodeResponse::DurableContextPackResolved { context } => {
                Self::DurableContextPackResolved {
                    context: context.clone(),
                }
            }
            NodeResponse::ContextPackBytesRead { pack } => Self::ContextPackBytesRead { pack: pack.clone() },
            NodeResponse::WorkspaceRegistered { workspace } => Self::WorkspaceRegistered {
                workspace: C2WorkspaceSnapshot::from(workspace),
            },
            NodeResponse::StandaloneWorkspaceCreated { workspace } => {
                Self::StandaloneWorkspaceCreated {
                    workspace: C2WorkspaceSnapshot::from(workspace),
                }
            }
            NodeResponse::WorkspaceUnregistered { workspace_id } => Self::WorkspaceUnregistered {
                workspace_id: workspace_id.clone(),
            },
            NodeResponse::WorktreeCreated { worktree, workspace } => Self::WorktreeCreated {
                worktree: C2GitWorktreeSnapshot::from(worktree),
                workspace: C2WorkspaceSnapshot::from(workspace),
            },
            NodeResponse::WorktreeRemoved { target_root, workspace_id } => Self::WorktreeRemoved {
                target_root: target_root.clone(),
                workspace_id: workspace_id.clone(),
            },
            NodeResponse::Accepted => Self::Accepted,
            NodeResponse::ShuttingDown => Self::ShuttingDown,
        }
    }
}

impl C2NodeResponse {
    /// Like `From`, but the control events of a `Resync` keep their sanitized
    /// telemetry detail; see [`control_telemetry_detail`].
    pub fn from_node_response_with_control_detail(response: &NodeResponse) -> Self {
        let mut projected = Self::from(response);
        if let (
            Self::Resync { events: projected_events, .. },
            NodeResponse::Resync { events, .. },
        ) = (&mut projected, response)
        {
            *projected_events = events
                .iter()
                .map(C2NodeEventEnvelope::from_node_envelope_with_control_detail)
                .collect();
        }
        projected
    }

    pub fn requires_control_detail_capability(&self) -> bool {
        matches!(self, Self::Resync { events, .. }
            if events.iter().any(|event| event.event.requires_control_detail_capability()))
    }

    pub fn strip_control_detail(&mut self) {
        if let Self::Resync { events, .. } = self {
            for event in events {
                event.event.strip_control_detail();
            }
        }
    }

    pub fn requires_session_record_context_export_capability(&self) -> bool {
        matches!(self, Self::ContextPackForSessionRecordExported { .. })
    }

    pub fn requires_harness_mcp_proxy_capability(&self) -> bool {
        matches!(self,
            Self::Armed { .. }
                | Self::Spawned { .. }
                | Self::Activated { .. }
                | Self::Aborted { .. }
                | Self::ReplyChunkAccepted { .. }
                | Self::CallRejected { .. })
            || matches!(self, Self::SpawnSpecAccepted { receipt }
                if receipt.harness_mcp_proxy.is_some())
            || matches!(self, Self::ManagedWorktreeSpawnAccepted { receipt }
                if receipt.spawn.harness_mcp_proxy.is_some())
    }
}

impl C2NodeResponse {
    pub fn native_session_catalog_contract_is_valid(&self) -> bool {
        match self {
            Self::NativeSessionsCataloged { route, entries, summary } => {
                route.validate().is_ok()
                    && entries.len()
                        <= usize::from(gate4agent_types::NATIVE_SESSION_CATALOG_LIMIT_MAX)
                    && entries.iter().enumerate().all(|(index, entry)| {
                        entry.validate_for_route(route).is_ok()
                            && !entries[..index].iter().any(|existing| {
                                existing.selection_id == entry.selection_id
                                    || entry.record_id.as_ref().is_some_and(|record_id| {
                                        existing.record_id.as_ref() == Some(record_id)
                                    })
                            })
                    })
                    && summary.as_ref().map_or(true, |summary| {
                        summary.validate_initial_entries(entries.len()).is_ok()
                    })
            }
            Self::NativeSessionsPaged { route, page } => {
                page.validate_for_route(route).is_ok()
            }
            _ => true,
        }
    }

    pub fn requires_native_session_catalog_capability(&self) -> bool {
        matches!(self, Self::NativeSessionsCataloged { .. })
    }

    pub fn requires_native_session_catalog_paging_capability(&self) -> bool {
        matches!(self, Self::NativeSessionsPaged { .. })
    }

    pub fn requires_native_session_preview_capability(&self) -> bool {
        matches!(
            self,
            Self::NativeSessionPreviewed { .. } | Self::SessionRecordPreviewed { .. }
        )
    }

    pub fn requires_native_session_index_capability(&self) -> bool {
        matches!(self, Self::NativeSessionIndexed { .. })
    }

    pub fn native_session_index_contract_is_valid(&self) -> bool {
        match self {
            Self::NativeSessionIndexed { selection, record } => {
                selection.validate().is_ok()
                    && selection.route.scope == NativeSessionCatalogScope::Workspace
                    && selection.route.workspace_id.as_ref() == Some(&record.workspace_id)
                    && selection.route.provider == record.provider
            }
            _ => false,
        }
    }

    pub fn native_session_preview_contract_is_valid(&self) -> bool {
        match self {
            Self::NativeSessionPreviewed { selection, preview } => {
                selection.validate().is_ok() && preview.validate().is_ok()
            }
            Self::SessionRecordPreviewed { preview, .. } => preview.validate().is_ok(),
            _ => true,
        }
    }

    pub fn requires_host_directory_browse_capability(&self) -> bool {
        matches!(self, Self::HostDirectoriesBrowsed { .. })
    }

    pub fn requires_workspace_entry_create_capability(&self) -> bool {
        matches!(
            self,
            Self::WorkspaceFileCreated { .. } | Self::WorkspaceDirectoryCreated { .. }
        )
    }

    pub fn requires_child_environment_profile_capability(&self) -> bool {
        match self {
            Self::Snapshot { snapshot, .. } => {
                snapshot.requires_child_environment_profile_capability()
            }
            Self::Resync {
                snapshot, events, ..
            } => {
                snapshot.requires_child_environment_profile_capability()
                    || events.iter().any(|event| {
                        event.event.requires_child_environment_profile_capability()
                    })
            }
            Self::SpawnSpecAccepted { receipt }
            | Self::Spawned { receipt, .. } => receipt.environment_profile.is_some(),
            Self::ManagedWorktreeSpawnAccepted { receipt } => {
                receipt.spawn.environment_profile.is_some()
            }
            Self::SessionRecordUpdated { record }
            | Self::ProviderSessionIndexed { record }
            | Self::NativeSessionIndexed { record, .. }
            | Self::SessionRecordResumed { record, .. } => {
                record.environment_profile.is_some()
            }
            Self::DurableContextPackResolved { .. } | Self::ContextPackBytesRead { .. } => false,
            Self::Armed { .. }
            | Self::Activated { .. }
            | Self::Aborted { .. }
            | Self::ReplyChunkAccepted { .. }
            | Self::CallRejected { .. }
            | Self::WorkspaceInspected { .. }
            | Self::DeliveryStageBegun { .. }
            | Self::DeliveryBlobChunkAccepted { .. }
            | Self::DeliveryCommitted { .. }
            | Self::DeliveryStageAborted { .. }
            | Self::HostDirectoriesBrowsed { .. }
            | Self::WorkspaceFileRead { .. }
            | Self::WorkspaceFileWritten { .. }
            | Self::WorkspaceFileCreated { .. }
            | Self::WorkspaceDirectoryCreated { .. }
            | Self::GitHistoryRead { .. }
            | Self::GitDiffRead { .. }
            | Self::Controller { .. }
            | Self::SpawnAccepted { .. }
            | Self::ManagedWorktreeCleanup { .. }
            | Self::SessionRecordForgotten { .. }
            | Self::NativeSessionsCataloged { .. }
            | Self::NativeSessionsPaged { .. }
            | Self::NativeSessionPreviewed { .. }
            | Self::SessionRecordPreviewed { .. }
            | Self::HistoryDiscovered { .. }
            | Self::HistoryLoaded { .. }
            | Self::ContextPackForSessionRecordExported { .. }
            | Self::ContextPackExported { .. }
            | Self::ContextPackForgotten { .. }
            | Self::WorkspaceRegistered { .. }
            | Self::StandaloneWorkspaceCreated { .. }
            | Self::WorkspaceUnregistered { .. }
            | Self::WorktreeCreated { .. }
            | Self::WorktreeRemoved { .. }
            | Self::Accepted
            | Self::ShuttingDown => false,
        }
    }

    pub fn requires_session_bundle_materialization_capability(&self) -> bool {
        match self {
            Self::Snapshot { snapshot, .. } => {
                snapshot.requires_session_bundle_materialization_capability()
            }
            Self::Resync {
                snapshot, events, ..
            } => {
                snapshot.requires_session_bundle_materialization_capability()
                    || events.iter().any(|event| {
                        event.event.requires_session_bundle_materialization_capability()
                    })
            }
            Self::SpawnSpecAccepted { receipt }
            | Self::Spawned { receipt, .. } => receipt.bundle.is_some(),
            Self::ManagedWorktreeSpawnAccepted { receipt } => receipt.spawn.bundle.is_some(),
            Self::SessionRecordUpdated { record }
            | Self::ProviderSessionIndexed { record }
            | Self::NativeSessionIndexed { record, .. }
            | Self::SessionRecordResumed { record, .. } => record.bundle.is_some(),
            Self::DurableContextPackResolved { .. } | Self::ContextPackBytesRead { .. } => false,
            Self::Armed { .. }
            | Self::Activated { .. }
            | Self::Aborted { .. }
            | Self::ReplyChunkAccepted { .. }
            | Self::CallRejected { .. }
            | Self::WorkspaceInspected { .. }
            | Self::DeliveryStageBegun { .. }
            | Self::DeliveryBlobChunkAccepted { .. }
            | Self::DeliveryCommitted { .. }
            | Self::DeliveryStageAborted { .. }
            | Self::HostDirectoriesBrowsed { .. }
            | Self::WorkspaceFileRead { .. }
            | Self::WorkspaceFileWritten { .. }
            | Self::WorkspaceFileCreated { .. }
            | Self::WorkspaceDirectoryCreated { .. }
            | Self::GitHistoryRead { .. }
            | Self::GitDiffRead { .. }
            | Self::Controller { .. }
            | Self::SpawnAccepted { .. }
            | Self::ManagedWorktreeCleanup { .. }
            | Self::SessionRecordForgotten { .. }
            | Self::NativeSessionsCataloged { .. }
            | Self::NativeSessionsPaged { .. }
            | Self::NativeSessionPreviewed { .. }
            | Self::SessionRecordPreviewed { .. }
            | Self::HistoryDiscovered { .. }
            | Self::HistoryLoaded { .. }
            | Self::ContextPackForSessionRecordExported { .. }
            | Self::ContextPackExported { .. }
            | Self::ContextPackForgotten { .. }
            | Self::WorkspaceRegistered { .. }
            | Self::StandaloneWorkspaceCreated { .. }
            | Self::WorkspaceUnregistered { .. }
            | Self::WorktreeCreated { .. }
            | Self::WorktreeRemoved { .. }
            | Self::Accepted
            | Self::ShuttingDown => false,
        }
    }

    pub fn requires_history_context_pack_capability(&self) -> bool {
        match self {
            Self::Snapshot { snapshot, .. } => {
                snapshot.requires_history_context_pack_capability()
            }
            Self::Resync {
                snapshot, events, ..
            } => {
                snapshot.requires_history_context_pack_capability()
                    || events.iter().any(|event| {
                        event.event.requires_history_context_pack_capability()
                    })
            }
            Self::SpawnSpecAccepted { receipt }
            | Self::Spawned { receipt, .. } => {
                receipt.context_id.is_some() || receipt.context.is_some()
            }
            Self::ManagedWorktreeSpawnAccepted { receipt } => {
                receipt.spawn.context_id.is_some() || receipt.spawn.context.is_some()
            }
            Self::SessionRecordUpdated { record }
            | Self::ProviderSessionIndexed { record }
            | Self::NativeSessionIndexed { record, .. }
            | Self::SessionRecordResumed { record, .. } => {
                record.requires_history_context_pack_capability()
            }
            Self::HistoryDiscovered { .. }
            | Self::HistoryLoaded { .. }
            | Self::ContextPackForSessionRecordExported { .. }
            | Self::ContextPackExported { .. }
            | Self::ContextPackForgotten { .. }
            | Self::DurableContextPackResolved { .. }
            | Self::ContextPackBytesRead { .. } => true,
            Self::Armed { .. }
            | Self::Activated { .. }
            | Self::Aborted { .. }
            | Self::ReplyChunkAccepted { .. }
            | Self::CallRejected { .. }
            | Self::WorkspaceInspected { .. }
            | Self::DeliveryStageBegun { .. }
            | Self::DeliveryBlobChunkAccepted { .. }
            | Self::DeliveryCommitted { .. }
            | Self::DeliveryStageAborted { .. }
            | Self::HostDirectoriesBrowsed { .. }
            | Self::WorkspaceFileRead { .. }
            | Self::WorkspaceFileWritten { .. }
            | Self::WorkspaceFileCreated { .. }
            | Self::WorkspaceDirectoryCreated { .. }
            | Self::GitHistoryRead { .. }
            | Self::GitDiffRead { .. }
            | Self::Controller { .. }
            | Self::SpawnAccepted { .. }
            | Self::ManagedWorktreeCleanup { .. }
            | Self::SessionRecordForgotten { .. }
            | Self::NativeSessionsCataloged { .. }
            | Self::NativeSessionsPaged { .. }
            | Self::NativeSessionPreviewed { .. }
            | Self::SessionRecordPreviewed { .. }
            | Self::WorkspaceRegistered { .. }
            | Self::StandaloneWorkspaceCreated { .. }
            | Self::WorkspaceUnregistered { .. }
            | Self::WorktreeCreated { .. }
            | Self::WorktreeRemoved { .. }
            | Self::Accepted
            | Self::ShuttingDown => false,
        }
    }

    pub fn requires_session_task_correlation_capability(&self) -> bool {
        match self {
            Self::Snapshot { snapshot, .. } => {
                snapshot.requires_session_task_correlation_capability()
            }
            Self::Resync { snapshot, events, .. } => {
                snapshot.requires_session_task_correlation_capability()
                    || events.iter().any(|event| {
                        event.event.requires_session_task_correlation_capability()
                    })
            }
            Self::SessionRecordUpdated { record }
            | Self::ProviderSessionIndexed { record }
            | Self::NativeSessionIndexed { record, .. }
            | Self::SessionRecordResumed { record, .. } => {
                record.requires_session_task_correlation_capability()
            }
            _ => false,
        }
    }
}

fn deserialize_c2_native_session_catalog_entries<'de, D>(
    deserializer: D,
) -> Result<Vec<NativeSessionCatalogEntry>, D::Error>
where
    D: Deserializer<'de>,
{
    let entries = Vec::<NativeSessionCatalogEntry>::deserialize(deserializer)?;
    if entries.len() > usize::from(gate4agent_types::NATIVE_SESSION_CATALOG_LIMIT_MAX) {
        return Err(serde::de::Error::custom(
            "C2 native session catalog exceeds the supported bounded range",
        ));
    }
    for (index, entry) in entries.iter().enumerate() {
        entry.validate().map_err(serde::de::Error::custom)?;
        if entries[..index]
            .iter()
            .any(|existing| {
                existing.selection_id == entry.selection_id
                    || entry.record_id.as_ref().is_some_and(|record_id| {
                        existing.record_id.as_ref() == Some(record_id)
                    })
            })
        {
            return Err(serde::de::Error::custom(
                "C2 native session catalog contains a duplicate selection or managed record",
            ));
        }
    }
    Ok(entries)
}

fn deserialize_optional_c2_native_session_catalog_summary<'de, D>(
    deserializer: D,
) -> Result<Option<NativeSessionCatalogSummary>, D::Error>
where
    D: Deserializer<'de>,
{
    let summary = Option::<NativeSessionCatalogSummary>::deserialize(deserializer)?;
    if let Some(summary) = summary.as_ref() {
        summary.validate().map_err(serde::de::Error::custom)?;
    }
    Ok(summary)
}

fn deserialize_c2_native_session_catalog_page<'de, D>(
    deserializer: D,
) -> Result<NativeSessionCatalogPage, D::Error>
where
    D: Deserializer<'de>,
{
    let page = NativeSessionCatalogPage::deserialize(deserializer)?;
    page.validate().map_err(serde::de::Error::custom)?;
    Ok(page)
}

fn deserialize_c2_native_session_preview<'de, D>(
    deserializer: D,
) -> Result<NativeSessionPreview, D::Error>
where
    D: Deserializer<'de>,
{
    let preview = NativeSessionPreview::deserialize(deserializer)?;
    preview.validate().map_err(serde::de::Error::custom)?;
    Ok(preview)
}

fn deserialize_c2_session_record_preview<'de, D>(
    deserializer: D,
) -> Result<SessionRecordPreview, D::Error>
where
    D: Deserializer<'de>,
{
    let preview = SessionRecordPreview::deserialize(deserializer)?;
    preview.validate().map_err(serde::de::Error::custom)?;
    Ok(preview)
}

fn deserialize_c2_history_candidates<'de, D>(
    deserializer: D,
) -> Result<Vec<HistoryCandidateSummary>, D::Error>
where
    D: Deserializer<'de>,
{
    let candidates = Vec::<HistoryCandidateSummary>::deserialize(deserializer)?;
    if candidates.len() > usize::from(gate4agent_types::HISTORY_DISCOVERY_LIMIT_MAX) {
        return Err(serde::de::Error::custom(
            "C2 history candidate count exceeds the discovery limit",
        ));
    }
    for (index, candidate) in candidates.iter().enumerate() {
        candidate.validate().map_err(serde::de::Error::custom)?;
        if candidates[..index]
            .iter()
            .any(|existing| existing.id == candidate.id)
        {
            return Err(serde::de::Error::custom(
                "C2 history candidates contain a duplicate candidate ID",
            ));
        }
    }
    Ok(candidates)
}

fn deserialize_c2_history_session_id<'de, D>(deserializer: D) -> Result<String, D::Error>
where
    D: Deserializer<'de>,
{
    let session_id = String::deserialize(deserializer)?;
    let validation = gate4agent_types::HistorySessionRecord {
        session_id: session_id.clone(),
        title: None,
        cwd: None,
        model: None,
        message_count: 0,
        completed_turn_count: None,
        total_tokens: 0,
        messages: Vec::new(),
    };
    validation.validate().map_err(serde::de::Error::custom)?;
    Ok(session_id)
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum C2RelayFailureCode {
    UnknownNode,
    NodeOffline,
    StaleNodeIncarnation,
    RelayBusy,
    OperatorAlreadyConnected,
    RequestIdReused,
    RequestForbidden,
    ClientLagged,
    ShuttingDown,
    /// The connecting side's [`BUILD_STAMP`] did not match this side's own
    /// -- carries nothing itself; the accompanying `C2RelayFailure::
    /// message` names both stamps (`"build stamp mismatch: local=<s>
    /// remote=<s>"`), the same "both values, one text" idiom the harness
    /// operator/read wires already use for their own build-stamp
    /// mismatches. Sent pre-handshake, before the connection is closed, so
    /// the peer sees a named refusal instead of a bare disconnect.
    BuildStampMismatch,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2RelayFailure {
    pub code: C2RelayFailureCode,
    pub message: String,
    pub current_incarnation_id: Option<NodeIncarnationId>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2ClientHello {
    pub build_stamp: String,
    pub client_nonce: [u8; C2_AUTH_NONCE_BYTES],
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub compatibility: Option<ClientCompatibilityOffer>,
}

impl C2ClientHello {
    pub fn new(client_nonce: [u8; C2_AUTH_NONCE_BYTES]) -> Self {
        Self {
            build_stamp: BUILD_STAMP.to_owned(),
            client_nonce,
            compatibility: None,
        }
    }

    pub fn negotiating(
        client_nonce: [u8; C2_AUTH_NONCE_BYTES],
        compatibility: ClientCompatibilityOffer,
    ) -> Self {
        Self {
            build_stamp: BUILD_STAMP.to_owned(),
            client_nonce,
            compatibility: Some(compatibility),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2ControlCompatibilitySupport {
    pub build_stamp: String,
    #[serde(default)]
    pub capabilities: Vec<CapabilityId>,
    pub host: HostDescriptor,
    pub path_semantics: PathSemantics,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct NegotiatedC2ControlCompatibility {
    pub build_stamp: String,
    #[serde(default)]
    pub capabilities: Vec<CapabilityId>,
    pub host: HostDescriptor,
    pub path_semantics: PathSemantics,
}

impl C2ControlCompatibilitySupport {
    pub fn negotiate(
        &self,
        hello: &C2ClientHello,
    ) -> Result<NegotiatedC2ControlCompatibility, ProtocolNegotiationError> {
        if self.build_stamp != hello.build_stamp {
            return Err(ProtocolNegotiationError::BuildStampMismatch {
                local: self.build_stamp.clone(),
                remote: hello.build_stamp.clone(),
            });
        }
        let legacy;
        let offer = match hello.compatibility.as_ref() {
            Some(offer) => offer,
            None => {
                legacy = ClientCompatibilityOffer::local();
                &legacy
            }
        };
        let capabilities = self
            .capabilities
            .iter()
            .filter(|capability| offer.capabilities.contains(capability))
            .cloned()
            .collect();
        Ok(NegotiatedC2ControlCompatibility {
            build_stamp: self.build_stamp.clone(),
            capabilities,
            host: self.host.clone(),
            path_semantics: self.path_semantics.clone(),
        })
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2ServerChallenge {
    pub build_stamp: String,
    pub server_nonce: [u8; C2_AUTH_NONCE_BYTES],
    pub server_proof: [u8; C2_AUTH_PROOF_BYTES],
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub compatibility: Option<NegotiatedC2ControlCompatibility>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2ClientAuthentication {
    pub client_proof: [u8; C2_AUTH_PROOF_BYTES],
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2Hello {
    pub build_stamp: String,
    pub connection_id: u64,
    pub status: StatusResponse,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub compatibility: Option<NegotiatedC2ControlCompatibility>,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum C2RelayRoute {
    #[default]
    Unknown,
    LocalIpc,
    SshForwardedLoopback,
}

impl C2RelayRoute {
    pub const fn is_unknown(&self) -> bool {
        matches!(self, Self::Unknown)
    }

    fn from_transport_label(transport_label: &str) -> Self {
        match transport_label {
            "windows-named-pipe" | "unix-domain-socket" => Self::LocalIpc,
            "ssh-forwarded-loopback" => Self::SshForwardedLoopback,
            _ => Self::Unknown,
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2TopologyNode {
    pub node_id: NodeId,
    pub endpoint: String,
    #[serde(default, skip_serializing_if = "C2RelayRoute::is_unknown")]
    pub relay_route: C2RelayRoute,
    pub transport: NodeTransportState,
    pub current_incarnation_id: Option<NodeIncarnationId>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub provider_contracts: Vec<ProviderContractSupport>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub provider_adapter_contracts: Vec<ProviderAdapterContractSupport>,
    #[serde(default, skip_serializing_if = "ProviderRuntimeStatuses::is_empty")]
    pub provider_runtime_statuses: ProviderRuntimeStatuses,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2Topology {
    pub nodes: Vec<C2TopologyNode>,
}

impl C2Topology {
    pub fn from_status(status: &StatusResponse) -> Self {
        Self::from_status_with_capabilities(status, true, true)
    }

    pub fn from_status_with_provider_contracts(
        status: &StatusResponse,
        include_provider_contracts: bool,
    ) -> Self {
        Self::from_status_with_capabilities(
            status,
            include_provider_contracts,
            include_provider_contracts,
        )
    }

    pub fn from_status_with_capabilities(
        status: &StatusResponse,
        include_provider_contracts: bool,
        include_provider_runtime_status: bool,
    ) -> Self {
        let nodes = status.nodes.iter().take(MAX_C2_NODES).map(|(node_id, observed)| {
            let provider_contract_inventory = include_provider_contracts
                .then_some(observed.inventory.as_ref())
                .flatten();
            let provider_runtime_statuses = if include_provider_runtime_status {
                observed
                    .inventory
                    .as_ref()
                    .map(|inventory| inventory.provider_runtime_statuses.clone())
                    .unwrap_or_default()
            } else {
                ProviderRuntimeStatuses::default()
            };
            C2TopologyNode {
                node_id: node_id.clone(),
                endpoint: observed.endpoint.clone(),
                relay_route: C2RelayRoute::from_transport_label(&observed.transport_label),
                transport: observed.transport,
                current_incarnation_id: observed.cursor.map(|cursor| cursor.incarnation_id),
                provider_contracts: provider_contract_inventory
                    .map(|inventory| inventory.provider_contracts.clone())
                    .unwrap_or_default(),
                provider_adapter_contracts: provider_contract_inventory
                    .map(|inventory| inventory.provider_adapter_contracts.clone())
                    .unwrap_or_default(),
                provider_runtime_statuses,
            }
        }).collect();
        Self { nodes }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2RequestEnvelope {
    pub request_id: C2RequestId,
    pub request: RoutedNodeRequest,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct C2ReplyEnvelope {
    pub request_id: C2RequestId,
    pub result: Result<RoutedNodeResponse, C2RelayFailure>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "payload", rename_all = "kebab-case")]
pub enum C2ClientFrame {
    Hello(C2ClientHello),
    Authenticate(C2ClientAuthentication),
    Request(C2RequestEnvelope),
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "payload", rename_all = "kebab-case")]
pub enum C2ServerFrame {
    Challenge(C2ServerChallenge),
    Hello(C2Hello),
    Reply(C2ReplyEnvelope),
    Event(RoutedNodeEvent),
    Topology(C2Topology),
    Rejected(C2RelayFailure),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum C2AuthDirection {
    Server,
    Client,
}

pub fn c2_auth_transcript(
    direction: C2AuthDirection,
    client_nonce: &[u8; C2_AUTH_NONCE_BYTES],
    server_nonce: &[u8; C2_AUTH_NONCE_BYTES],
) -> Vec<u8> {
    let mut message = Vec::with_capacity(32 + (C2_AUTH_NONCE_BYTES * 2));
    message.extend_from_slice(b"gate4agent-c2-control-auth-v2\0");
    encode_bounded_str(&mut message, BUILD_STAMP);
    message.push(match direction { C2AuthDirection::Server => 1, C2AuthDirection::Client => 2 });
    message.extend_from_slice(client_nonce);
    message.extend_from_slice(server_nonce);
    message
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum C2AuthTranscriptError {
    TooManyCapabilities {
        section: &'static str,
        count: usize,
        max: usize,
    },
    TooLong {
        len: usize,
        max: usize,
    },
}

impl fmt::Display for C2AuthTranscriptError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooManyCapabilities { section, count, max } => write!(
                formatter,
                "{section} contains {count} capabilities, exceeding the {max}-entry authentication limit",
            ),
            Self::TooLong { len, max } => write!(
                formatter,
                "C2 compatibility authentication transcript is {len} bytes, exceeding the {max}-byte limit",
            ),
        }
    }
}

impl std::error::Error for C2AuthTranscriptError {}

pub fn c2_bound_auth_transcript(
    direction: C2AuthDirection,
    client_nonce: &[u8; C2_AUTH_NONCE_BYTES],
    server_nonce: &[u8; C2_AUTH_NONCE_BYTES],
    offer: &ClientCompatibilityOffer,
    selected: &NegotiatedC2ControlCompatibility,
) -> Result<Vec<u8>, C2AuthTranscriptError> {
    validate_auth_capabilities("offer", &offer.capabilities)?;
    validate_auth_capabilities("selection", &selected.capabilities)?;

    let mut message = Vec::with_capacity(512);
    message.extend_from_slice(b"gate4agent-c2-control-auth-v2-compatibility\0");
    encode_bounded_str(&mut message, BUILD_STAMP);
    message.push(match direction { C2AuthDirection::Server => 1, C2AuthDirection::Client => 2 });
    message.extend_from_slice(client_nonce);
    message.extend_from_slice(server_nonce);

    message.extend_from_slice(b"offer\0");
    encode_bounded_str(&mut message, &offer.build_stamp);
    encode_capabilities(&mut message, &offer.capabilities);
    match offer.state_schema {
        Some(state_schema) => {
            message.push(1);
            encode_protocol_range(&mut message, state_schema.versions);
        }
        None => message.push(0),
    }

    message.extend_from_slice(b"selected\0");
    encode_bounded_str(&mut message, &selected.build_stamp);
    encode_capabilities(&mut message, &selected.capabilities);
    encode_bounded_str(&mut message, selected.host.operating_system.as_str());
    encode_bounded_str(&mut message, selected.host.architecture.as_str());
    message.push(match selected.path_semantics.style {
        PathStyle::Windows => 1,
        PathStyle::Posix => 2,
    });
    message.push(match selected.path_semantics.encoding {
        PathEncoding::Utf8 => 1,
        PathEncoding::UnixBytes => 2,
    });

    if message.len() > MAX_C2_BOUND_AUTH_TRANSCRIPT_BYTES {
        return Err(C2AuthTranscriptError::TooLong {
            len: message.len(),
            max: MAX_C2_BOUND_AUTH_TRANSCRIPT_BYTES,
        });
    }
    Ok(message)
}

fn validate_auth_capabilities(
    section: &'static str,
    capabilities: &[CapabilityId],
) -> Result<(), C2AuthTranscriptError> {
    if capabilities.len() > MAX_C2_AUTH_COMPATIBILITY_CAPABILITIES {
        return Err(C2AuthTranscriptError::TooManyCapabilities {
            section,
            count: capabilities.len(),
            max: MAX_C2_AUTH_COMPATIBILITY_CAPABILITIES,
        });
    }
    Ok(())
}

fn encode_protocol_range(message: &mut Vec<u8>, range: ProtocolRange) {
    message.extend_from_slice(&range.minimum().to_le_bytes());
    message.extend_from_slice(&range.maximum().to_le_bytes());
}

fn encode_capabilities(message: &mut Vec<u8>, capabilities: &[CapabilityId]) {
    message.extend_from_slice(&(capabilities.len() as u16).to_le_bytes());
    for capability in capabilities {
        encode_bounded_str(message, capability.as_str());
    }
}

fn encode_bounded_str(message: &mut Vec<u8>, value: &str) {
    debug_assert!(value.len() <= u16::MAX as usize);
    message.extend_from_slice(&(value.len() as u16).to_le_bytes());
    message.extend_from_slice(value.as_bytes());
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum NodeTransportState {
    Online,
    Offline,
    Parked,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum NodeFreshness {
    Fresh,
    Stale,
    Unavailable,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum GapKind {
    IncarnationChanged,
    HistoryEvicted,
    NonContiguousEvents,
    CursorRegression,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct NodeGap {
    pub kind: GapKind,
    pub detected_at_unix_ms: u64,
    pub previous: Option<NodeCursor>,
    pub observed: NodeCursor,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum C2ErrorCategory {
    Authentication,
    Identity,
    Protocol,
    Transport,
    Timeout,
    Internal,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SanitizedError {
    pub category: C2ErrorCategory,
    pub message: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SlimSession {
    pub instance_id: AgentInstanceId,
    pub generation: SessionGeneration,
    pub agent_id: String,
    pub transport: TransportKind,
    pub status: SlimSessionStatus,
    pub process_id: Option<u32>,
    pub terminal_size: Option<TerminalSize>,
    pub operation_pending: bool,
    pub input_pending: bool,
    /// The node's current screen classification for this session, carried
    /// on the inventory rather than only on terminal frames so a consumer
    /// deciding whether to hand this session work can read it without
    /// subscribing to frames for every session it owns. `#[serde(default)]`
    /// decodes an older node's silence as `Unknown`.
    #[serde(default)]
    pub screen_state: PtyScreenState,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum SlimSessionStatus {
    Registered,
    Starting,
    Running,
    Stopping,
    Exited,
    Failed,
}

impl From<&SessionStatus> for SlimSessionStatus {
    fn from(status: &SessionStatus) -> Self {
        match status {
            SessionStatus::Registered => Self::Registered,
            SessionStatus::Starting => Self::Starting,
            SessionStatus::Running => Self::Running,
            SessionStatus::Stopping => Self::Stopping,
            SessionStatus::Exited { .. } => Self::Exited,
            SessionStatus::Failed { .. } => Self::Failed,
        }
    }
}

impl From<&C2SessionStatus> for SlimSessionStatus {
    fn from(status: &C2SessionStatus) -> Self {
        match status {
            C2SessionStatus::Registered => Self::Registered,
            C2SessionStatus::Starting => Self::Starting,
            C2SessionStatus::Running => Self::Running,
            C2SessionStatus::Stopping => Self::Stopping,
            C2SessionStatus::Exited { .. } => Self::Exited,
            C2SessionStatus::Failed => Self::Failed,
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SlimWorkspace {
    pub workspace_id: WorkspaceId,
    pub canonical_root: String,
    pub canonical_root_truncated: bool,
    pub sessions: Vec<SlimSession>,
    pub session_count: usize,
    pub sessions_truncated: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub worktree_service_mode: Option<WorktreeServiceMode>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub managed_worktree_profiles: Option<WorktreeProfileInventory>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SlimManagedSessionRecord {
    pub record_id: SessionRecordId,
    pub display_name: String,
    pub display_name_truncated: bool,
    pub provider: AgentId,
    pub mode: SessionMode,
    pub state: ManagedSessionState,
    pub workspace_id: WorkspaceId,
    pub active_session: Option<SessionAddress>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bundle: Option<ResolvedBundleReceipt>,
    pub provider_identity_present: bool,
    pub updated_at_unix_ms: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SlimNodeInventory {
    pub node_id: NodeId,
    pub enabled_providers: Vec<AgentId>,
    #[serde(default, skip_serializing_if = "ProviderRuntimeStatuses::is_empty")]
    pub provider_runtime_statuses: ProviderRuntimeStatuses,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub provider_contracts: Vec<ProviderContractSupport>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub provider_adapter_contracts: Vec<ProviderAdapterContractSupport>,
    pub workspaces: BTreeMap<WorkspaceId, SlimWorkspace>,
    pub workspace_count: usize,
    pub workspaces_truncated: bool,
    pub session_count: usize,
    pub sessions_truncated: bool,
    #[serde(default)]
    pub managed_sessions: Vec<SlimManagedSessionRecord>,
    #[serde(default)]
    pub managed_session_count: usize,
    #[serde(default)]
    pub managed_sessions_truncated: bool,
    /// Lifetime count of managed session records the node's own retention
    /// sweep has retired (`NodeShared::retired_records_total`, gate4agent-
    /// node). Additive and informational only -- unlike `managed_session_
    /// count`/`managed_sessions_truncated` it is not cross-checked against
    /// `managed_sessions` (retired records are gone from that list by
    /// definition, not merely paged out of it). `0` on any snapshot source
    /// that does not yet carry the node's own counter through.
    #[serde(default, skip_serializing_if = "usize_is_zero")]
    pub retired_count: usize,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub managed_worktrees: Vec<ManagedWorktreeLeaseSnapshot>,
    #[serde(default, skip_serializing_if = "usize_is_zero")]
    pub managed_worktree_count: usize,
    #[serde(default, skip_serializing_if = "bool_is_false")]
    pub managed_worktrees_truncated: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub launch_inventory: Option<LaunchInventory>,
}

fn usize_is_zero(value: &usize) -> bool { *value == 0 }

fn bool_is_false(value: &bool) -> bool { !*value }

/// Orders `ManagedSessionRecord`/`C2ManagedSessionRecord` entries by liveness
/// before the `MAX_C2_MANAGED_SESSIONS_PER_NODE` page bound is applied, so a
/// page cut never evicts a live record behind stale ones that merely sort
/// earlier by id. `Live` sessions are the ones actively serving a run and
/// must never be cut. `IdentityPending` is a real, just-spawned process
/// still waiting on its identity observation -- exactly the window in which
/// the harness must still see the record on the page for credential
/// binding and context-source correlation -- so it ranks right behind
/// `Live`, ahead of `Dormant`. `Dormant` has no live process but can still
/// be resumed, so it ranks next. `Unavailable` is not usable at all and
/// ranks last, ordered by record id within each tier like before.
fn managed_session_liveness_rank(state: &ManagedSessionState) -> u8 {
    match state {
        ManagedSessionState::Live => 0,
        ManagedSessionState::IdentityPending => 1,
        ManagedSessionState::Dormant => 2,
        ManagedSessionState::Unavailable => 3,
    }
}

/// Same liveness-first rationale as `managed_session_liveness_rank`, applied
/// to `gate4agent_types::SessionStatus` PTY sessions ahead of the
/// `MAX_C2_SESSIONS_PER_NODE` page bound: `Running` sessions rank first,
/// `Starting`/`Stopping` (mid-transition, still attached to a live process)
/// next, not-yet-started `Registered` sessions next, and terminal
/// `Exited`/`Failed` sessions rank last.
fn session_liveness_rank(status: &SessionStatus) -> u8 {
    match status {
        SessionStatus::Running => 0,
        SessionStatus::Starting | SessionStatus::Stopping => 1,
        SessionStatus::Registered => 2,
        SessionStatus::Exited { .. } | SessionStatus::Failed { .. } => 3,
    }
}

/// Same as `session_liveness_rank`, for the C2-side `C2SessionStatus` mirror.
fn c2_session_liveness_rank(status: &C2SessionStatus) -> u8 {
    match status {
        C2SessionStatus::Running => 0,
        C2SessionStatus::Starting | C2SessionStatus::Stopping => 1,
        C2SessionStatus::Registered => 2,
        C2SessionStatus::Exited { .. } | C2SessionStatus::Failed => 3,
    }
}

impl SlimNodeInventory {
    pub fn from_snapshot(snapshot: &NodeSnapshot) -> Self {
        let mut providers = snapshot.enabled_providers.clone();
        providers.sort();
        providers.dedup();
        let workspace_count = snapshot.workspaces.len();
        let session_count = snapshot.workspaces.iter().map(|workspace| workspace.sessions.len()).sum();
        let mut remaining_sessions = MAX_C2_SESSIONS_PER_NODE;
        let mut workspaces = BTreeMap::new();
        let mut ordered = snapshot.workspaces.iter().collect::<Vec<_>>();
        ordered.sort_by(|left, right| left.workspace_id.cmp(&right.workspace_id));
        for workspace in ordered.into_iter().take(MAX_C2_WORKSPACES_PER_NODE) {
            let mut sessions = workspace.sessions.iter().collect::<Vec<_>>();
            sessions.sort_by(|left, right| {
                session_liveness_rank(&left.status)
                    .cmp(&session_liveness_rank(&right.status))
                    .then_with(|| left.instance_id.cmp(&right.instance_id))
                    .then_with(|| left.generation.cmp(&right.generation))
            });
            let take = remaining_sessions.min(sessions.len());
            let slim_sessions = sessions.into_iter().take(take).map(|session| SlimSession {
                instance_id: session.instance_id,
                generation: session.generation,
                agent_id: session.agent_id.as_str().to_owned(),
                transport: session.transport,
                status: SlimSessionStatus::from(&session.status),
                process_id: session.process_id,
                terminal_size: session.terminal_size,
                operation_pending: session.pending_operation.is_some(),
                input_pending: session.pending_input.is_some(),
                screen_state: session.screen_state.clone(),
            }).collect();
            remaining_sessions -= take;
            let display_root = sanitize_host_path_display(&workspace.canonical_root);
            let (canonical_root, canonical_root_truncated) =
                truncate_utf8(&display_root, MAX_C2_ROOT_BYTES);
            workspaces.insert(workspace.workspace_id.clone(), SlimWorkspace {
                workspace_id: workspace.workspace_id.clone(),
                canonical_root,
                canonical_root_truncated,
                sessions: slim_sessions,
                session_count: workspace.sessions.len(),
                sessions_truncated: workspace.sessions.len() > take,
                worktree_service_mode: workspace.worktree_service_mode,
                managed_worktree_profiles: workspace.managed_worktree_profiles.clone(),
            });
        }
        let included_session_count = workspaces
            .values()
            .map(|workspace| workspace.sessions.len())
            .sum::<usize>();
        let managed_session_count = snapshot.session_records.len();
        let mut ordered_records = snapshot.session_records.iter().collect::<Vec<_>>();
        ordered_records.sort_by(|left, right| {
            managed_session_liveness_rank(&left.state)
                .cmp(&managed_session_liveness_rank(&right.state))
                .then_with(|| left.record_id.cmp(&right.record_id))
        });
        let managed_sessions = ordered_records
            .into_iter()
            .take(MAX_C2_MANAGED_SESSIONS_PER_NODE)
            .map(SlimManagedSessionRecord::from)
            .collect::<Vec<_>>();
        let managed_worktree_count = snapshot.managed_worktrees.len();
        let mut managed_worktrees = snapshot.managed_worktrees.clone();
        managed_worktrees.sort_by(|left, right| left.lease_id.cmp(&right.lease_id));
        managed_worktrees.truncate(MAX_C2_MANAGED_WORKTREES_PER_NODE);
        Self {
            node_id: snapshot.node_id.clone(),
            enabled_providers: providers,
            provider_runtime_statuses: snapshot.provider_runtime_statuses.clone(),
            provider_contracts: Vec::new(),
            provider_adapter_contracts: Vec::new(),
            workspaces,
            workspace_count,
            workspaces_truncated: workspace_count > MAX_C2_WORKSPACES_PER_NODE,
            session_count,
            sessions_truncated: included_session_count < session_count,
            managed_sessions_truncated: managed_sessions.len() < managed_session_count,
            managed_sessions,
            managed_session_count,
            // Neither `NodeSnapshot` nor `C2NodeSnapshot` carries the
            // node's own `retired_records_total` atomic through yet -- see
            // this field's own doc comment. `0` here is the honest value
            // for what this projection can see today, not a placeholder
            // masking a real number.
            retired_count: 0,
            managed_worktrees_truncated:
                managed_worktrees.len() < managed_worktree_count,
            managed_worktrees,
            managed_worktree_count,
            launch_inventory: snapshot.launch_inventory.clone(),
        }
    }

    pub fn from_c2_snapshot(snapshot: &C2NodeSnapshot) -> Self {
        let mut providers = snapshot.enabled_providers.clone();
        providers.sort();
        providers.dedup();
        let workspace_count = snapshot.workspaces.len();
        let session_count = snapshot.workspaces.iter().map(|workspace| workspace.sessions.len()).sum();
        let mut remaining_sessions = MAX_C2_SESSIONS_PER_NODE;
        let mut workspaces = BTreeMap::new();
        let mut ordered = snapshot.workspaces.iter().collect::<Vec<_>>();
        ordered.sort_by(|left, right| left.workspace_id.cmp(&right.workspace_id));
        for workspace in ordered.into_iter().take(MAX_C2_WORKSPACES_PER_NODE) {
            let mut sessions = workspace.sessions.iter().collect::<Vec<_>>();
            sessions.sort_by(|left, right| {
                c2_session_liveness_rank(&left.status)
                    .cmp(&c2_session_liveness_rank(&right.status))
                    .then_with(|| left.instance_id.cmp(&right.instance_id))
                    .then_with(|| left.generation.cmp(&right.generation))
            });
            let take = remaining_sessions.min(sessions.len());
            let slim_sessions = sessions.into_iter().take(take).map(|session| SlimSession {
                instance_id: session.instance_id,
                generation: session.generation,
                agent_id: session.agent_id.as_str().to_owned(),
                transport: session.transport,
                status: SlimSessionStatus::from(&session.status),
                process_id: session.process_id,
                terminal_size: session.terminal_size,
                operation_pending: session.pending_operation.is_some(),
                input_pending: session.pending_input.is_some(),
                screen_state: session.screen_state.clone(),
            }).collect();
            remaining_sessions -= take;
            let display_root = sanitize_host_path_display(&workspace.canonical_root);
            let (canonical_root, canonical_root_truncated) =
                truncate_utf8(&display_root, MAX_C2_ROOT_BYTES);
            workspaces.insert(workspace.workspace_id.clone(), SlimWorkspace {
                workspace_id: workspace.workspace_id.clone(),
                canonical_root,
                canonical_root_truncated,
                sessions: slim_sessions,
                session_count: workspace.sessions.len(),
                sessions_truncated: workspace.sessions.len() > take,
                worktree_service_mode: workspace.worktree_service_mode,
                managed_worktree_profiles: workspace.managed_worktree_profiles.clone(),
            });
        }
        let included_session_count = workspaces.values()
            .map(|workspace| workspace.sessions.len())
            .sum::<usize>();
        let managed_session_count = snapshot.session_records.len();
        let mut ordered_records = snapshot.session_records.iter().collect::<Vec<_>>();
        ordered_records.sort_by(|left, right| {
            managed_session_liveness_rank(&left.state)
                .cmp(&managed_session_liveness_rank(&right.state))
                .then_with(|| left.record_id.cmp(&right.record_id))
        });
        let managed_sessions = ordered_records.into_iter()
            .take(MAX_C2_MANAGED_SESSIONS_PER_NODE)
            .map(SlimManagedSessionRecord::from)
            .collect::<Vec<_>>();
        let managed_worktree_count = snapshot.managed_worktrees.len();
        let mut managed_worktrees = snapshot.managed_worktrees.clone();
        managed_worktrees.sort_by(|left, right| left.lease_id.cmp(&right.lease_id));
        managed_worktrees.truncate(MAX_C2_MANAGED_WORKTREES_PER_NODE);
        Self {
            node_id: snapshot.node_id.clone(),
            enabled_providers: providers,
            provider_runtime_statuses: snapshot.provider_runtime_statuses.clone(),
            provider_contracts: Vec::new(),
            provider_adapter_contracts: Vec::new(),
            workspaces,
            workspace_count,
            workspaces_truncated: workspace_count > MAX_C2_WORKSPACES_PER_NODE,
            session_count,
            sessions_truncated: included_session_count < session_count,
            managed_sessions_truncated: managed_sessions.len() < managed_session_count,
            managed_sessions,
            managed_session_count,
            // Neither `NodeSnapshot` nor `C2NodeSnapshot` carries the
            // node's own `retired_records_total` atomic through yet -- see
            // this field's own doc comment. `0` here is the honest value
            // for what this projection can see today, not a placeholder
            // masking a real number.
            retired_count: 0,
            managed_worktrees_truncated:
                managed_worktrees.len() < managed_worktree_count,
            managed_worktrees,
            managed_worktree_count,
            launch_inventory: snapshot.launch_inventory.clone(),
        }
    }
}

impl From<&ManagedSessionRecord> for SlimManagedSessionRecord {
    fn from(record: &ManagedSessionRecord) -> Self {
        let (display_name, display_name_truncated) =
            truncate_utf8(&record.display_name, MAX_C2_SESSION_DISPLAY_NAME_BYTES);
        Self {
            // Legacy Slim inventory deliberately strips context_id and context receipts.
            record_id: record.record_id.clone(),
            display_name,
            display_name_truncated,
            provider: record.provider.clone(),
            mode: record.mode,
            state: record.state,
            workspace_id: record.workspace_id.clone(),
            active_session: record.active_session.clone(),
            environment_profile: record.environment_profile.clone(),
            bundle: record.bundle.clone(),
            provider_identity_present: record.provider_session.is_some(),
            updated_at_unix_ms: record.updated_at_unix_ms,
        }
    }
}

impl From<&C2ManagedSessionRecord> for SlimManagedSessionRecord {
    fn from(record: &C2ManagedSessionRecord) -> Self {
        let (display_name, display_name_truncated) =
            truncate_utf8(&record.display_name, MAX_C2_SESSION_DISPLAY_NAME_BYTES);
        Self {
            // Legacy Slim inventory deliberately strips context_id and context receipts.
            record_id: record.record_id.clone(),
            display_name,
            display_name_truncated,
            provider: record.provider.clone(),
            mode: record.mode,
            state: record.state,
            workspace_id: record.workspace_id.clone(),
            active_session: record.active_session.clone(),
            environment_profile: record.environment_profile.clone(),
            bundle: record.bundle.clone(),
            provider_identity_present: record.provider_identity_present,
            updated_at_unix_ms: record.updated_at_unix_ms,
        }
    }
}

fn truncate_utf8(value: &str, max_bytes: usize) -> (String, bool) {
    if value.len() <= max_bytes {
        return (value.to_owned(), false);
    }
    let mut end = max_bytes;
    while !value.is_char_boundary(end) {
        end -= 1;
    }
    (value[..end].to_owned(), true)
}

fn sanitize_host_path_display(path: &OpaqueHostPath) -> String {
    let mut sanitized = String::new();
    for ch in path.display_text().chars() {
        if ch.is_control() {
            sanitized.extend(ch.escape_default());
        } else {
            sanitized.push(ch);
        }
    }
    sanitized
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ObservedNode {
    pub endpoint: String,
    pub transport_label: String,
    pub transport: NodeTransportState,
    pub freshness: NodeFreshness,
    pub cursor: Option<NodeCursor>,
    pub inventory: Option<SlimNodeInventory>,
    pub last_attempt_unix_ms: Option<u64>,
    pub last_success_unix_ms: Option<u64>,
    pub consecutive_failures: u32,
    pub last_error: Option<SanitizedError>,
    pub gaps: Vec<NodeGap>,
    pub gaps_truncated: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct HealthResponse {
    pub ok: bool,
    pub service: String,
    pub api_version: u16,
    pub pid: u32,
    pub version: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ReadyResponse {
    pub ready: bool,
    pub api_version: u16,
    pub configured_nodes: usize,
    pub attempted_nodes: usize,
    pub online_nodes: usize,
    pub offline_nodes: usize,
    pub parked_nodes: usize,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct StatusResponse {
    pub api_version: u16,
    pub ready: bool,
    pub observed_at_unix_ms: u64,
    pub nodes: BTreeMap<NodeId, ObservedNode>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use gate4agent_node_protocol::{
        GitSnapshot, GitStatusEntry, GitWorktreeSnapshot, NodeFailureCode, NodeSnapshot,
        WorkspaceEntry, WorkspaceEntryKind, WorkspaceInspection, WorkspaceSnapshot,
    };
    use gate4agent_types::{
        AdapterBinding, AdapterFamily, AdapterId, AdapterVerification, AgentId, AgentInstanceId,
        CapabilitySnapshot, ControlEvent, ControlEventKind, ForegroundSnapshot, HistorySnapshot,
        OperationId, PreparedInputKind, ProviderEvent, ProviderInteractionOutcome,
        ProviderSessionIdentity, ProviderSessionKey, ProviderSnapshot, ProviderSource,
        PtyScreenState, ResumeSessionSummary, ResumeSnapshot, SessionGeneration, SessionSnapshot,
        SessionStatus, TerminalSize, TransportKind,
    };

    #[test]
    fn c2_harness_mcp_projection_is_value_exact_and_sanitized() {
        let reservation_id = HarnessMcpReservationId::new(format!(
            "hmcpres_{}", "a".repeat(24),
        )).unwrap();
        let activation_digest = HarnessMcpActivationDigest::new(format!(
            "sha256:{}", "b".repeat(64),
        )).unwrap();
        let call_id = HarnessMcpCallId::new(format!("hmcpcall_{}", "c".repeat(24))).unwrap();
        let session = SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(7),
                generation: SessionGeneration(1),
            },
        };
        let source = NodeEvent::HarnessMcpReadCall {
            reservation_id: reservation_id.clone(),
            activation_digest: activation_digest.clone(),
            record_id: SessionRecordId::new("record-a").unwrap(),
            session: session.clone(),
            call_id: call_id.clone(),
            request: HarnessMcpOpaquePayloadV1 {
                content_type: HarnessMcpContentTypeV1::HarnessReadRequestJsonV1,
                body: br#"{"kind":"context-get"}"#.to_vec(),
            },
            deadline_unix_ms: 4_000,
        };
        let projected = C2NodeEvent::from(&source);
        assert!(projected.requires_harness_mcp_proxy_capability());
        assert!(projected.harness_mcp_contract_is_valid_at(1_000));
        let json = serde_json::to_string(&projected).unwrap();
        assert!(!json.contains("g4ah3_"));
        assert!(!json.contains("endpoint"));
        assert!(!json.contains("path"));
        assert_eq!(serde_json::from_str::<C2NodeEvent>(&json).unwrap(), projected);

        let response = NodeResponse::ReplyChunkAccepted {
            reservation_id,
            activation_digest,
            record_id: SessionRecordId::new("record-a").unwrap(),
            session,
            call_id,
            next_offset: 2,
            completed: true,
        };
        let projected_response = C2NodeResponse::from(&response);
        assert!(projected_response.requires_harness_mcp_proxy_capability());
        assert_eq!(
            serde_json::from_value::<C2NodeResponse>(
                serde_json::to_value(&projected_response).unwrap(),
            ).unwrap(),
            projected_response,
        );

        let failure = C2NodeFailure::from(&NodeFailure {
            code: NodeFailureCode::ChunkOutOfOrder,
            message: "secret raw detail".to_owned(),
        });
        assert_eq!(failure.message, "harness MCP reply chunk out of order");
        assert!(!failure.message.contains("secret"));
        assert!(failure.requires_harness_mcp_proxy_capability());
        assert_eq!(C2_HARNESS_MCP_READ_PROXY_CAPABILITY, "harness-mcp-read-proxy-v1");
    }

    fn host_path(value: impl Into<String>) -> OpaqueHostPath {
        OpaqueHostPath::utf8(value.into()).unwrap()
    }

    fn repository_path(value: impl Into<String>) -> RepositoryPath {
        RepositoryPath::utf8(value.into()).unwrap()
    }

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

    fn fixture_session() -> SessionSnapshot {
        SessionSnapshot {
            instance_id: AgentInstanceId(7),
            agent_id: AgentId::new("codex").unwrap(),
            transport: TransportKind::Pty,
            generation: SessionGeneration(2),
            status: SessionStatus::Running,
            pending_operation: Some(OperationId(9)),
            pending_input: Some(PreparedInputKind::TerminalText),
            process_id: Some(1234),
            terminal_size: Some(TerminalSize { rows: 40, columns: 120 }),
            terminal_frame: None,
            terminal_stale: None,
            session_options: None,
            capabilities: CapabilitySnapshot::default(),
            history: HistorySnapshot::default(),
            resume: ResumeSnapshot::default(),
            foreground: ForegroundSnapshot::default(),
            provider: ProviderSnapshot::default(),
            screen_state: PtyScreenState::default(),
        }
    }

    fn fixture_agent_progress() -> SessionAgentProgress {
        SessionAgentProgress {
            address: SessionAddress {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                session: gate4agent_node_protocol::SessionKey {
                    instance_id: AgentInstanceId(7),
                    generation: SessionGeneration(2),
                },
            },
            progress: AgentProgressV1 {
                provider_sequence: 19,
                activity: ProviderActivity::Working,
                completed_turns: 3,
                usage: Some(AgentProgressUsageV1 {
                    input_tokens: 101,
                    output_tokens: 37,
                    cache_read_tokens: 11,
                    cache_write_tokens: 5,
                    reasoning_tokens: 13,
                }),
                current: AgentProgressCurrentV1::Working,
                // Capitalized because `validate_agent_progress_tool_label`
                // accepts a fixed capitalized class vocabulary. Lowercase
                // decodes as a label outside it, so this fixture stood for
                // an invalid value rather than the safe class it meant.
                active_tool_labels: vec!["Shell".to_owned()],
                active_tool_count: 1,
                attention: None,
                subagent_count: 2,
                last_event_kind: Some(AgentProgressEventKindV1::ToolStarted),
                gap_count: 0,
                stale: false,
                truncated: false,
            },
        }
    }

    #[test]
    fn c2_agent_progress_snapshot_roundtrip_is_exact_and_private() {
        assert_eq!(
            C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY,
            "agent-progress-snapshot-v1",
        );
        assert_eq!(
            C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY,
            gate4agent_node_protocol::NODE_AGENT_PROGRESS_SNAPSHOT_CAPABILITY,
        );
        let progress = fixture_agent_progress();
        let snapshot = C2NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            agent_progress: vec![progress.clone()],
            managed_worktrees: Vec::new(),
            launch_inventory: None,
        };
        let json = serde_json::to_string(&snapshot).unwrap();
        let decoded = serde_json::from_str::<C2NodeSnapshot>(&json).unwrap();
        assert_eq!(decoded.agent_progress, vec![progress]);
        for forbidden in [
            "prompt", "arguments", "contents", "transcript", "provider_session",
            "canonical_root", "executable",
        ] {
            assert!(!json.contains(forbidden), "agent progress leaked {forbidden}");
        }

        let mut value = serde_json::to_value(snapshot).unwrap();
        value["agent_progress"][0]["progress"]["prompt"] =
            serde_json::json!("private prompt");
        assert!(serde_json::from_value::<C2NodeSnapshot>(value).is_err());

        let empty = serde_json::json!({
            "node_id": "node-a",
            "enabled_providers": [],
            "workspaces": [],
            "session_records": [],
            "managed_worktrees": [],
            "launch_inventory": null
        });
        let empty = serde_json::from_value::<C2NodeSnapshot>(empty).unwrap();
        assert!(empty.agent_progress.is_empty());
        assert!(serde_json::to_value(empty).unwrap().get("agent_progress").is_none());
    }

    #[test]
    fn c2_workspace_event_does_not_synthesize_agent_progress() {
        let event = C2NodeEvent::from(&NodeEvent::WorkspaceAdded {
            workspace: WorkspaceSnapshot {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                canonical_root: host_path(r"C:\private\workspace"),
                sessions: vec![fixture_session()],
                worktree_service_mode: None,
                managed_worktree_profiles: None,
            },
        });
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("agent_progress"));
        assert!(!json.contains("provider_sequence"));
    }

    fn private_session_record() -> ManagedSessionRecord {
        ManagedSessionRecord {
            record_id: SessionRecordId::new("session-private").unwrap(),
            display_name: "release shepherd".to_owned(),
            provider: provider("codex"),
            mode: SessionMode::Pty,
            state: ManagedSessionState::Live,
            workspace_id: WorkspaceId::new("primary").unwrap(),
            canonical_root: host_path(r"C:\private\canonical-root"),
            provider_session: Some(ProviderSessionIdentity {
                key: ProviderSessionKey::SessionId,
                id: "private-provider-session-id".to_owned(),
                transcript_path: Some(r"C:\private\transcript-secret.jsonl".to_owned()),
            }),
            active_session: Some(SessionAddress {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                session: gate4agent_node_protocol::SessionKey {
                    instance_id: AgentInstanceId(41),
                    generation: SessionGeneration(3),
                },
            }),
            environment_profile: Some(ResolvedEnvironmentProfileReceipt {
                profile_id: SpawnEnvironmentProfileId::new("local-default").unwrap(),
                profile_revision: SpawnEnvironmentProfileRevision::new(
                    "local-default.2026-08",
                )
                .unwrap(),
                network_allowlist: None,
                browser_profile_id: None,
            }),
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            created_at_unix_ms: 10,
            updated_at_unix_ms: 20,
            last_error: Some("private-error-with-secret-token".to_owned()),
        }
    }

    fn context_pack_receipt() -> ResolvedContextPackReceipt {
        ResolvedContextPackReceipt {
            id: SpawnContextId::new("context-review-7").unwrap(),
            digest: SpawnContextDigest::new(format!("sha256:{}", "c".repeat(64))).unwrap(),
            lineage: ContextPackLineageReceipt {
                source_node_id: NodeId::new("node-source").unwrap(),
                source_session: SessionAddress {
                    workspace_id: WorkspaceId::new("source-workspace").unwrap(),
                    session: gate4agent_node_protocol::SessionKey {
                        instance_id: AgentInstanceId(17),
                        generation: SessionGeneration(4),
                    },
                },
                source_provider: provider("codex"),
            },
            source_message_count: 9,
            retained_message_count: 7,
            byte_len: 4096,
            truncated: true,
        }
    }

    fn assert_private_record_fields_absent(json: &str) {
        assert!(!json.contains("canonical_root"));
        assert!(!json.contains("provider_session"));
        assert!(!json.contains("private-provider-session-id"));
        assert!(!json.contains("transcript_path"));
        assert!(!json.contains("transcript-secret.jsonl"));
        assert!(!json.contains("last_error"));
        assert!(!json.contains("private-error-with-secret-token"));
    }

    #[test]
    fn c2_task_binding_projection_is_exact_safe_and_legacy_slim_omits_it() {
        let mut source = private_session_record();
        source.task_binding = Some(SessionTaskBindingV1 {
            revision: 3,
            task_id: Some(TaskId::from_nonce([3; 12])),
            changed_at_unix_ms: 15,
        });
        let projected = C2ManagedSessionRecord::from(&source);
        assert_eq!(projected.task_binding, source.task_binding);
        assert!(projected.task_binding_is_valid());
        let slim = serde_json::to_value(SlimManagedSessionRecord::from(&projected)).unwrap();
        assert!(slim.get("task_binding").is_none());

        source.task_binding.as_mut().unwrap().changed_at_unix_ms = 21;
        assert!(C2ManagedSessionRecord::from(&source).task_binding.is_none());

        let mut invalid = serde_json::to_value(projected).unwrap();
        invalid["task_binding"]["changed_at_unix_ms"] = serde_json::json!(9);
        assert!(serde_json::from_value::<C2ManagedSessionRecord>(invalid).is_err());
    }

    fn routed_response(response: NodeResponse) -> RoutedNodeResponse {
        RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::from(&response)),
        }
    }

    fn control_node_event(event: ControlEventKind) -> NodeEvent {
        NodeEvent::Control {
            address: SessionAddress {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                session: gate4agent_node_protocol::SessionKey {
                    instance_id: AgentInstanceId(41),
                    generation: SessionGeneration(3),
                },
            },
            event: ControlEvent {
                sequence: 12,
                command_id: None,
                instance_id: AgentInstanceId(41),
                generation: SessionGeneration(3),
                event,
            },
        }
    }

    #[test]
    fn history_summary_event_projects_exactly_and_carries_counts_only() {
        let summary = gate4agent_node_protocol::SessionHistorySummaryV1 {
            message_count: 12,
            message_count_exact: true,
            completed_turn_count: Some(5),
            total_tokens: Some(4_000),
            modified_at_unix_ms: Some(1_786_000_000_000),
        };
        let source = NodeEvent::SessionRecordHistorySummarized {
            record_id: SessionRecordId::new("record-a").unwrap(),
            summary,
        };
        let projected = C2NodeEvent::from(&source);
        assert_eq!(
            projected,
            C2NodeEvent::SessionRecordHistorySummarized {
                record_id: SessionRecordId::new("record-a").unwrap(),
                summary,
            },
        );
        assert_eq!(projected, C2NodeEvent::from_node_event_with_control_detail(&source));
        let json = serde_json::to_string(&projected).unwrap();
        assert_eq!(
            json,
            r#"{"kind":"session-record-history-summarized","record_id":"record-a","summary":{"message_count":12,"message_count_exact":true,"completed_turn_count":5,"total_tokens":4000,"modified_at_unix_ms":1786000000000}}"#,
        );
        assert_eq!(serde_json::from_str::<C2NodeEvent>(&json).unwrap(), projected);
    }

    fn tool_started_event() -> ControlEventKind {
        ControlEventKind::ProviderEvent {
            sequence: 19,
            source: provider_source(),
            source_sequence: 7,
            event: ProviderEvent::ToolStarted {
                id: "toolu_1".to_owned(),
                name: "Bash".to_owned(),
                input_json: r#"{"command":"cat private.txt"}"#.to_owned(),
                agent_id: None,
            },
        }
    }

    #[test]
    fn control_detail_is_absent_by_default_and_sanitized_when_requested() {
        let source = control_node_event(tool_started_event());

        let default = C2NodeEvent::from(&source);
        assert!(!default.requires_control_detail_capability());
        assert!(!serde_json::to_string(&default).unwrap().contains("detail"));

        let detailed = C2NodeEvent::from_node_event_with_control_detail(&source);
        assert!(detailed.requires_control_detail_capability());
        let json = serde_json::to_string(&detailed).unwrap();
        assert!(json.contains("toolu_1"));
        assert!(json.contains("Bash"));
        assert!(!json.contains("private.txt"));
        assert_eq!(serde_json::from_str::<C2NodeEvent>(&json).unwrap(), detailed);

        let mut stripped = detailed;
        stripped.strip_control_detail();
        assert_eq!(stripped, default);
    }

    #[test]
    fn content_only_control_events_carry_their_kind_but_never_their_content() {
        let source = control_node_event(ControlEventKind::ProviderEvent {
            sequence: 20,
            source: provider_source(),
            source_sequence: 8,
            event: ProviderEvent::Text { text: "private words".to_owned(), is_delta: true },
        });
        let detailed = C2NodeEvent::from_node_event_with_control_detail(&source);
        assert!(detailed.requires_control_detail_capability());
        assert!(!serde_json::to_string(&detailed).unwrap().contains("private words"));
    }

    #[test]
    fn control_plane_bookkeeping_has_no_detail_even_when_requested() {
        let source = control_node_event(ControlEventKind::Registered);
        let detailed = C2NodeEvent::from_node_event_with_control_detail(&source);
        assert!(!detailed.requires_control_detail_capability());
        assert_eq!(detailed, C2NodeEvent::from(&source));
    }

    #[test]
    fn resync_control_detail_follows_the_same_rule_and_strips_to_the_default() {
        let events = vec![
            gate4agent_node_protocol::NodeEventEnvelope {
                sequence: 3,
                event: control_node_event(tool_started_event()),
            },
            gate4agent_node_protocol::NodeEventEnvelope {
                sequence: 4,
                event: control_node_event(ControlEventKind::Removed),
            },
        ];
        let snapshot = NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        };
        let source = NodeResponse::Resync {
            event_sequence: 4,
            oldest_available_sequence: 1,
            snapshot,
            events,
        };
        let default = C2NodeResponse::from(&source);
        assert!(!default.requires_control_detail_capability());
        let mut detailed = C2NodeResponse::from_node_response_with_control_detail(&source);
        assert!(detailed.requires_control_detail_capability());
        detailed.strip_control_detail();
        assert_eq!(detailed, default);
    }

    fn routed_control_event(event: ControlEventKind) -> RoutedNodeEvent {
        let address = SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(41),
                generation: SessionGeneration(3),
            },
        };
        RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([8; 16]),
                sequence: 9,
            },
            event: C2NodeEvent::from(&NodeEvent::Control {
                address,
                event: ControlEvent {
                    sequence: 12,
                    command_id: None,
                    instance_id: AgentInstanceId(41),
                    generation: SessionGeneration(3),
                    event,
                },
            }),
        }
    }

    fn provider_source() -> ProviderSource {
        ProviderSource {
            family: AdapterFamily::PtySemantic,
            binding: AdapterBinding::new(
                AdapterId::new("codex").unwrap(),
                "fixture/v1",
                AdapterVerification::SyntheticFixture,
            ).unwrap(),
        }
    }

    fn provider_contract_manifest() -> (
        Vec<ProviderContractSupport>,
        Vec<ProviderAdapterContractSupport>,
    ) {
        (
            vec![ProviderContractSupport {
                provider: provider("codex"),
                revision: ProviderContractRevision::new("codex.2026-08").unwrap(),
            }],
            vec![ProviderAdapterContractSupport {
                provider: provider("codex"),
                family: AdapterFamily::PtySemantic,
                adapter_id: AdapterId::new("codex").unwrap(),
                revision: AdapterContractRevision::new("pty-semantic.2026-08").unwrap(),
            }],
        )
    }

    fn c2_compatibility_support(
        capabilities: Vec<CapabilityId>,
    ) -> C2ControlCompatibilitySupport {
        C2ControlCompatibilitySupport {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities,
            host: HostDescriptor {
                operating_system: OperatingSystemId::new("darwin").unwrap(),
                architecture: ArchitectureId::new("aarch64").unwrap(),
            },
            path_semantics: PathSemantics {
                style: PathStyle::Posix,
                encoding: PathEncoding::Utf8,
            },
        }
    }

    /// Pins the SHAPE of the legacy hello -- which keys, in what order,
    /// and that `compatibility` is absent rather than serialized as
    /// `null` -- against a decoder that predates negotiation and would
    /// reject an unknown key.
    ///
    /// The stamp is interpolated from `BUILD_STAMP` rather than frozen,
    /// because freezing it could not survive its own design:
    /// `C2ClientHello::new` fills that field FROM the constant, and the
    /// constant is recomputed from the working tree on every build, so a
    /// frozen copy would go red on the very next unrelated source edit
    /// anywhere in the tree. The key set and their order are what must not
    /// drift, and those are still pinned literally.
    #[test]
    fn the_legacy_client_hello_json_carries_exactly_these_keys_in_this_order() {
        let hello = C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]);
        let json = serde_json::to_string(&hello).unwrap();
        let owned = format!(
            concat!(
                r#"{{"build_stamp":"{}","client_nonce":["#,
                "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,",
                "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]}}"
            ),
            BUILD_STAMP,
        );
        let expected = owned.as_str();

        assert_eq!(json, expected);
        // Stated separately so it cannot read as incidental to the string
        // above: to a strict decoder an absent key and a null one are
        // different messages.
        assert!(!json.contains("compatibility"));
        assert_eq!(
            serde_json::from_str::<C2ClientHello>(expected).unwrap(),
            hello,
        );
    }

    /// `BUILD_STAMP` as the bound-auth transcript encodes it: a
    /// little-endian u16 byte-length prefix followed by its ASCII bytes,
    /// rendered as hex.
    ///
    /// The transcript tests below still pin their entire byte layout
    /// literally -- domain tag, direction byte, both nonces, the offer and
    /// selection blocks with their length prefixes, the host descriptor --
    /// and interpolate only this. That split is deliberate: reorder a
    /// field, drop a length prefix, or change an encoding and they still
    /// fail; edit any file in the tree and they do not, because the
    /// transcript derives its stamp from the same constant the expectation
    /// does. Freezing the stamp alongside the layout is what would leave
    /// these permanently red.
    fn build_stamp_hex() -> String {
        let mut bytes = Vec::with_capacity(2 + BUILD_STAMP.len());
        bytes.extend_from_slice(&(BUILD_STAMP.len() as u16).to_le_bytes());
        bytes.extend_from_slice(BUILD_STAMP.as_bytes());
        bytes.iter().map(|byte| format!("{byte:02x}")).collect()
    }

    #[test]
    fn c2_compatibility_legacy_server_json_is_byte_equivalent() {
        #[derive(Serialize)]
        struct LegacyChallenge {
            build_stamp: String,
            server_nonce: [u8; C2_AUTH_NONCE_BYTES],
            server_proof: [u8; C2_AUTH_PROOF_BYTES],
        }

        #[derive(Serialize)]
        struct LegacyHello<'a> {
            build_stamp: String,
            connection_id: u64,
            status: &'a StatusResponse,
        }

        let challenge = C2ServerChallenge {
            build_stamp: BUILD_STAMP.to_owned(),
            server_nonce: [1; C2_AUTH_NONCE_BYTES],
            server_proof: [2; C2_AUTH_PROOF_BYTES],
            compatibility: None,
        };
        let legacy_challenge = LegacyChallenge {
            build_stamp: BUILD_STAMP.to_owned(),
            server_nonce: [1; C2_AUTH_NONCE_BYTES],
            server_proof: [2; C2_AUTH_PROOF_BYTES],
        };
        assert_eq!(
            serde_json::to_vec(&challenge).unwrap(),
            serde_json::to_vec(&legacy_challenge).unwrap(),
        );

        let status = StatusResponse {
            api_version: C2_API_VERSION,
            ready: true,
            observed_at_unix_ms: 7,
            nodes: BTreeMap::new(),
        };
        let hello = C2Hello {
            build_stamp: BUILD_STAMP.to_owned(),
            connection_id: 11,
            status: status.clone(),
            compatibility: None,
        };
        let legacy_hello = LegacyHello {
            build_stamp: BUILD_STAMP.to_owned(),
            connection_id: 11,
            status: &status,
        };
        assert_eq!(
            serde_json::to_vec(&hello).unwrap(),
            serde_json::to_vec(&legacy_hello).unwrap(),
        );
    }

    #[test]
    fn c2_terminal_frame_event_projection_wire_contract_is_exact() {
        let source = NodeEvent::TerminalFrame {
            address: SessionAddress {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                session: gate4agent_node_protocol::SessionKey {
                    instance_id: AgentInstanceId(7),
                    generation: SessionGeneration(3),
                },
            },
            frame: TerminalFrame {
                sequence: 11,
                size: TerminalSize { rows: 24, columns: 80 },
                cursor_row: 2,
                cursor_column: 4,
                contents: "ready".to_owned(),
                formatted: b"ready".to_vec(),
                scrollback_formatted: vec![b"previous".to_vec()],
                alternate_screen: false,
                mouse_protocol_enabled: false,
                mouse_protocol_encoding:
                    gate4agent_types::TerminalMouseProtocolEncoding::Default,
                produced_at_unix_ms: 0,
                screen_state: gate4agent_types::PtyScreenState::default(),
                bracketed_paste: None,
            },
        };
        let event = C2NodeEvent::from(&source);
        let json = serde_json::to_string(&event).unwrap();
        assert_eq!(
            json,
            r#"{"kind":"terminal-frame","address":{"workspace_id":"primary","session":{"instance_id":7,"generation":3}},"frame":{"sequence":11,"size":{"rows":24,"columns":80},"cursor_row":2,"cursor_column":4,"contents":"ready","formatted":[114,101,97,100,121],"scrollback_formatted":[[112,114,101,118,105,111,117,115]],"alternate_screen":false,"mouse_protocol_enabled":false,"mouse_protocol_encoding":"default","produced_at_unix_ms":0,"screen_state":{"kind":"unknown"}}}"#,
        );
        assert_eq!(serde_json::from_str::<C2NodeEvent>(&json).unwrap(), event);
    }

    /// `C2NodeEvent::AgentStream` passes `gate4agent_node_protocol::
    /// AgentStreamChunkV1` straight through -- no re-typed mirror of
    /// `AgentStreamChunkKindV1` in this crate -- so the new `Blocked`
    /// variant round-trips across the c2 wire unchanged, the same way
    /// `Text`/`InteractionPrompt`/every other chunk kind already does.
    #[test]
    fn c2_agent_stream_projection_passes_the_blocked_chunk_through_unchanged() {
        let address = SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(7),
                generation: SessionGeneration(3),
            },
        };
        let chunk = AgentStreamChunkV1 {
            source_sequence: 11,
            kind: gate4agent_node_protocol::AgentStreamChunkKindV1::Blocked {
                correlation_id: None,
                tool_class: "Write".to_owned(),
                authority: gate4agent_node_protocol::BlockAuthorityV1::HostGate,
                reason_kind: None,
                reason: "blocked by dangerous-command gate: rule=filesystem-wipe".to_owned(),
                help: None,
            },
        };
        let source = NodeEvent::AgentStream { address: address.clone(), chunk: chunk.clone() };
        let projected = C2NodeEvent::from(&source);
        assert_eq!(projected, C2NodeEvent::AgentStream { address, chunk: chunk.clone() });
        let json = serde_json::to_string(&projected).unwrap();
        assert_eq!(
            json,
            r#"{"kind":"agent-stream","address":{"workspace_id":"primary","session":{"instance_id":7,"generation":3}},"chunk":{"source_sequence":11,"kind":{"kind":"blocked","correlation_id":null,"tool_class":"Write","authority":"host-gate","reason_kind":null,"reason":"blocked by dangerous-command gate: rule=filesystem-wipe","help":null}}}"#,
        );
        assert_eq!(serde_json::from_str::<C2NodeEvent>(&json).unwrap(), projected);
    }

    #[test]
    fn c2_terminal_frame_events_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(
            C2_TERMINAL_FRAME_EVENTS_CAPABILITY,
            "terminal-frame-events-v1",
        );
        assert_eq!(
            C2_TERMINAL_FRAME_EVENTS_CAPABILITY,
            NODE_TERMINAL_FRAME_EVENTS_CAPABILITY,
        );
        let capability = CapabilityId::new(C2_TERMINAL_FRAME_EVENTS_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());

        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
        let transcript = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        let hex = transcript
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect::<String>();
        let stamp = build_stamp_hex();
        assert_eq!(
            hex,
            format!(
                concat!(
                    "67617465346167656e742d63322d636f6e74726f6c2d617574682d76322d636f6d7061746962696c69747900",
                    "{stamp}",
                    "01",
                    "1111111111111111111111111111111111111111111111111111111111111111",
                    "2222222222222222222222222222222222222222222222222222222222222222",
                    "6f6666657200",
                    "{stamp}",
                    "0100",
                    "18007465726d696e616c2d6672616d652d6576656e74732d763100",
                    "73656c656374656400",
                    "{stamp}",
                    "0100",
                    "18007465726d696e616c2d6672616d652d6576656e74732d7631",
                    "060064617277696e0700616172636836340201",
                ),
                stamp = stamp,
            ),
        );
    }

    #[test]
    fn c2_acp_control_capability_is_optional_and_negotiable() {
        assert_eq!(C2_ACP_CONTROL_CAPABILITY, "acp-control-v1");
        assert_eq!(C2_ACP_CONTROL_CAPABILITY, NODE_ACP_CONTROL_CAPABILITY);
        let capability = CapabilityId::new(C2_ACP_CONTROL_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());

        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer,
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
    }

    #[test]
    fn c2_spawn_spec_defaults_overrides_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(
            C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY,
            "spawn-spec.defaults-overrides-v1",
        );
        assert_eq!(
            C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY,
            NODE_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY,
        );
        let capability =
            CapabilityId::new(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());

        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        let without_spawn_spec = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &ClientCompatibilityOffer {
                build_stamp: offer.build_stamp.clone(),
                capabilities: Vec::new(),
                state_schema: None,
            },
            &NegotiatedC2ControlCompatibility {
                build_stamp: selected.build_stamp.clone(),
                capabilities: Vec::new(),
                host: selected.host.clone(),
                path_semantics: selected.path_semantics.clone(),
            },
        )
        .unwrap();
        assert_ne!(bound, without_spawn_spec);
        assert!(bound.windows(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY.len()).any(
            |window| window == C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY.as_bytes()
        ));
    }

    #[test]
    fn c2_spawn_profile_revision_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(
            C2_SPAWN_PROFILE_REVISION_CAPABILITY,
            "spawn-spec.profile-revision-v1",
        );
        assert_eq!(
            C2_SPAWN_PROFILE_REVISION_CAPABILITY,
            NODE_SPAWN_PROFILE_REVISION_CAPABILITY,
        );
        let capability =
            CapabilityId::new(C2_SPAWN_PROFILE_REVISION_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());

        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        assert!(bound.windows(C2_SPAWN_PROFILE_REVISION_CAPABILITY.len()).any(
            |window| window == C2_SPAWN_PROFILE_REVISION_CAPABILITY.as_bytes()
        ));
    }

    #[test]
    fn c2_worktree_selection_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(C2_WORKTREE_SELECTION_CAPABILITY, "worktree-selection-v1");
        assert_eq!(
            C2_WORKTREE_SELECTION_CAPABILITY,
            NODE_WORKTREE_SELECTION_CAPABILITY,
        );
        let capability = CapabilityId::new(C2_WORKTREE_SELECTION_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());

        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        let without_worktree_selection = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &ClientCompatibilityOffer {
                build_stamp: offer.build_stamp.clone(),
                capabilities: Vec::new(),
                state_schema: None,
            },
            &NegotiatedC2ControlCompatibility {
                build_stamp: selected.build_stamp.clone(),
                capabilities: Vec::new(),
                host: selected.host.clone(),
                path_semantics: selected.path_semantics.clone(),
            },
        )
        .unwrap();
        assert_ne!(bound, without_worktree_selection);
        assert!(bound.windows(C2_WORKTREE_SELECTION_CAPABILITY.len()).any(
            |window| window == C2_WORKTREE_SELECTION_CAPABILITY.as_bytes()
        ));
    }

    #[test]
    fn c2_managed_worktree_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(
            C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY,
            "managed-worktree-lifecycle-v1",
        );
        assert_eq!(
            C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY,
            "managed-worktree-spawn-v2",
        );
        assert_eq!(
            C2NodeFailure::from(&NodeFailure {
                code: NodeFailureCode::ManagedWorktreeProfileRevisionMismatch,
                message: "private node detail".to_owned(),
            })
            .message,
            "managed worktree profile revision mismatch",
        );
        let capability =
            CapabilityId::new(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        assert!(bound
            .windows(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY.len())
            .any(|window| {
                window == C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY.as_bytes()
            }));
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());
    }

    #[test]
    fn c2_child_environment_profile_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(
            C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY,
            "child-environment-profile-v1",
        );
        let capability =
            CapabilityId::new(C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        assert!(bound
            .windows(C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY.len())
            .any(|window| {
                window == C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY.as_bytes()
            }));
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());
    }

    #[test]
    fn c2_session_bundle_materialization_is_auth_bound_and_projected_as_opaque_metadata() {
        assert_eq!(
            C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY,
            "session-bundle-materialization-v1",
        );
        assert_eq!(
            C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY,
            NODE_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY,
        );
        let capability =
            CapabilityId::new(C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        assert!(bound
            .windows(C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY.len())
            .any(|window| {
                window == C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY.as_bytes()
            }));

        let mut record = private_session_record();
        record.bundle = Some(ResolvedBundleReceipt {
            id: SpawnBundleId::new("review-bundle").unwrap(),
            revision: SpawnBundleRevision::new("review-bundle.r1").unwrap(),
            digest: SpawnBundleDigest::new(format!("sha256:{}", "a".repeat(64)))
                .unwrap(),
        });
        let projected = C2ManagedSessionRecord::from(&record);
        let json = serde_json::to_string(&projected).unwrap();
        assert!(json.contains(
            r#""bundle":{"id":"review-bundle","revision":"review-bundle.r1","digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}"#,
        ));
        assert_private_record_fields_absent(&json);
        assert!(C2NodeEvent::SessionRecordUpserted { record: projected }
            .requires_session_bundle_materialization_capability());
    }

    #[test]
    fn c2_history_context_pack_capability_is_optional_and_auth_bound_exactly() {
        assert_eq!(C2_HISTORY_CONTEXT_PACK_CAPABILITY, "history-context-pack-v1");
        assert_eq!(
            C2_HISTORY_CONTEXT_PACK_CAPABILITY,
            NODE_HISTORY_CONTEXT_PACK_CAPABILITY,
        );
        let capability = CapabilityId::new(C2_HISTORY_CONTEXT_PACK_CAPABILITY).unwrap();
        let support = c2_compatibility_support(vec![capability.clone()],
        );
        assert!(support
            .negotiate(&C2ClientHello::new([0; C2_AUTH_NONCE_BYTES]))
            .unwrap()
            .capabilities
            .is_empty());

        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = support
            .negotiate(&C2ClientHello::negotiating(
                [0; C2_AUTH_NONCE_BYTES],
                offer.clone(),
            ))
            .unwrap();
        assert_eq!(selected.capabilities, vec![capability]);
        let bound = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        )
        .unwrap();
        assert!(bound
            .windows(C2_HISTORY_CONTEXT_PACK_CAPABILITY.len())
            .any(|window| window == C2_HISTORY_CONTEXT_PACK_CAPABILITY.as_bytes()));

        for code in [
            NodeFailureCode::UnknownContextPack,
            NodeFailureCode::ContextPackBusy,
            NodeFailureCode::ContextPackMaterializationFailed,
        ] {
            let failure = C2NodeFailure::from(&NodeFailure {
                code,
                message: "private context backend detail".to_owned(),
            });
            assert!(failure.requires_history_context_pack_capability());
            assert!(!failure.message.contains("private"));
        }
        assert!(!C2NodeFailure::from(&NodeFailure {
            code: NodeFailureCode::UnknownSession,
            message: "private session backend detail".to_owned(),
        })
        .requires_history_context_pack_capability());
    }

    #[test]
    fn c2_history_context_pack_requests_are_exact_and_capability_gated() {
        let session = SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(41),
                generation: SessionGeneration(3),
            },
        };
        let requests = [
            (
                NodeRequest::DiscoverHistory { session: session.clone(), limit: 4 },
                r#"{"kind":"discover-history","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"limit":4}"#,
            ),
            (
                NodeRequest::LoadHistory {
                    session: session.clone(),
                    candidate_id: "candidate-7".to_owned(),
                },
                r#"{"kind":"load-history","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"candidate_id":"candidate-7"}"#,
            ),
            (
                NodeRequest::ExportContextPack { session },
                r#"{"kind":"export-context-pack","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}}}"#,
            ),
            (
                NodeRequest::ForgetContextPack {
                    context_id: SpawnContextId::new("context-review-7").unwrap(),
                },
                r#"{"kind":"forget-context-pack","context_id":"context-review-7"}"#,
            ),
        ];

        for (request, expected) in requests {
            assert_eq!(request.required_capability(), Some(C2_HISTORY_CONTEXT_PACK_CAPABILITY));
            assert!(request.requires_history_context_pack_capability());
            assert_eq!(serde_json::to_string(&request).unwrap(), expected);
            assert_eq!(serde_json::from_str::<NodeRequest>(expected).unwrap(), request);
        }
        assert!(serde_json::from_str::<NodeRequest>(
            r#"{"kind":"export-context-pack","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"path":"private.jsonl"}"#,
        )
        .is_err());
    }

    #[test]
    fn c2_session_record_context_export_is_exact_correlated_and_private() {
        assert_eq!(
            C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY,
            "session-record-context-export-v1",
        );
        let session = SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(41),
                generation: SessionGeneration(3),
            },
        };
        let record_id = SessionRecordId::new("record-context-41").unwrap();
        let request = NodeRequest::ExportContextPackForSessionRecord {
            record_id: record_id.clone(),
            session: session.clone(),
        };
        assert_eq!(
            request.required_capability(),
            Some(C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY),
        );
        assert!(request.requires_history_context_pack_capability());
        let request_json = serde_json::to_string(&request).unwrap();
        assert_eq!(serde_json::from_str::<NodeRequest>(&request_json).unwrap(), request);

        let response = C2NodeResponse::from(
            &NodeResponse::ContextPackForSessionRecordExported {
                record_id,
                session,
                context: context_pack_receipt(),
            },
        );
        assert!(response.requires_history_context_pack_capability());
        assert!(response.requires_session_record_context_export_capability());
        let response_json = serde_json::to_string(&response).unwrap();
        assert_eq!(
            serde_json::from_str::<C2NodeResponse>(&response_json).unwrap(),
            response,
        );
        for private in [
            "candidate_id",
            "session_id_hint",
            "provider_session",
            "messages",
            "model",
            "path",
        ] {
            assert!(!request_json.contains(private), "request leaked {private}");
            assert!(!response_json.contains(private), "response leaked {private}");
        }
    }

    #[test]
    fn c2_context_metadata_projects_through_records_snapshots_events_and_responses() {
        let context = context_pack_receipt();
        let mut source = private_session_record();
        source.context_id = Some(context.id.clone());
        source.context = Some(context.clone());
        let projected = C2ManagedSessionRecord::from(&source);
        assert!(projected.context_binding_is_valid());
        assert!(projected.requires_history_context_pack_capability());
        assert_eq!(projected.context_id.as_ref(), Some(&context.id));
        assert_eq!(projected.context.as_ref(), Some(&context));

        let record_json = serde_json::to_string(&projected).unwrap();
        assert_private_record_fields_absent(&record_json);
        assert_eq!(
            serde_json::from_str::<C2ManagedSessionRecord>(&record_json).unwrap(),
            projected,
        );
        let mut mismatched = serde_json::from_str::<serde_json::Value>(&record_json).unwrap();
        mismatched["context_id"] = serde_json::json!("context-other");
        assert!(serde_json::from_value::<C2ManagedSessionRecord>(mismatched).is_err());

        let snapshot = C2NodeSnapshot::from(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: vec![source.clone()],
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        assert!(snapshot.requires_history_context_pack_capability());
        let event = C2NodeEvent::from(&NodeEvent::SessionRecordUpserted {
            record: source.clone(),
        });
        assert!(event.requires_history_context_pack_capability());

        let updated = C2NodeResponse::from(&NodeResponse::SessionRecordUpdated {
            record: source,
        });
        assert!(updated.requires_history_context_pack_capability());
        assert!(serde_json::to_string(&updated).unwrap().contains("context-review-7"));

        let exported = C2NodeResponse::from(&NodeResponse::ContextPackExported {
            context: context.clone(),
        });
        assert!(exported.requires_history_context_pack_capability());
        let json = serde_json::to_string(&exported).unwrap();
        assert!(json.contains(r#""kind":"context-pack-exported""#));
        assert!(json.contains(r#""source_message_count":9"#));
        assert!(!json.contains("messages"));
        assert!(!json.contains("path"));
        assert_eq!(serde_json::from_str::<C2NodeResponse>(&json).unwrap(), exported);
    }

    #[test]
    fn c2_exported_context_projects_through_records_and_is_provider_correlated() {
        let exported = context_pack_receipt();
        let mut source = private_session_record();
        source.exported_context = Some(exported.clone());
        let projected = C2ManagedSessionRecord::from(&source);
        assert!(projected.exported_context_is_valid());
        assert_eq!(projected.exported_context.as_ref(), Some(&exported));

        let record_json = serde_json::to_string(&projected).unwrap();
        assert_private_record_fields_absent(&record_json);
        assert_eq!(
            serde_json::from_str::<C2ManagedSessionRecord>(&record_json).unwrap(),
            projected,
        );

        let mut mismatched_provider =
            serde_json::from_str::<serde_json::Value>(&record_json).unwrap();
        mismatched_provider["exported_context"]["lineage"]["source_provider"] =
            serde_json::json!("claude");
        assert!(serde_json::from_value::<C2ManagedSessionRecord>(mismatched_provider).is_err());
    }

    #[test]
    fn c2_history_responses_are_exact_bounded_metadata_without_messages_or_paths() {
        let session = SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(41),
                generation: SessionGeneration(3),
            },
        };
        let discovered = C2NodeResponse::from(&NodeResponse::HistoryDiscovered {
            session: session.clone(),
            candidates: vec![HistoryCandidateSummary {
                id: "candidate-7".to_owned(),
                session_id_hint: "session-hint-7".to_owned(),
                modified_at_unix_ms: Some(77),
            }],
        });
        let loaded = C2NodeResponse::from(&NodeResponse::HistoryLoaded {
            session,
            session_id: "session-loaded-7".to_owned(),
            message_count: 12,
            completed_turn_count: Some(5),
        });
        let forgotten = C2NodeResponse::from(&NodeResponse::ContextPackForgotten {
            context_id: SpawnContextId::new("context-review-7").unwrap(),
        });
        assert_eq!(
            serde_json::to_string(&discovered).unwrap(),
            r#"{"kind":"history-discovered","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"candidates":[{"id":"candidate-7","session_id_hint":"session-hint-7","modified_at_unix_ms":77}]}"#,
        );
        assert_eq!(
            serde_json::to_string(&loaded).unwrap(),
            r#"{"kind":"history-loaded","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"session_id":"session-loaded-7","message_count":12,"completed_turn_count":5}"#,
        );
        let legacy_loaded_json = r#"{"kind":"history-loaded","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"session_id":"session-loaded-7","message_count":12}"#;
        let legacy_loaded = serde_json::from_str::<C2NodeResponse>(legacy_loaded_json).unwrap();
        assert_eq!(serde_json::to_string(&legacy_loaded).unwrap(), legacy_loaded_json);
        assert!(matches!(
            legacy_loaded,
            C2NodeResponse::HistoryLoaded {
                completed_turn_count: None,
                ..
            }
        ));
        assert_eq!(
            serde_json::to_string(&forgotten).unwrap(),
            r#"{"kind":"context-pack-forgotten","context_id":"context-review-7"}"#,
        );
        for response in [discovered, loaded, forgotten] {
            assert!(response.requires_history_context_pack_capability());
            let json = serde_json::to_string(&response).unwrap();
            assert!(!json.contains("messages"));
            assert!(!json.contains("path"));
            assert_eq!(serde_json::from_str::<C2NodeResponse>(&json).unwrap(), response);
        }
        assert!(serde_json::from_str::<C2NodeResponse>(
            r#"{"kind":"history-discovered","session":{"workspace_id":"primary","session":{"instance_id":41,"generation":3}},"candidates":[{"id":"duplicate","session_id_hint":"one","modified_at_unix_ms":null},{"id":"duplicate","session_id_hint":"two","modified_at_unix_ms":null}]}"#,
        )
        .is_err());
    }

    #[test]
    fn native_session_catalog_projection_is_exact_and_capability_bound() {
        assert_eq!(
            C2_NATIVE_SESSION_CATALOG_CAPABILITY,
            "native-session-catalog-v2"
        );
        let route = NativeSessionCatalogRoute::workspace(
            WorkspaceId::new("primary").unwrap(),
            provider("codex"),
        );
        let projected = C2NodeResponse::from(&NodeResponse::NativeSessionsCataloged {
            route: route.clone(),
            entries: vec![NativeSessionCatalogEntry {
                selection_id: "hist_selection_7".to_owned(),
                title: Some("Review".to_owned()),
                modified_at_unix_ms: Some(77),
                model: Some("model-7".to_owned()),
                message_count: 8,
                completed_turn_count: Some(4),
                external_group: None,
                record_id: Some(SessionRecordId::new("record-7").unwrap()),
            }],
            summary: Some(NativeSessionCatalogSummary {
                catalog_revision: 7,
                recent_cutoff_unix_ms: 70,
                recent_total_count: 1,
                older_total_count: 3,
                recent_next_after_selection_id: None,
                recent_has_more: false,
            }),
        });
        assert!(projected.requires_native_session_catalog_capability());
        let json = serde_json::to_string(&projected).unwrap();
        for forbidden in [
            "session_id",
            "cwd",
            "candidate",
            "path",
            "messages",
            "tokens",
            "documents",
            "raw",
        ] {
            assert!(!json.contains(forbidden));
        }
        assert_eq!(serde_json::from_str::<C2NodeResponse>(&json).unwrap(), projected);
        assert!(serde_json::from_str::<C2NodeResponse>(
            r#"{"kind":"native-sessions-cataloged","workspace_id":"primary","provider":"codex","entries":[]}"#,
        )
        .is_err());

        let paged = C2NodeResponse::from(&NodeResponse::NativeSessionsPaged {
            route,
            page: NativeSessionCatalogPage {
                window: NativeSessionCatalogWindow::Older,
                revision: 7,
                entries: Vec::new(),
                next_after_selection_id: None,
                remaining_count: 0,
                has_more: false,
            },
        });
        assert!(paged.requires_native_session_catalog_paging_capability());
    }

    #[test]
    fn session_record_preview_projection_does_not_expose_native_identity() {
        let response = NodeResponse::SessionRecordPreviewed {
            record_id: SessionRecordId::new("record-7").unwrap(),
            preview: gate4agent_types::SessionRecordPreview {
                title: Some("Review".to_owned()),
                modified_at_unix_ms: Some(77),
                model: Some("model-7".to_owned()),
                message_count: 3,
                message_count_exact: true,
                completed_turn_count: Some(1),
                total_tokens: None,
                truncated: true,
                messages: vec![gate4agent_types::NativeSessionPreviewMessage {
                    role: gate4agent_types::HistoryMessageRole::User,
                    text: "visible dialogue".to_owned(),
                }],
            },
        };
        let projected = C2NodeResponse::from(&response);
        assert!(projected.requires_native_session_preview_capability());
        let json = serde_json::to_string(&projected).unwrap();
        for forbidden in ["session_id", "provider", "workspace_id", "cwd", "path", "tokens"] {
            assert!(!json.contains(forbidden));
        }
        assert_eq!(serde_json::from_str::<C2NodeResponse>(&json).unwrap(), projected);
    }

    #[test]
    fn legacy_slim_projection_deliberately_strips_context_metadata() {
        let context = context_pack_receipt();
        let mut source = private_session_record();
        source.context_id = Some(context.id.clone());
        source.context = Some(context);
        let c2 = C2ManagedSessionRecord::from(&source);

        for json in [
            serde_json::to_value(SlimManagedSessionRecord::from(&source)).unwrap(),
            serde_json::to_value(SlimManagedSessionRecord::from(&c2)).unwrap(),
        ] {
            assert!(json.get("context_id").is_none());
            assert!(json.get("context").is_none());
        }
    }

    #[test]
    fn c2_legacy_node_event_bytes_remain_exact_after_terminal_frame_addition() {
        assert_eq!(
            serde_json::to_vec(&C2NodeEvent::ResyncRequired {
                oldest_available_sequence: 7,
            })
            .unwrap(),
            br#"{"kind":"resync-required","oldest_available_sequence":7}"#,
        );
    }

    #[test]
    fn c2_compatibility_missing_offer_negotiates_local_build_stamp() {
        let support = c2_compatibility_support(Vec::new());

        let negotiated = support
            .negotiate(&C2ClientHello::new([1; C2_AUTH_NONCE_BYTES]))
            .unwrap();

        assert_eq!(negotiated.build_stamp, BUILD_STAMP);
        assert!(negotiated.capabilities.is_empty());
    }

    #[test]
    fn c2_compatibility_selects_local_build_stamp_and_capability_intersection() {
        let shared = CapabilityId::new("terminal-stream").unwrap();
        let server_only = CapabilityId::new("server-only").unwrap();
        let client_only = CapabilityId::new("client-only").unwrap();
        let support = c2_compatibility_support(vec![shared.clone(), server_only]);
        let hello = C2ClientHello::negotiating(
            [2; C2_AUTH_NONCE_BYTES],
            ClientCompatibilityOffer {
                build_stamp: BUILD_STAMP.to_owned(),
                capabilities: vec![client_only, shared.clone()],
                state_schema: None,
            },
        );

        let negotiated = support.negotiate(&hello).unwrap();

        assert_eq!(negotiated.build_stamp, BUILD_STAMP);
        assert_eq!(negotiated.capabilities, vec![shared]);
    }

    #[test]
    fn c2_compatibility_negotiate_rejects_a_foreign_build_stamp_naming_both_values() {
        let support = c2_compatibility_support(Vec::new());
        let foreign_stamp = "f".repeat(40);
        let hello = C2ClientHello {
            build_stamp: foreign_stamp.clone(),
            client_nonce: [3; C2_AUTH_NONCE_BYTES],
            compatibility: None,
        };

        let error = support.negotiate(&hello).unwrap_err();
        assert!(matches!(
            &error,
            ProtocolNegotiationError::BuildStampMismatch { local, remote }
                if local == BUILD_STAMP && remote == &foreign_stamp,
        ));
        assert_eq!(
            error.to_string(),
            format!("build stamp mismatch: local={BUILD_STAMP} remote={foreign_stamp}"),
        );
    }

    #[test]
    fn c2_compatibility_bound_auth_transcript_is_exact_and_selection_sensitive() {
        let capability = CapabilityId::new(C2_COMPATIBILITY_METADATA_CAPABILITY).unwrap();
        let offer = ClientCompatibilityOffer {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability.clone()],
            state_schema: None,
        };
        let selected = NegotiatedC2ControlCompatibility {
            build_stamp: BUILD_STAMP.to_owned(),
            capabilities: vec![capability],
            host: HostDescriptor {
                operating_system: OperatingSystemId::new("windows").unwrap(),
                architecture: ArchitectureId::new("x86_64").unwrap(),
            },
            path_semantics: PathSemantics {
                style: PathStyle::Windows,
                encoding: PathEncoding::Utf8,
            },
        };
        let transcript = c2_bound_auth_transcript(
            C2AuthDirection::Server,
            &[0x11; C2_AUTH_NONCE_BYTES],
            &[0x22; C2_AUTH_NONCE_BYTES],
            &offer,
            &selected,
        ).unwrap();
        let hex = transcript.iter().map(|byte| format!("{byte:02x}")).collect::<String>();

        let stamp = build_stamp_hex();
        assert_eq!(
            hex,
            format!(
                concat!(
                    "67617465346167656e742d63322d636f6e74726f6c2d617574682d76322d636f6d7061746962696c69747900",
                    "{stamp}",
                    "01",
                    "1111111111111111111111111111111111111111111111111111111111111111",
                    "2222222222222222222222222222222222222222222222222222222222222222",
                    "6f6666657200",
                    "{stamp}",
                    "0100",
                    "1600636f6d7061746962696c6974792e6d6574616461746100",
                    "73656c656374656400",
                    "{stamp}",
                    "0100",
                    "1600636f6d7061746962696c6974792e6d65746164617461",
                    "070077696e646f777306007838365f36340101",
                ),
                stamp = stamp,
            ),
        );

        let mut tampered = selected;
        tampered.path_semantics.style = PathStyle::Posix;
        assert_ne!(
            transcript,
            c2_bound_auth_transcript(
                C2AuthDirection::Server,
                &[0x11; C2_AUTH_NONCE_BYTES],
                &[0x22; C2_AUTH_NONCE_BYTES],
                &offer,
                &tampered,
            ).unwrap(),
        );
    }

    #[test]
    fn c2_compatibility_preserves_foreign_host_and_opaque_path() {
        let support = c2_compatibility_support(Vec::new(),
        );
        let negotiated = support
            .negotiate(&C2ClientHello::new([4; C2_AUTH_NONCE_BYTES]))
            .unwrap();
        let projected = C2NodeSnapshot::from(&NodeSnapshot {
            node_id: NodeId::new("remote-mac").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: vec![WorkspaceSnapshot {
                workspace_id: WorkspaceId::new("repo").unwrap(),
                canonical_root: host_path("/srv/CaseSensitive/../literal-root"),
                sessions: Vec::new(),
                worktree_service_mode: None,
                managed_worktree_profiles: None,
            }],
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        let json = serde_json::to_string(&(projected, negotiated)).unwrap();
        let (projected, negotiated) = serde_json::from_str::<(
            C2NodeSnapshot,
            NegotiatedC2ControlCompatibility,
        )>(&json).unwrap();

        assert_eq!(
            projected.workspaces[0].canonical_root.display_text(),
            "/srv/CaseSensitive/../literal-root",
        );
        assert_eq!(negotiated.host.operating_system.as_str(), "darwin");
        assert_eq!(negotiated.host.architecture.as_str(), "aarch64");
        assert_eq!(negotiated.path_semantics.style, PathStyle::Posix);
    }

    #[test]
    fn c2_workspace_inspection_preserves_legacy_utf8_repository_path_shape() {
        let inspection = C2WorkspaceInspection {
            workspace_id: WorkspaceId::new("repo").unwrap(),
            entries: vec![WorkspaceEntry {
                relative_path: repository_path(r"src\literal/main.rs"),
                kind: WorkspaceEntryKind::File,
            }],
            tree_truncated: false,
            git: C2GitSnapshot {
                is_repository: true,
                branch: Some("main".to_owned()),
                status: Vec::new(),
                recent_commits: Vec::new(),
                worktrees: Vec::new(),
                managed_worktree: None,
                truncated: false,
                diagnostic_present: false,
            },
            truncation: None,
        };

        let json = serde_json::to_string(&inspection).unwrap();
        assert_eq!(
            json,
            r#"{"workspace_id":"repo","entries":[{"relative_path":"src\\literal/main.rs","kind":"file"}],"tree_truncated":false,"git":{"is_repository":true,"branch":"main","status":[],"recent_commits":[],"worktrees":[],"truncated":false,"diagnostic_present":false}}"#,
        );
        assert_eq!(serde_json::from_str::<C2WorkspaceInspection>(&json).unwrap(), inspection);
    }

    #[test]
    fn c2_workspace_inspection_roundtrips_all_tagged_repository_path_fields() {
        let entry_path = RepositoryPath::unix_bytes(vec![b's', b'r', b'c', b'/', 0xfe]).unwrap();
        let status_path = RepositoryPath::unix_bytes(vec![b's', b'r', b'c', b'/', 0xff]).unwrap();
        let previous_path = RepositoryPath::unix_bytes(vec![b'o', b'l', b'd', b'/', 0xfd]).unwrap();
        let inspection = C2WorkspaceInspection {
            workspace_id: WorkspaceId::new("repo").unwrap(),
            entries: vec![WorkspaceEntry {
                relative_path: entry_path,
                kind: WorkspaceEntryKind::File,
            }],
            tree_truncated: false,
            git: C2GitSnapshot {
                is_repository: true,
                branch: None,
                status: vec![GitStatusEntry {
                    index_status: "R".to_owned(),
                    worktree_status: " ".to_owned(),
                    path: status_path,
                    previous_path: Some(previous_path),
                }],
                recent_commits: Vec::new(),
                worktrees: Vec::new(),
                managed_worktree: None,
                truncated: false,
                diagnostic_present: false,
            },
            truncation: None,
        };

        let json = serde_json::to_string(&inspection).unwrap();
        assert!(json.contains(r#""relative_path":{"kind":"unix-bytes""#));
        assert!(json.contains(r#""path":{"kind":"unix-bytes""#));
        assert!(json.contains(r#""previous_path":{"kind":"unix-bytes""#));
        assert_eq!(serde_json::from_str::<C2WorkspaceInspection>(&json).unwrap(), inspection);
    }

    #[test]
    fn c2_workspace_inspection_projects_and_validates_managed_git_scope_without_paths() {
        let workspace_id = WorkspaceId::new("managed-a").unwrap();
        let source_workspace_id = WorkspaceId::new("primary").unwrap();
        let inspection = WorkspaceInspection {
            workspace_id: workspace_id.clone(),
            entries: Vec::new(),
            tree_truncated: false,
            git: GitSnapshot {
                is_repository: true,
                branch: Some("gate4agent/a".to_owned()),
                status: Vec::new(),
                recent_commits: Vec::new(),
                worktrees: Vec::new(),
                managed_worktree: Some(ManagedWorktreeGitScope {
                    lease_id: ManagedWorktreeLeaseId::new("mw-a").unwrap(),
                    source_workspace_id: source_workspace_id.clone(),
                    branch: "gate4agent/a".to_owned(),
                    base_commit: gate4agent_node_protocol::GitObjectId::new(
                        "0123456789abcdef0123456789abcdef01234567".to_owned(),
                    )
                    .unwrap(),
                    active_session_count: 1,
                    managed_record_count: 1,
                }),
                truncated: false,
                diagnostic: None,
            },
            truncation: None,
        };
        let projected = C2WorkspaceInspection::from(&inspection);
        let json = serde_json::to_string(&projected).unwrap();
        assert!(json.contains("\"managed_worktree\""));
        assert!(json.contains("\"lease_id\":\"mw-a\""));
        assert!(!json.contains("target_root"));
        assert_eq!(
            serde_json::from_str::<C2WorkspaceInspection>(&json).unwrap(),
            projected,
        );

        let invalid = json.replace(
            "\"source_workspace_id\":\"primary\"",
            "\"source_workspace_id\":\"managed-a\"",
        );
        assert!(serde_json::from_str::<C2WorkspaceInspection>(&invalid).is_err());

        let mut invalid_node = inspection;
        invalid_node.git.branch = Some("gate4agent/b".to_owned());
        assert!(C2WorkspaceInspection::from(&invalid_node)
            .git
            .managed_worktree
            .is_none());
    }

    #[test]
    fn slim_inventory_is_deterministic_and_excludes_terminal_history() {
        let snapshot = NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex"), provider("claude"), provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: vec![
                WorkspaceSnapshot { workspace_id: WorkspaceId::new("z-work").unwrap(), canonical_root: host_path("z"), sessions: Vec::new(), worktree_service_mode: None, managed_worktree_profiles: None },
                WorkspaceSnapshot {
                    workspace_id: WorkspaceId::new("a-work").unwrap(),
                    canonical_root: host_path("a"),
                    sessions: vec![fixture_session()],
                    worktree_service_mode: None,
                    managed_worktree_profiles: None,
                },
            ],
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        };
        let slim = SlimNodeInventory::from_snapshot(&snapshot);
        assert_eq!(slim.enabled_providers, vec![provider("claude"), provider("codex")]);
        assert_eq!(slim.workspaces.keys().map(WorkspaceId::as_str).collect::<Vec<_>>(), vec!["a-work", "z-work"]);
        let session = &slim.workspaces[&WorkspaceId::new("a-work").unwrap()].sessions[0];
        assert_eq!(session.transport, TransportKind::Pty);
        assert_eq!(session.process_id, Some(1234));
        assert_eq!(session.terminal_size, Some(TerminalSize { rows: 40, columns: 120 }));
        assert!(session.operation_pending);
        assert!(session.input_pending);
        let json = serde_json::to_string(&slim).unwrap();
        assert!(!json.contains("terminal_frame"));
        assert!(!json.contains("history"));
    }

    /// A `SlimSession` payload from a node that predates `screen_state`
    /// decodes the missing key as `Unknown`, never as the optimistic
    /// `Ready` -- the entire point of `#[serde(default)]` on that field.
    #[test]
    fn slim_session_json_omitting_screen_state_decodes_to_unknown_not_ready() {
        let json = r#"{
            "instance_id": 7,
            "generation": 2,
            "agent_id": "codex",
            "transport": "pty",
            "status": "running",
            "process_id": 1234,
            "terminal_size": null,
            "operation_pending": true,
            "input_pending": false
        }"#;
        let session = serde_json::from_str::<SlimSession>(json).unwrap();
        assert_eq!(session.screen_state, PtyScreenState::Unknown);
        assert_ne!(session.screen_state, PtyScreenState::Ready);
    }

    /// The slim projection carries the session's non-default screen
    /// classification through unchanged from the `SessionSnapshot` it was
    /// built from -- proving `SlimNodeInventory::from_snapshot` does not
    /// silently drop it back to the default the way an unset field would.
    #[test]
    fn slim_inventory_projection_carries_a_non_default_screen_state_from_the_session_snapshot() {
        let mut session = fixture_session();
        session.screen_state = PtyScreenState::OperatorGate {
            gate: gate4agent_types::OperatorGateState::new(gate4agent_types::OperatorGateKind::WorkspaceTrust),
        };
        let snapshot = NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: vec![WorkspaceSnapshot {
                workspace_id: WorkspaceId::new("a-work").unwrap(),
                canonical_root: host_path("a"),
                sessions: vec![session.clone()],
                worktree_service_mode: None,
                managed_worktree_profiles: None,
            }],
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        };
        let slim = SlimNodeInventory::from_snapshot(&snapshot);
        let slim_session = &slim.workspaces[&WorkspaceId::new("a-work").unwrap()].sessions[0];
        assert_eq!(slim_session.screen_state, session.screen_state);
    }

    #[test]
    fn c2_full_and_slim_snapshots_preserve_launch_and_worktree_inventory() {
        let worktree_profiles = WorktreeProfileInventory {
            profiles: vec![ManagedWorktreeProfileSummary {
                id: WorktreeProfileId::new("review").unwrap(),
                revision: WorktreeProfileRevision::new("v1").unwrap(),
                retention: ManagedWorktreeRetention::Retain,
            }],
        };
        let launch_inventory = LaunchInventory {
            spawn_profiles: Some(vec![SpawnProfileSummary {
                id: SpawnProfileId::new("default").unwrap(),
                revision: SpawnProfileRevision::new("v1").unwrap(),
                environment_profile: None,
            }]),
            bundles: Some(vec![ResolvedBundleReceipt {
                id: SpawnBundleId::new("review").unwrap(),
                revision: SpawnBundleRevision::new("v1").unwrap(),
                digest: SpawnBundleDigest::new(format!("sha256:{}", "0".repeat(64))).unwrap(),
            }]),
            network_allowlists: None,
        };
        let snapshot = NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: vec![WorkspaceSnapshot {
                workspace_id: WorkspaceId::new("repo").unwrap(),
                canonical_root: host_path("repo"),
                sessions: Vec::new(),
                worktree_service_mode: Some(WorktreeServiceMode::Managed),
                managed_worktree_profiles: Some(worktree_profiles.clone()),
            }],
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: Some(launch_inventory.clone()),
            agent_progress: Vec::new(),
        };

        let full = C2NodeSnapshot::from(&snapshot);
        let slim_from_node = SlimNodeInventory::from_snapshot(&snapshot);
        let slim_from_c2 = SlimNodeInventory::from_c2_snapshot(&full);
        assert_eq!(full.launch_inventory.as_ref(), Some(&launch_inventory));
        assert_eq!(slim_from_node.launch_inventory.as_ref(), Some(&launch_inventory));
        assert_eq!(slim_from_c2.launch_inventory.as_ref(), Some(&launch_inventory));
        assert_eq!(
            full.workspaces[0].worktree_service_mode,
            Some(WorktreeServiceMode::Managed),
        );
        assert_eq!(
            slim_from_node.workspaces[&WorkspaceId::new("repo").unwrap()]
                .worktree_service_mode,
            Some(WorktreeServiceMode::Managed),
        );
        assert_eq!(
            slim_from_c2.workspaces[&WorkspaceId::new("repo").unwrap()]
                .worktree_service_mode,
            Some(WorktreeServiceMode::Managed),
        );
        assert_eq!(
            full.workspaces[0].managed_worktree_profiles.as_ref(),
            Some(&worktree_profiles),
        );
        assert_eq!(
            slim_from_node.workspaces[&WorkspaceId::new("repo").unwrap()]
                .managed_worktree_profiles
                .as_ref(),
            Some(&worktree_profiles),
        );

        let legacy = serde_json::from_value::<C2WorkspaceSnapshot>(serde_json::json!({
            "workspace_id": "legacy",
            "canonical_root": "legacy-root",
            "sessions": [],
        }))
        .unwrap();
        assert_eq!(legacy.worktree_service_mode, None);
        assert_eq!(legacy.managed_worktree_profiles, None);
        let encoded = serde_json::to_value(legacy).unwrap();
        assert!(encoded.get("worktree_service_mode").is_none());
        assert!(encoded.get("managed_worktree_profiles").is_none());
    }

    #[test]
    fn slim_inventory_provider_contract_projection_is_exact_and_private() {
        let mut slim = SlimNodeInventory::from_snapshot(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        let (provider_contracts, provider_adapter_contracts) =
            provider_contract_manifest();
        slim.provider_contracts = provider_contracts;
        slim.provider_adapter_contracts = provider_adapter_contracts;

        let value = serde_json::to_value(&slim).unwrap();
        assert_eq!(
            value["provider_contracts"],
            serde_json::json!([{
                "provider": "codex",
                "revision": "codex.2026-08",
            }]),
        );
        assert_eq!(
            value["provider_adapter_contracts"],
            serde_json::json!([{
                "provider": "codex",
                "family": "pty-semantic",
                "adapter_id": "codex",
                "revision": "pty-semantic.2026-08",
            }]),
        );
        let json = serde_json::to_string(&value).unwrap();
        for forbidden in [
            "installed_cli_version",
            "executable_path",
            "auth_state",
            "canary_verdict",
            "events",
            "routed_response",
        ] {
            assert!(!json.contains(forbidden), "leaked private field {forbidden}");
        }
    }

    #[test]
    fn c2_projection_omits_path_stdout_stderr_env_and_fallback_reason() {
        let runtime_statuses = ProviderRuntimeStatuses::new([
            ProviderRuntimeStatus::raw_passthrough(
                provider("codex"),
                Some(ProviderRuntimeVersion::new("0.999.0").unwrap()),
            ),
        ])
        .unwrap();
        let snapshot = NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: runtime_statuses.clone(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        };
        let projected = C2NodeSnapshot::from(&snapshot);
        let inventory = SlimNodeInventory::from_snapshot(&snapshot);
        assert_eq!(projected.provider_runtime_statuses, runtime_statuses);
        assert_eq!(inventory.provider_runtime_statuses, runtime_statuses);
        for encoded in [
            serde_json::to_string(&projected).unwrap(),
            serde_json::to_string(&inventory).unwrap(),
        ] {
            assert!(encoded.contains("\"version\":\"0.999.0\""));
            for forbidden in [
                "launcher", "executable", "stdout", "stderr", "environment", "fallback",
                "reason", "arguments",
            ] {
                assert!(!encoded.contains(forbidden), "leaked field {forbidden}");
            }
        }
    }

    #[test]
    fn slim_inventory_reports_sessions_hidden_by_workspace_truncation() {
        let mut workspaces = (0..MAX_C2_WORKSPACES_PER_NODE)
            .map(|index| WorkspaceSnapshot {
                workspace_id: WorkspaceId::new(format!("work-{index:02}")).unwrap(),
                canonical_root: host_path(format!("root-{index:02}")),
                sessions: Vec::new(),
                worktree_service_mode: None,
                managed_worktree_profiles: None,
            })
            .collect::<Vec<_>>();
        workspaces.push(WorkspaceSnapshot {
            workspace_id: WorkspaceId::new("work-zz").unwrap(),
            canonical_root: host_path("hidden-root"),
            sessions: vec![fixture_session()],
            worktree_service_mode: None,
            managed_worktree_profiles: None,
        });
        let slim = SlimNodeInventory::from_snapshot(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces,
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        assert!(slim.workspaces_truncated);
        assert_eq!(slim.session_count, 1);
        assert!(slim.sessions_truncated);
    }

    #[test]
    fn control_auth_transcript_is_direction_and_protocol_domain_separated() {
        assert_eq!(C2_API_VERSION, 2);
        let client_nonce = [3; C2_AUTH_NONCE_BYTES];
        let server_nonce = [7; C2_AUTH_NONCE_BYTES];
        let server = c2_auth_transcript(C2AuthDirection::Server, &client_nonce, &server_nonce);
        let client = c2_auth_transcript(C2AuthDirection::Client, &client_nonce, &server_nonce);
        assert_ne!(server, client);
        assert!(server.starts_with(b"gate4agent-c2-control-auth-v2\0"));
        assert!(!server.windows(b"gate4agent-node-auth-v3".len()).any(|window| window == b"gate4agent-node-auth-v3"));
        assert_eq!(&server[server.len() - (C2_AUTH_NONCE_BYTES * 2)..server.len() - C2_AUTH_NONCE_BYTES], &client_nonce);
        assert_eq!(&server[server.len() - C2_AUTH_NONCE_BYTES..], &server_nonce);
    }

    #[test]
    fn topology_projection_is_sorted_bounded_and_minimal() {
        let nodes = (0..=MAX_C2_NODES).rev().map(|index| {
            let node_id = NodeId::new(format!("node-{index:03}")).unwrap();
            let incarnation_id = NodeIncarnationId::from_bytes([index as u8; 16]);
            let observed = ObservedNode {
                endpoint: format!(r"\\.\pipe\node-{index:03}"),
                transport_label: "windows-named-pipe".to_owned(),
                transport: NodeTransportState::Online,
                freshness: NodeFreshness::Fresh,
                cursor: Some(NodeCursor { incarnation_id, sequence: 99 }),
                inventory: None,
                last_attempt_unix_ms: Some(10),
                last_success_unix_ms: Some(10),
                consecutive_failures: 0,
                last_error: None,
                gaps: Vec::new(),
                gaps_truncated: 0,
            };
            (node_id, observed)
        }).collect();
        let topology = C2Topology::from_status(&StatusResponse {
            api_version: C2_API_VERSION,
            ready: true,
            observed_at_unix_ms: 10,
            nodes,
        });

        assert_eq!(topology.nodes.len(), MAX_C2_NODES);
        assert_eq!(topology.nodes.first().unwrap().node_id.as_str(), "node-000");
        assert_eq!(topology.nodes.last().unwrap().node_id.as_str(), "node-063");
        assert_eq!(
            topology.nodes[0].current_incarnation_id,
            Some(NodeIncarnationId::from_bytes([0; 16])),
        );
        let json = serde_json::to_string(&C2ServerFrame::Topology(topology)).unwrap();
        assert!(!json.contains("observed_at_unix_ms"));
        assert!(!json.contains("sequence"));
        assert!(!json.contains("inventory"));
        assert!(!json.contains("managed_worktrees"));
    }

    #[test]
    fn topology_node_without_relay_route_decodes_as_unknown() {
        let legacy = r#"{
            "node_id":"node-a",
            "endpoint":"legacy-endpoint",
            "transport":"offline",
            "current_incarnation_id":null
        }"#;

        let decoded: C2TopologyNode = serde_json::from_str(legacy).unwrap();

        assert_eq!(decoded.relay_route, C2RelayRoute::Unknown);
        assert!(!serde_json::to_string(&decoded).unwrap().contains("relay_route"));
    }

    #[test]
    fn topology_provider_contract_projection_is_capability_gated_and_change_sensitive() {
        let (provider_contracts, provider_adapter_contracts) =
            provider_contract_manifest();
        let mut inventory = SlimNodeInventory::from_snapshot(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        inventory.provider_contracts = provider_contracts;
        inventory.provider_adapter_contracts = provider_adapter_contracts;
        let status = StatusResponse {
            api_version: C2_API_VERSION,
            ready: true,
            observed_at_unix_ms: 10,
            nodes: BTreeMap::from([(
                NodeId::new("node-a").unwrap(),
                ObservedNode {
                    endpoint: r"\\.\pipe\node-a".to_owned(),
                    transport_label: "windows-named-pipe".to_owned(),
                    transport: NodeTransportState::Online,
                    freshness: NodeFreshness::Fresh,
                    cursor: Some(NodeCursor {
                        incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                        sequence: 9,
                    }),
                    inventory: Some(inventory),
                    last_attempt_unix_ms: Some(10),
                    last_success_unix_ms: Some(10),
                    consecutive_failures: 0,
                    last_error: None,
                    gaps: Vec::new(),
                    gaps_truncated: 0,
                },
            )]),
        };

        let projected = C2Topology::from_status_with_provider_contracts(&status, true);
        assert_eq!(projected.nodes[0].provider_contracts.len(), 1);
        assert_eq!(projected.nodes[0].provider_adapter_contracts.len(), 1);
        let legacy = C2Topology::from_status_with_provider_contracts(&status, false);
        assert!(legacy.nodes[0].provider_contracts.is_empty());
        assert!(legacy.nodes[0].provider_adapter_contracts.is_empty());
        #[derive(Serialize)]
        struct LegacyTopologyNode<'a> {
            node_id: &'a NodeId,
            endpoint: &'a str,
            relay_route: C2RelayRoute,
            transport: NodeTransportState,
            current_incarnation_id: Option<NodeIncarnationId>,
        }
        #[derive(Serialize)]
        struct LegacyTopology<'a> {
            nodes: Vec<LegacyTopologyNode<'a>>,
        }
        let legacy_shape = LegacyTopology {
            nodes: legacy.nodes.iter().map(|node| LegacyTopologyNode {
                node_id: &node.node_id,
                endpoint: &node.endpoint,
                relay_route: node.relay_route,
                transport: node.transport,
                current_incarnation_id: node.current_incarnation_id,
            }).collect(),
        };
        let legacy_json = serde_json::to_string(&legacy).unwrap();
        assert_eq!(
            serde_json::to_vec(&legacy).unwrap(),
            serde_json::to_vec(&legacy_shape).unwrap(),
        );
        assert!(!legacy_json.contains("provider_contracts"));
        assert!(!legacy_json.contains("provider_adapter_contracts"));
        assert_ne!(legacy, projected);

        let projected_json = serde_json::to_string(&projected).unwrap();
        assert!(projected_json.contains("provider_contracts"));
        assert!(projected_json.contains("provider_adapter_contracts"));
        assert!(!projected_json.contains("inventory"));
        assert!(!projected_json.contains("events"));
        assert!(!projected_json.contains("routed_response"));
    }

    #[test]
    fn routed_durable_session_responses_round_trip_without_private_record_fields() {
        let record = private_session_record();
        let session = record.active_session.clone().unwrap();
        let native_selection = NativeSessionSelection {
            route: NativeSessionCatalogRoute::workspace(
                record.workspace_id.clone(),
                record.provider.clone(),
            ),
            catalog_revision: 7,
            recent_cutoff_unix_ms: 70,
            selection_id: "selection-private".to_owned(),
        };
        let responses = vec![
            NodeResponse::Snapshot {
                event_sequence: 4,
                controller: None,
                snapshot: NodeSnapshot {
                    node_id: NodeId::new("node-a").unwrap(),
                    enabled_providers: vec![provider("codex")],
                    provider_runtime_statuses: ProviderRuntimeStatuses::default(),
                    workspaces: Vec::new(),
                    session_records: vec![record.clone()],
                    managed_worktrees: Vec::new(),
                    launch_inventory: None,
                    agent_progress: Vec::new(),
                },
            },
            NodeResponse::ProviderSessionIndexed { record: record.clone() },
            NodeResponse::NativeSessionIndexed {
                selection: native_selection.clone(),
                record: record.clone(),
            },
            NodeResponse::SessionRecordUpdated { record: record.clone() },
            NodeResponse::SessionRecordResumed {
                record: record.clone(),
                session: session.clone(),
            },
            NodeResponse::Resync {
                event_sequence: 5,
                oldest_available_sequence: 1,
                snapshot: NodeSnapshot {
                    node_id: NodeId::new("node-a").unwrap(),
                    enabled_providers: vec![provider("codex")],
                    provider_runtime_statuses: ProviderRuntimeStatuses::default(),
                    workspaces: Vec::new(),
                    session_records: vec![record.clone()],
                    managed_worktrees: Vec::new(),
                    launch_inventory: None,
                    agent_progress: Vec::new(),
                },
                events: vec![gate4agent_node_protocol::NodeEventEnvelope {
                    sequence: 5,
                    event: NodeEvent::SessionRecordUpserted { record: record.clone() },
                }],
            },
        ];

        for response in responses {
            let json = serde_json::to_string(&routed_response(response)).unwrap();
            assert_private_record_fields_absent(&json);
            assert!(json.contains("provider_identity_present"));
            assert!(json.contains(r#""profile_id":"local-default""#));
            assert!(json.contains(
                r#""profile_revision":"local-default.2026-08""#,
            ));
            let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
            let decoded_record = match decoded.response.unwrap() {
                C2NodeResponse::Snapshot { snapshot, .. } => snapshot.session_records.into_iter().next().unwrap(),
                C2NodeResponse::ProviderSessionIndexed { record }
                | C2NodeResponse::NativeSessionIndexed { record, .. }
                | C2NodeResponse::SessionRecordUpdated { record }
                | C2NodeResponse::SessionRecordResumed { record, .. } => record,
                C2NodeResponse::Resync { snapshot, events, .. } => {
                    assert!(matches!(events.into_iter().next().unwrap().event,
                        C2NodeEvent::SessionRecordUpserted { ref record }
                        if record.provider_identity_present));
                    snapshot.session_records.into_iter().next().unwrap()
                }
                response => panic!("unexpected response after C2 round trip: {response:?}"),
            };
            assert_eq!(decoded_record.record_id.as_str(), "session-private");
            assert_eq!(decoded_record.display_name, "release shepherd");
            assert_eq!(decoded_record.active_session.as_ref(), Some(&session));
            assert!(decoded_record.provider_identity_present);
        }
    }

    #[test]
    fn c2_resync_projection_preserves_authoritative_replay_floor() {
        let source = NodeResponse::Resync {
            event_sequence: 12,
            oldest_available_sequence: 9,
            snapshot: NodeSnapshot {
                node_id: NodeId::new("node-a").unwrap(),
                enabled_providers: Vec::new(),
                provider_runtime_statuses: ProviderRuntimeStatuses::default(),
                workspaces: Vec::new(),
                session_records: Vec::new(),
                managed_worktrees: Vec::new(),
                launch_inventory: None,
                agent_progress: Vec::new(),
            },
            events: Vec::new(),
        };
        let projected = C2NodeResponse::from(&source);
        assert!(matches!(
            projected,
            C2NodeResponse::Resync {
                event_sequence: 12,
                oldest_available_sequence: 9,
                ..
            }
        ));
        let encoded = serde_json::to_vec(&projected).unwrap();
        assert_eq!(
            serde_json::from_slice::<C2NodeResponse>(&encoded).unwrap(),
            projected,
        );
    }

    #[test]
    fn routed_session_record_event_round_trips_without_private_record_fields() {
        let routed = RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([8; 16]),
                sequence: 9,
            },
            event: C2NodeEvent::from(&NodeEvent::SessionRecordUpserted {
                record: private_session_record(),
            }),
        };

        let json = serde_json::to_string(&routed).unwrap();
        assert_private_record_fields_absent(&json);
        assert!(json.contains("provider_identity_present"));
        let decoded = serde_json::from_str::<RoutedNodeEvent>(&json).unwrap();
        let C2NodeEvent::SessionRecordUpserted { record } = decoded.event else {
            panic!("unexpected routed event after C2 round trip");
        };
        assert_eq!(record.record_id.as_str(), "session-private");
        assert_eq!(record.display_name, "release shepherd");
        assert!(record.provider_identity_present);
    }

    #[test]
    fn routed_node_failure_replaces_raw_message_with_fixed_category() {
        let raw = NodeFailure {
            code: NodeFailureCode::BackendOperationFailed,
            message: r"provider token-secret failed at C:\private\relay.log".to_owned(),
        };
        let routed = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Err(C2NodeFailure::from(&raw)),
        };

        let json = serde_json::to_string(&routed).unwrap();
        assert!(!json.contains("token-secret"));
        assert!(!json.contains("private"));
        assert!(!json.contains("relay.log"));
        let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
        assert_eq!(decoded.response.unwrap_err(), C2NodeFailure {
            code: NodeFailureCode::BackendOperationFailed,
            message: "node backend operation failed".to_owned(),
        });
    }

    #[test]
    fn routed_workspace_inspection_omits_raw_git_diagnostics_and_reasons() {
        let worktree = GitWorktreeSnapshot {
            path: host_path(r"C:\work\feature"),
            head: "abc123".to_owned(),
            branch: Some("feature/privacy".to_owned()),
            is_bare: false,
            is_main: false,
            locked: true,
            lock_reason: Some("lock-secret-provider-token".to_owned()),
            prunable: true,
            prunable_reason: Some("prunable-secret-private-path".to_owned()),
            workspace_id: Some(WorkspaceId::new("feature").unwrap()),
        };
        let response = routed_response(NodeResponse::WorkspaceInspected {
            inspection: WorkspaceInspection {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                entries: Vec::new(),
                tree_truncated: false,
                git: GitSnapshot {
                    is_repository: true,
                    branch: Some("main".to_owned()),
                    status: Vec::new(),
                    recent_commits: Vec::new(),
                    worktrees: vec![worktree],
                    managed_worktree: None,
                    truncated: false,
                    diagnostic: Some("diagnostic-secret C:\\private\\git.stderr".to_owned()),
                },
                truncation: None,
            },
        });

        let json = serde_json::to_string(&response).unwrap();
        for secret in [
            "lock-secret-provider-token",
            "prunable-secret-private-path",
            "diagnostic-secret",
            "git.stderr",
        ] {
            assert!(!json.contains(secret));
        }
        assert!(!json.contains("lock_reason"));
        assert!(!json.contains("prunable_reason"));
        assert!(!json.contains("\"diagnostic\":"));
        let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
        let Ok(C2NodeResponse::WorkspaceInspected { inspection }) = decoded.response else {
            panic!("unexpected routed workspace response");
        };
        assert!(inspection.git.diagnostic_present);
        assert!(inspection.git.worktrees[0].locked);
        assert!(inspection.git.worktrees[0].prunable);
        assert_eq!(inspection.git.worktrees[0].path.display_text(), r"C:\work\feature");
    }

    #[test]
    fn routed_workspace_file_read_projects_only_correlated_operator_content() {
        let path = RepositoryPath::utf8("src/lib.rs".to_owned()).unwrap();
        let response = routed_response(NodeResponse::WorkspaceFileRead {
            file: WorkspaceFileRead {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                path: path.clone(),
                content: WorkspaceFileContent::Utf8 {
                    text: "pub fn fixture() {}\n".to_owned(),
                    byte_len: 20,
                },
                revision: None,
            },
        });

        let json = serde_json::to_string(&response).unwrap();
        assert!(!json.contains("canonical_root"));
        assert!(!json.contains("diagnostic"));
        assert!(!json.contains("controller"));
        assert!(!json.contains("inventory"));
        assert!(!json.contains("event_sequence"));
        let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
        assert_eq!(decoded.response, Ok(C2NodeResponse::WorkspaceFileRead {
            file: WorkspaceFileRead {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                path,
                content: WorkspaceFileContent::Utf8 {
                    text: "pub fn fixture() {}\n".to_owned(),
                    byte_len: 20,
                },
                revision: None,
            },
        }));
    }

    #[test]
    fn workspace_entry_create_projects_exact_bounded_c2_contracts() {
        assert_eq!(
            C2_WORKSPACE_ENTRY_CREATE_CAPABILITY,
            "workspace-entry-create-v1",
        );
        let workspace_id = WorkspaceId::new("primary").unwrap();
        let file = WorkspaceFileRead {
            workspace_id: workspace_id.clone(),
            path: repository_path("src/new.rs"),
            content: WorkspaceFileContent::Utf8 {
                text: String::new(),
                byte_len: 0,
            },
            revision: Some(
                gate4agent_node_protocol::WorkspaceFileRevision::new(
                    "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
                        .to_owned(),
                )
                .unwrap(),
            ),
        };
        let file_response = C2NodeResponse::from(
            &NodeResponse::WorkspaceFileCreated { file: file.clone() },
        );
        assert_eq!(
            file_response,
            C2NodeResponse::WorkspaceFileCreated { file },
        );
        assert!(file_response.requires_workspace_entry_create_capability());

        let entry = WorkspaceEntry {
            relative_path: repository_path("src/new"),
            kind: WorkspaceEntryKind::Directory,
        };
        let directory_response = C2NodeResponse::from(
            &NodeResponse::WorkspaceDirectoryCreated {
                workspace_id: workspace_id.clone(),
                entry: entry.clone(),
            },
        );
        assert_eq!(
            directory_response,
            C2NodeResponse::WorkspaceDirectoryCreated {
                workspace_id,
                entry,
            },
        );
        assert!(directory_response.requires_workspace_entry_create_capability());
        let encoded = serde_json::to_string(&directory_response).unwrap();
        assert_eq!(
            serde_json::from_str::<C2NodeResponse>(&encoded).unwrap(),
            directory_response,
        );

        for (code, message) in [
            (NodeFailureCode::RepositoryEntryAlreadyExists, "repository entry already exists"),
            (NodeFailureCode::RepositoryParentNotFound, "repository parent directory unavailable"),
            (NodeFailureCode::RepositoryParentNotDirectory, "repository parent path is not a directory"),
            (NodeFailureCode::RepositoryEntryCreateTimedOut, "repository entry creation timed out"),
            (NodeFailureCode::RepositoryEntryCreateFailed, "repository entry creation failed"),
        ] {
            assert_eq!(
                C2NodeFailure::from(&NodeFailure {
                    code,
                    message: "private node detail".to_owned(),
                })
                .message,
                message,
            );
        }
    }

    #[test]
    fn c2_host_directory_browse_projection_is_bounded_and_sanitizes_failures() {
        assert_eq!(C2_HOST_DIRECTORY_BROWSE_CAPABILITY, "host-directory-browse-v1");
        let directory = host_path(r"C:\Users");
        let entry = HostDirectoryEntry {
            path: host_path(r"C:\Users\Public"),
            display_name: "Public".to_owned(),
            is_link: false,
        };
        let response = routed_response(NodeResponse::HostDirectoriesBrowsed {
            listing: HostDirectoryListing {
                directory: Some(directory.clone()),
                parent: Some(host_path(r"C:\")),
                entries: vec![entry.clone()],
                next_after: Some(entry.path.clone()),
                incomplete: true,
            },
        });
        let json = serde_json::to_string(&response).unwrap();
        let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
        assert_eq!(decoded, response);
        assert!(matches!(
            decoded.response,
            Ok(C2NodeResponse::HostDirectoriesBrowsed {
                listing: HostDirectoryListing {
                    directory: Some(ref actual),
                    incomplete: true,
                    ..
                }
            }) if actual == &directory
        ));

        let mut oversized = serde_json::to_value(C2NodeResponse::HostDirectoriesBrowsed {
            listing: HostDirectoryListing {
                directory: None,
                parent: None,
                entries: vec![entry; gate4agent_node_protocol::MAX_HOST_DIRECTORY_ENTRIES + 1],
                next_after: None,
                incomplete: false,
            },
        }).unwrap();
        assert!(serde_json::from_value::<C2NodeResponse>(oversized.take()).is_err());

        for (code, expected) in [
            (NodeFailureCode::HostDirectoryInvalid, "host directory invalid"),
            (NodeFailureCode::HostDirectoryReadFailed, "host directory read failed"),
            (NodeFailureCode::HostDirectoryReadTimedOut, "host directory read timed out"),
        ] {
            let failure = C2NodeFailure::from(&NodeFailure {
                code,
                message: r"private C:\Users\owner\secret".to_owned(),
            });
            assert_eq!(failure.message, expected);
            assert!(failure.requires_host_directory_browse_capability());
            assert!(!failure.message.contains("private"));
        }
    }

    #[test]
    fn c2_projection_roundtrips_non_utf8_host_paths_without_interpretation() {
        let opaque = OpaqueHostPath::unix_bytes(vec![b'/', b's', b'r', b'v', b'/', 0xff, b'\n', 0x1b]).unwrap();
        let projected = C2NodeSnapshot::from(&NodeSnapshot {
            node_id: NodeId::new("remote-linux").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: vec![WorkspaceSnapshot {
                workspace_id: WorkspaceId::new("repo").unwrap(),
                canonical_root: opaque.clone(),
                sessions: Vec::new(),
                worktree_service_mode: None,
                managed_worktree_profiles: None,
            }],
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });

        let encoded = serde_json::to_string(&projected).unwrap();
        let decoded = serde_json::from_str::<C2NodeSnapshot>(&encoded).unwrap();

        assert_eq!(decoded.workspaces[0].canonical_root, opaque);
        assert_eq!(decoded.workspaces[0].workspace_id.as_str(), "repo");
        let slim = SlimNodeInventory::from_c2_snapshot(&decoded);
        let display = &slim.workspaces[&WorkspaceId::new("repo").unwrap()].canonical_root;
        assert!(!display.chars().any(char::is_control));
        assert!(display.contains("\\n"));
        assert!(display.contains("\\u{1b}"));
    }

    #[test]
    fn standalone_workspace_response_roundtrips_authoritative_snapshot() {
        assert_eq!(
            C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY,
            NODE_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY,
        );
        let workspace = WorkspaceSnapshot {
            workspace_id: WorkspaceId::new("standalone").unwrap(),
            canonical_root: host_path(r"C:\standalone"),
            sessions: Vec::new(),
            worktree_service_mode: Some(WorktreeServiceMode::Manual),
            managed_worktree_profiles: Some(WorktreeProfileInventory {
                profiles: Vec::new(),
            }),
        };
        let projected = C2NodeResponse::from(
            &NodeResponse::StandaloneWorkspaceCreated {
                workspace: workspace.clone(),
            },
        );
        let encoded = serde_json::to_string(&projected).unwrap();
        let decoded = serde_json::from_str::<C2NodeResponse>(&encoded).unwrap();

        assert_eq!(
            decoded,
            C2NodeResponse::StandaloneWorkspaceCreated {
                workspace: C2WorkspaceSnapshot::from(&workspace),
            },
        );
    }

    #[test]
    fn routed_worktree_created_omits_raw_git_reasons() {
        let response = routed_response(NodeResponse::WorktreeCreated {
            worktree: GitWorktreeSnapshot {
                path: host_path(r"C:\work\feature"),
                head: "abc123".to_owned(),
                branch: Some("feature/privacy".to_owned()),
                is_bare: false,
                is_main: false,
                locked: true,
                lock_reason: Some("created-lock-secret".to_owned()),
                prunable: true,
                prunable_reason: Some("created-prunable-secret".to_owned()),
                workspace_id: Some(WorkspaceId::new("feature").unwrap()),
            },
            workspace: WorkspaceSnapshot {
                workspace_id: WorkspaceId::new("feature").unwrap(),
                canonical_root: host_path(r"C:\work\feature"),
                sessions: Vec::new(),
                worktree_service_mode: None,
                managed_worktree_profiles: None,
            },
        });

        let json = serde_json::to_string(&response).unwrap();
        assert!(!json.contains("created-lock-secret"));
        assert!(!json.contains("created-prunable-secret"));
        assert!(!json.contains("lock_reason"));
        assert!(!json.contains("prunable_reason"));
        let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
        assert!(matches!(decoded.response, Ok(C2NodeResponse::WorktreeCreated {
            worktree: C2GitWorktreeSnapshot { locked: true, prunable: true, .. },
            ..
        })));
    }

    #[test]
    fn routed_resume_authorized_control_event_omits_provider_session_identity() {
        let routed = routed_control_event(ControlEventKind::ResumeAuthorized {
            session: ResumeSessionSummary {
                key: ProviderSessionKey::SessionId,
                id: "private-resume-session-id".to_owned(),
            },
        });

        let json = serde_json::to_string(&routed).unwrap();
        assert!(json.contains("resume-authorized"));
        assert!(!json.contains("private-resume-session-id"));
        assert!(!json.contains("provider_session"));
        assert!(!json.contains("transcript_path"));
        let decoded = serde_json::from_str::<RoutedNodeEvent>(&json).unwrap();
        assert!(matches!(decoded.event, C2NodeEvent::Control {
            event: C2ControlEvent { event: C2ControlEventKind::ResumeAuthorized, .. },
            ..
        }));
    }

    #[test]
    fn routed_session_identity_observed_control_event_omits_provider_identity_and_path() {
        let routed = routed_control_event(ControlEventKind::ProviderEvent {
            sequence: 19,
            source: provider_source(),
            source_sequence: 7,
            event: ProviderEvent::SessionIdentityObserved {
                identity: ProviderSessionIdentity {
                    key: ProviderSessionKey::ConversationId,
                    id: "private-observed-provider-id".to_owned(),
                    transcript_path: Some(r"C:\private\provider-transcript.jsonl".to_owned()),
                },
            },
        });

        let json = serde_json::to_string(&routed).unwrap();
        assert!(json.contains("session-identity-observed"));
        assert!(!json.contains("private-observed-provider-id"));
        assert!(!json.contains("provider-transcript.jsonl"));
        assert!(!json.contains("transcript_path"));
        assert!(!json.contains("\"identity\":"));
        let decoded = serde_json::from_str::<RoutedNodeEvent>(&json).unwrap();
        assert!(matches!(decoded.event, C2NodeEvent::Control {
            event: C2ControlEvent {
                event: C2ControlEventKind::ProviderEvent {
                    event: C2ProviderEventKind::SessionIdentityObserved,
                },
                ..
            },
            ..
        }));
    }

    #[test]
    fn routed_provider_interaction_resolution_is_categorical_only() {
        let routed = routed_control_event(ControlEventKind::ProviderEvent {
            sequence: 20,
            source: provider_source(),
            source_sequence: 8,
            event: ProviderEvent::InteractionResolved {
                request_id: "private-provider-request-id".to_owned(),
                outcome: ProviderInteractionOutcome::Denied,
            },
        });

        let json = serde_json::to_string(&routed).unwrap();
        assert!(json.contains("interaction-resolved"));
        assert!(!json.contains("private-provider-request-id"));
        assert!(!json.contains("denied"));
        let decoded = serde_json::from_str::<RoutedNodeEvent>(&json).unwrap();
        assert!(matches!(decoded.event, C2NodeEvent::Control {
            event: C2ControlEvent {
                event: C2ControlEventKind::ProviderEvent {
                    event: C2ProviderEventKind::InteractionResolved,
                },
                ..
            },
            ..
        }));
    }

    #[test]
    fn routed_snapshot_recursively_omits_provider_identity_and_error_state() {
        let mut session = fixture_session();
        session.status = SessionStatus::Failed {
            message: "private-session-failure".to_owned(),
        };
        session.terminal_stale = Some("private-terminal-error".to_owned());
        session.history.last_error = Some("private-history-error".to_owned());
        session.resume.last_session = Some(ResumeSessionSummary {
            key: ProviderSessionKey::SessionId,
            id: "private-resume-summary-id".to_owned(),
        });
        session.resume.last_error = Some("private-resume-error".to_owned());
        session.foreground.stale_reason = Some("private-foreground-error".to_owned());
        session.provider.session = Some(ProviderSessionIdentity {
            key: ProviderSessionKey::SessionId,
            id: "private-snapshot-provider-id".to_owned(),
            transcript_path: Some(r"C:\private\snapshot-transcript.jsonl".to_owned()),
        });
        session.provider.current_prompt = Some("private-current-prompt".to_owned());
        session.provider.last_event = Some(ProviderEvent::Error {
            message: "private-provider-error".to_owned(),
        });
        let routed = routed_response(NodeResponse::Snapshot {
            event_sequence: 4,
            controller: None,
            snapshot: NodeSnapshot {
                node_id: NodeId::new("node-a").unwrap(),
                enabled_providers: vec![provider("codex")],
                provider_runtime_statuses: ProviderRuntimeStatuses::default(),
                workspaces: vec![WorkspaceSnapshot {
                    workspace_id: WorkspaceId::new("primary").unwrap(),
                    canonical_root: host_path(r"C:\workspace"),
                    sessions: vec![session],
                    worktree_service_mode: None,
                    managed_worktree_profiles: None,
                }],
                session_records: Vec::new(),
                managed_worktrees: Vec::new(),
                launch_inventory: None,
                agent_progress: Vec::new(),
            },
        });

        let json = serde_json::to_string(&routed).unwrap();
        for secret in [
            "private-session-failure",
            "private-terminal-error",
            "private-history-error",
            "private-resume-summary-id",
            "private-resume-error",
            "private-foreground-error",
            "private-snapshot-provider-id",
            "snapshot-transcript.jsonl",
            "private-current-prompt",
            "private-provider-error",
        ] {
            assert!(!json.contains(secret));
        }
        assert!(!json.contains("provider_session"));
        assert!(!json.contains("transcript_path"));
        assert!(!json.contains("last_error"));
        let decoded = serde_json::from_str::<RoutedNodeResponse>(&json).unwrap();
        assert!(matches!(decoded.response, Ok(C2NodeResponse::Snapshot { snapshot, .. })
            if matches!(snapshot.workspaces[0].sessions[0].status, C2SessionStatus::Failed)
                && snapshot.workspaces[0].sessions[0].provider_identity_present));
    }

    #[test]
    fn slim_managed_sessions_are_sorted_bounded_and_privacy_minimized() {
        let long_name = "ж".repeat((MAX_C2_SESSION_DISPLAY_NAME_BYTES / 2) + 4);
        let make_record = |record_id: &str, display_name: String| ManagedSessionRecord {
            record_id: SessionRecordId::new(record_id).unwrap(),
            display_name,
            provider: provider("codex"),
            mode: SessionMode::Pty,
            state: ManagedSessionState::Dormant,
            workspace_id: WorkspaceId::new("primary").unwrap(),
            canonical_root: host_path(r"C:\private\workspace"),
            provider_session: Some(ProviderSessionIdentity {
                key: ProviderSessionKey::SessionId,
                id: "5af75a6b-3e64-41dd-96fa-private-provider-id".to_owned(),
                transcript_path: Some(r"C:\private\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: 20,
            last_error: Some("private backend diagnostic".to_owned()),
        };
        let snapshot = NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: vec![provider("codex")],
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: vec![
                make_record("session-z", "z".to_owned()),
                make_record("session-a", long_name),
            ],
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        };

        let slim = SlimNodeInventory::from_snapshot(&snapshot);
        assert_eq!(slim.managed_session_count, 2);
        assert!(!slim.managed_sessions_truncated);
        assert_eq!(
            slim.managed_sessions.iter().map(|record| record.record_id.as_str()).collect::<Vec<_>>(),
            vec!["session-a", "session-z"],
        );
        assert_eq!(slim.managed_sessions[0].display_name.len(), MAX_C2_SESSION_DISPLAY_NAME_BYTES);
        assert!(slim.managed_sessions[0].display_name_truncated);
        assert!(slim.managed_sessions[0].provider_identity_present);

        let json = serde_json::to_string(&slim).unwrap();
        assert!(!json.contains("5af75a6b-3e64-41dd-96fa-private-provider-id"));
        assert!(!json.contains("transcript.jsonl"));
        assert!(!json.contains("private backend diagnostic"));
        assert!(!json.contains("private\\\\workspace"));
        assert!(!json.contains("provider_session"));
        assert!(!json.contains("last_error"));
    }

    #[test]
    fn slim_inventory_defaults_managed_sessions_for_v1_payloads() {
        let legacy = r#"{"node_id":"node-a","enabled_providers":[],"workspaces":{},"workspace_count":0,"workspaces_truncated":false,"session_count":0,"sessions_truncated":false}"#;
        let inventory = serde_json::from_str::<SlimNodeInventory>(legacy).unwrap();
        assert!(inventory.provider_contracts.is_empty());
        assert!(inventory.provider_adapter_contracts.is_empty());
        assert!(inventory.managed_sessions.is_empty());
        assert_eq!(inventory.managed_session_count, 0);
        assert!(!inventory.managed_sessions_truncated);
        let reencoded = serde_json::to_string(&inventory).unwrap();
        assert!(!reencoded.contains("provider_contracts"));
        assert!(!reencoded.contains("provider_adapter_contracts"));
        assert_eq!(inventory.retired_count, 0);
        assert!(!reencoded.contains("retired_count"));
    }

    /// Slice R (`gate4agent-node`'s session-record retention sweep):
    /// `retired_count` is additive, defaults to `0` for a pre-existing
    /// payload that never had it, and round-trips intact once a producer
    /// sets a real, nonzero value.
    #[test]
    fn slim_inventory_retired_count_round_trips_and_defaults_to_zero() {
        let legacy = r#"{"node_id":"node-a","enabled_providers":[],"workspaces":{},"workspace_count":0,"workspaces_truncated":false,"session_count":0,"sessions_truncated":false}"#;
        let inventory = serde_json::from_str::<SlimNodeInventory>(legacy).unwrap();
        assert_eq!(inventory.retired_count, 0);

        let mut inventory = inventory;
        inventory.retired_count = 7;
        let encoded = serde_json::to_string(&inventory).unwrap();
        assert!(encoded.contains("\"retired_count\":7"));
        let decoded: SlimNodeInventory = serde_json::from_str(&encoded).unwrap();
        assert_eq!(decoded.retired_count, 7);
    }

    #[test]
    fn slim_inventory_reports_managed_session_truncation() {
        let session_records = (0..=MAX_C2_MANAGED_SESSIONS_PER_NODE)
            .map(|index| ManagedSessionRecord {
                record_id: SessionRecordId::new(format!("session-{index:03}")).unwrap(),
                display_name: format!("session {index}"),
                provider: provider("claude"),
                mode: SessionMode::Inline,
                state: ManagedSessionState::Unavailable,
                workspace_id: WorkspaceId::new("primary").unwrap(),
                canonical_root: host_path(r"C:\repo"),
                provider_session: None,
                active_session: None,
                environment_profile: None,
                bundle: None,
                context_id: None,
                context: None,
                exported_context: None,
                task_binding: None,
                created_at_unix_ms: index as u64,
                updated_at_unix_ms: index as u64,
                last_error: None,
            })
            .collect::<Vec<_>>();
        let slim = SlimNodeInventory::from_snapshot(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records,
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        assert_eq!(slim.managed_session_count, MAX_C2_MANAGED_SESSIONS_PER_NODE + 1);
        assert_eq!(slim.managed_sessions.len(), MAX_C2_MANAGED_SESSIONS_PER_NODE);
        assert!(slim.managed_sessions_truncated);
    }

    #[test]
    fn slim_managed_sessions_page_keeps_the_live_record_even_when_it_sorts_last_by_id() {
        // 130 stale `Unavailable` records (ids sort ahead of both records
        // below) plus one `Live` record and one `IdentityPending` record
        // whose ids sort LAST of all -- the `IdentityPending` one sorts last
        // by id even among the `Unavailable` records. An id-only sort
        // followed by `.take(MAX_C2_MANAGED_SESSIONS_PER_NODE)` would cut
        // both the live record and the just-spawned, identity-pending one
        // from the page; liveness-first ordering must keep both in,
        // `IdentityPending` right behind `Live` and ahead of every
        // `Unavailable` record, regardless of where their ids fall.
        let stale_count = MAX_C2_MANAGED_SESSIONS_PER_NODE + 2;
        let mut session_records = (0..stale_count)
            .map(|index| ManagedSessionRecord {
                record_id: SessionRecordId::new(format!("session-{index:03}")).unwrap(),
                display_name: format!("session {index}"),
                provider: provider("claude"),
                mode: SessionMode::Inline,
                state: ManagedSessionState::Unavailable,
                workspace_id: WorkspaceId::new("primary").unwrap(),
                canonical_root: host_path(r"C:\repo"),
                provider_session: None,
                active_session: None,
                environment_profile: None,
                bundle: None,
                context_id: None,
                context: None,
                exported_context: None,
                task_binding: None,
                created_at_unix_ms: index as u64,
                updated_at_unix_ms: index as u64,
                last_error: None,
            })
            .collect::<Vec<_>>();
        session_records.push(ManagedSessionRecord {
            record_id: SessionRecordId::new(format!("session-{stale_count:03}")).unwrap(),
            display_name: "live-run".to_owned(),
            provider: provider("claude"),
            mode: SessionMode::Pty,
            state: ManagedSessionState::Live,
            workspace_id: WorkspaceId::new("primary").unwrap(),
            canonical_root: host_path(r"C:\repo"),
            provider_session: None,
            active_session: None,
            environment_profile: None,
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            created_at_unix_ms: stale_count as u64,
            updated_at_unix_ms: stale_count as u64,
            last_error: None,
        });
        let pending_index = stale_count + 1;
        session_records.push(ManagedSessionRecord {
            record_id: SessionRecordId::new(format!("session-{pending_index:03}")).unwrap(),
            display_name: "pending-spawn".to_owned(),
            provider: provider("claude"),
            mode: SessionMode::Pty,
            state: ManagedSessionState::IdentityPending,
            workspace_id: WorkspaceId::new("primary").unwrap(),
            canonical_root: host_path(r"C:\repo"),
            provider_session: None,
            active_session: None,
            environment_profile: None,
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            created_at_unix_ms: pending_index as u64,
            updated_at_unix_ms: pending_index as u64,
            last_error: None,
        });
        let total = session_records.len();
        assert!(total > MAX_C2_MANAGED_SESSIONS_PER_NODE);

        let slim = SlimNodeInventory::from_snapshot(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records,
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        });

        assert_eq!(slim.managed_session_count, total);
        assert!(slim.managed_sessions_truncated);
        assert_eq!(slim.managed_sessions.len(), MAX_C2_MANAGED_SESSIONS_PER_NODE);
        assert!(
            slim.managed_sessions.iter().any(|record| record.state == ManagedSessionState::Live),
            "the live record must survive the page cut even though its id sorts last",
        );
        assert!(
            slim.managed_sessions.iter()
                .any(|record| record.state == ManagedSessionState::IdentityPending),
            "the identity-pending record must survive the page cut even though its id sorts last",
        );
        assert_eq!(slim.managed_sessions[0].state, ManagedSessionState::Live);
        assert_eq!(slim.managed_sessions[1].state, ManagedSessionState::IdentityPending);
        assert!(
            slim.managed_sessions[2..].iter().all(|record| record.state == ManagedSessionState::Unavailable),
            "every Unavailable record on the page must rank after Live and IdentityPending",
        );
    }

    #[test]
    fn managed_worktree_projection_is_bounded_and_contains_no_git_or_path_details() {
        fn lease(index: usize) -> ManagedWorktreeLeaseSnapshot {
            ManagedWorktreeLeaseSnapshot {
                lease_id: ManagedWorktreeLeaseId::new(format!("lease-{index}")).unwrap(),
                source_workspace_id: WorkspaceId::new("primary").unwrap(),
                workspace_id: WorkspaceId::new(format!("managed-{index}")).unwrap(),
                profile_id: WorktreeProfileId::new("review").unwrap(),
                profile_revision: WorktreeProfileRevision::new("review.r1").unwrap(),
                retention: ManagedWorktreeRetention::RemoveWhenReleased,
                state: ManagedWorktreeLeaseState::Ready,
                active_session_count: 0,
                managed_record_count: 0,
                cleanup_failure: None,
                created_at_unix_ms: 1,
                updated_at_unix_ms: 2,
            }
        }

        let projected = C2NodeSnapshot::from(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: vec![lease(0)],
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        let json = serde_json::to_string(&projected).unwrap();
        assert!(json.contains("managed_worktrees"));
        for forbidden in ["canonical", "path", "gitdir", "branch", "base_commit", "diagnostic"] {
            assert!(!json.contains(forbidden), "managed projection leaked {forbidden}");
        }

        let overflow = C2NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            agent_progress: Vec::new(),
            managed_worktrees: (0..=MAX_C2_MANAGED_WORKTREES_PER_NODE)
                .map(lease)
                .collect(),
            launch_inventory: None,
        };
        let encoded = serde_json::to_value(overflow).unwrap();
        assert!(serde_json::from_value::<C2NodeSnapshot>(encoded).is_err());
    }
}