bcode_ipc 0.0.1-alpha.0

#![cfg_attr(feature = "fail-on-warnings", deny(warnings))]
#![warn(clippy::all, clippy::pedantic, clippy::nursery, clippy::cargo)]
#![allow(clippy::multiple_crate_versions)]

//! Client/server IPC protocol for bcode.

use bcode_agent_profile::{AgentInfo, PolicyStatusResponse};
use bcode_metrics::MetricsSnapshot;
#[cfg(unix)]
use bcode_plugin_sdk::path::display_from_current_dir;
use bcode_session_models::{
    ClientId, ProjectionWindowRequest, SessionDerivationPromptPage, SessionDerivationPromptQuery,
    SessionDerivationRequest, SessionDerivationSourceSnapshot, SessionDerivationTerminalOutcome,
    SessionEvent, SessionHistoryAroundQuery, SessionHistoryPage, SessionHistoryQuery,
    SessionHistoryWindow, SessionId, SessionInputHistoryEntry, SessionInspectionPage,
    SessionInspectionQuery, SessionLiveEvent, SessionOpenOperationId, SessionOpenOperationSnapshot,
    SessionSummary, WorkId,
};
use bcode_skill_models::{SkillContextResponse, SkillId, SkillList, SkillManifest};
pub use bcode_workflow_view_models::{
    AttemptObservation, SessionProgressSnapshot, WORKFLOW_ACTIVITY_VERSION, WorkflowActivityPage,
    WorkflowActivityPageRequest, WorkflowActivitySection, WorkflowStageProgress,
};
pub use bcode_worktree_models::{
    WorktreeCreateOperationStatus, WorktreeCreateRequest, WorktreeCreateResponse,
    WorktreeListRequest, WorktreeListResponse, WorktreeRemoveRequest, WorktreeRemoveResponse,
};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use sha2::{Digest as _, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::env;
use std::fmt;
#[cfg(unix)]
use std::fs;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use thiserror::Error;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

pub use bmux_ipc::IpcEndpoint;
pub use bmux_ipc::transport::{IpcTransportError, LocalIpcStream};

/// Local IPC listener that avoids unlinking live Unix socket endpoints.
#[derive(Debug)]
pub struct LocalIpcListener {
    inner: bmux_ipc::transport::LocalIpcListener,
}

impl LocalIpcListener {
    /// Bind a local listener for the provided endpoint.
    ///
    /// # Errors
    ///
    /// Returns an error when the endpoint is unsupported on this platform, the
    /// endpoint appears to already have a live listener, stale endpoint cleanup
    /// fails, or the listener cannot be created.
    pub fn bind(endpoint: &IpcEndpoint) -> Result<Self, IpcTransportError> {
        #[cfg(unix)]
        prepare_endpoint_for_bind(endpoint)?;
        match bmux_ipc::transport::LocalIpcListener::bind(endpoint) {
            Ok(inner) => Ok(Self { inner }),
            Err(IpcTransportError::Io(error))
                if error.kind() == std::io::ErrorKind::AlreadyExists =>
            {
                #[cfg(unix)]
                prepare_endpoint_for_bind(endpoint)?;
                Ok(Self {
                    inner: bmux_ipc::transport::LocalIpcListener::bind(endpoint)?,
                })
            }
            Err(error) => Err(error),
        }
    }

    /// Accept an incoming local connection.
    ///
    /// # Errors
    ///
    /// Returns an error when accepting fails.
    pub async fn accept(&self) -> Result<LocalIpcStream, IpcTransportError> {
        self.inner.accept().await
    }
}

const FRAME_LEN_BYTES: usize = 4;

/// Maximum accepted encoded envelope payload size.
pub const MAX_FRAME_PAYLOAD_SIZE: usize = 1_048_576;

const MAX_CHUNK_DATA_SIZE: usize = MAX_FRAME_PAYLOAD_SIZE / 2;

/// Current Bcode IPC protocol version.
///
/// Same-build client/server compatibility is expected. Bump this when IPC DTO
/// enum layouts or envelope payload shapes change incompatibly so stale
/// client/daemon pairs fail explicitly during envelope decode instead of
/// interpreting payloads with mismatched positional layouts.
///
/// Version 30 adds `state_location_id` to `Request::Hello` and `DaemonStatus` so a
/// client can only reach a daemon serving the state location it resolved. The field
/// is `Option` for decode compatibility; a daemon that does not advertise an
/// identity is treated as unverifiable and rejected rather than assumed to match.
///
/// Version 31 widens that identity from the state root alone to the runtime scope
/// pair `(state root, config directory)`. Two config directories can select
/// different plugins and permission policy, so they must not share a daemon. The
/// field name is unchanged; only its derivation widened, and a stale peer computing
/// the narrower identity now mismatches and is refused rather than silently sharing
/// a daemon across config directories.
/// Version 35 includes the selected edge revision in workflow graph inspection.
/// Older positional payloads are rejected rather than assigned a guessed revision.
/// Version 38 restores pre-storage-usage positional request/response tags by appending
/// storage-usage variants. Version 37 peers are rejected rather than misdecoded.
/// Version 39 adds bounded usage reporting and explicit snapshot collection.
/// Version 40 adds bounded native usage catalog discovery.
/// Version 41 combines session usage and publication acceptance with distinct positional tags.
/// Version 42 adds revision-pinned scheduling pages; older positional layouts reject.
/// Version 43 adds ordinary child creation, exact non-owning metadata, and bounded activity;
/// changed positional layouts and parent-bearing summaries reject older peers.
/// Version 44 adds explicit child relationship maintenance and own-session activity evidence.
/// Version 45 adds optional judgement, verification, and advisory session categorization;
/// changed positional layouts reject older peers.
/// Version 46 adds explicit receipt-backed workflow diagnostic ranges. Older positional
/// request/response layouts are rejected rather than interpreted as another operation.
/// Version 47 adds read-only projection history pages to both positional envelopes.
/// Older peers are rejected rather than interpreting the combined payload layout.
/// Version 48 adds scoped session discovery and bounded catalog pages.
/// Version 49 adds content-free session progress snapshots; version 48 peers are rejected.
/// Version 50 adds retained workflow activity snapshots and host progress timestamps.
/// Version 51 adds bounded request-only turn judgement observations to that layout.
/// Version 52 adds bounded cross-location session discovery.
/// Version 53 adds contextual plugin invocation, cancellation, composer discovery and agent hints.
/// Version 54 combines those contextual additions with the updated upstream positional layout.
/// Version 55 adds secret-free queued-submission recovery confirmation and targeted operations.
/// Older positional request/response layouts are rejected rather than misinterpreted.
pub const CURRENT_PROTOCOL_VERSION: u16 = 55;

/// Durable session-storage writer epoch expected by this IPC build.
pub const CURRENT_SESSION_STORAGE_WRITER_EPOCH: u32 =
    bcode_session_models::CURRENT_SESSION_STORAGE_WRITER_EPOCH;

/// Build-scoped daemon fingerprint generated at compile time.
pub const BUILD_FINGERPRINT: &str = env!("BCODE_BUILD_FINGERPRINT");

/// Producer label embedded at compilation; raw builds resolve final-byte identity at runtime.
/// Use [`ArtifactId::current`] for routing rather than this label.
pub const ARTIFACT_ID: &str = env!("BCODE_ARTIFACT_ID");

/// Exact produced-artifact identity used for daemon routing.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct ArtifactId(String);

impl ArtifactId {
    /// Parse an exact artifact identity.
    ///
    /// # Errors
    ///
    /// Returns an error when the identity is empty, too long, or contains unsupported bytes.
    pub fn parse(value: impl Into<String>) -> Result<Self, ArtifactIdParseError> {
        let value = value.into();
        if value.is_empty() {
            return Err(ArtifactIdParseError::Empty);
        }
        if value.len() > 128 {
            return Err(ArtifactIdParseError::TooLong);
        }
        if !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
        {
            return Err(ArtifactIdParseError::InvalidCharacter);
        }
        Ok(Self(value))
    }

    /// Return the identity of the current produced artifact.
    ///
    /// Raw Cargo builds resolve identity from final executable bytes once per process:
    /// a dependency build script cannot observe final-target-only relinks. Release
    /// automation instead supplies an explicit, unique embedded identity.
    ///
    /// # Panics
    ///
    /// Panics if a raw build's executable cannot be located or read. An unverifiable
    /// executable must not fall back to a potentially shared build-script label.
    #[must_use]
    pub fn current() -> Self {
        static CURRENT: std::sync::OnceLock<ArtifactId> = std::sync::OnceLock::new();
        CURRENT
            .get_or_init(|| {
                if env!("BCODE_ARTIFACT_ID_EXPLICIT") == "true" {
                    return Self(ARTIFACT_ID.to_owned());
                }
                let path = std::env::current_exe().expect("locate raw-build executable identity");
                let mut file =
                    std::fs::File::open(path).expect("open raw-build executable identity");
                Self::for_executable(&mut file).expect("read raw-build executable identity")
            })
            .clone()
    }

    /// Resolve identity for a retained executable handle.
    ///
    /// Explicit producer IDs are preserved; raw builds hash the supplied bytes.
    /// The reader is consumed from its current position.
    ///
    /// # Errors
    ///
    /// Returns an I/O error if raw executable bytes cannot be read.
    pub fn for_executable(reader: &mut impl std::io::Read) -> std::io::Result<Self> {
        if env!("BCODE_ARTIFACT_ID_EXPLICIT") == "true" {
            return Ok(Self(ARTIFACT_ID.to_owned()));
        }
        Self::from_executable(reader)
    }

    fn from_executable(reader: &mut impl std::io::Read) -> std::io::Result<Self> {
        let mut digest = Sha256::new();
        let mut buffer = vec![0_u8; 65536].into_boxed_slice();
        loop {
            let read = reader.read(&mut buffer)?;
            if read == 0 {
                break;
            }
            digest.update(&buffer[..read]);
        }
        Ok(Self(format!(
            "cargo-sha256-{}",
            hex::encode(digest.finalize())
        )))
    }

    /// Return the artifact identity as text.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for ArtifactId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl FromStr for ArtifactId {
    type Err = ArtifactIdParseError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        Self::parse(value)
    }
}

/// Invalid exact artifact identity.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
pub enum ArtifactIdParseError {
    /// Identity was empty.
    #[error("artifact identity cannot be empty")]
    Empty,
    /// Identity exceeded the contract limit.
    #[error("artifact identity is longer than 128 bytes")]
    TooLong,
    /// Identity contained a byte outside the portable identifier alphabet.
    #[error("artifact identity must contain only ASCII letters, digits, '-' or '_'")]
    InvalidCharacter,
}

/// Protocol version used in IPC envelopes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct ProtocolVersion(pub u16);

impl ProtocolVersion {
    /// The currently supported protocol version.
    #[must_use]
    pub const fn current() -> Self {
        Self(CURRENT_PROTOCOL_VERSION)
    }
}

impl Default for ProtocolVersion {
    fn default() -> Self {
        Self::current()
    }
}

pub use bcode_session_models::{MessageAcceptanceDisposition, PromptPlacement};

/// Envelope discriminant for payload interpretation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EnvelopeKind {
    Request,
    Response,
    Event,
    /// Internal continuation frame for logical envelopes that exceed one IPC frame.
    Chunk,
    /// Internal compressed frame for large logical envelopes.
    Compressed,
}

/// Versioned IPC envelope with request correlation support.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Envelope {
    pub version: ProtocolVersion,
    pub request_id: u64,
    pub kind: EnvelopeKind,
    pub payload: Vec<u8>,
}

impl Envelope {
    /// Build a new envelope.
    #[must_use]
    pub const fn new(request_id: u64, kind: EnvelopeKind, payload: Vec<u8>) -> Self {
        Self {
            version: ProtocolVersion::current(),
            request_id,
            kind,
            payload,
        }
    }
}

/// Compatibility export for the session-owned composer draft scope.
pub use bcode_session_models::ComposerDraftScope;

/// Maximum per-session outcomes retained in one bulk migration operation response.
pub const MAX_SESSION_BULK_MIGRATION_OUTCOMES: usize = 256;
/// Confirmation required before bulk canonical mutation starts.
pub const SESSION_BULK_MIGRATION_CONFIRMATION: &str = "migrate-supported-sessions";

/// Explicit bulk migration execution mode.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionBulkMigrationMode {
    Inventory,
    Migrate,
}

/// Bounded selection and execution request for bulk canonical migration.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionBulkMigrationStartRequest {
    pub mode: SessionBulkMigrationMode,
    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
    pub session_ids: BTreeSet<SessionId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after_timestamp_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_timestamp_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub confirmation: Option<String>,
}

/// Transient aggregate bulk migration state.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionBulkMigrationState {
    Running,
    CancellationRequested,
    Completed,
    NeedsAttention,
    Cancelled,
}

/// Bounded terminal result for one selected session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionBulkMigrationOutcome {
    pub session_id: SessionId,
    pub category: SessionCompatibilityCategory,
    pub action: SessionCompatibilityAction,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

/// Addressable but transient aggregate bulk migration operation snapshot.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionBulkMigrationOperationStatus {
    pub operation_id: String,
    pub revision: u64,
    pub state: SessionBulkMigrationState,
    pub mode: SessionBulkMigrationMode,
    pub selected: u64,
    pub visited: u64,
    pub migrated: u64,
    pub blocked: u64,
    pub failed: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub current_session_id: Option<SessionId>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub outcomes: Vec<SessionBulkMigrationOutcome>,
}

/// Maximum entries returned by one bounded compatibility inventory page.
pub const MAX_SESSION_COMPATIBILITY_INVENTORY_ENTRIES: usize = 256;
/// Maximum sample entries retained for one compatibility category.
pub const MAX_SESSION_COMPATIBILITY_SAMPLES_PER_CATEGORY: usize = 3;
/// Maximum byte length for one portable compatibility diagnostic.
pub const MAX_SESSION_COMPATIBILITY_MESSAGE_BYTES: usize = 1_024;

/// Stable compatibility category for one canonical session.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionCompatibilityCategory {
    Ready,
    MigrationRequired,
    OwnerBlocked,
    TemporarilyLocked,
    RepairRequired,
    FormatIncompatible,
    Missing,
}

/// Stable action associated with a compatibility category.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionCompatibilityAction {
    None,
    Retry,
    Migrate,
    Repair,
    Upgrade,
    Locate,
}

/// Exclusive cursor for compatibility inventory pagination.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionCompatibilityCursor {
    pub updated_at_ms: u64,
    pub session_id: SessionId,
}

/// Bounded compatibility inventory request.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionCompatibilityInventoryRequest {
    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
    pub session_ids: BTreeSet<SessionId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after_timestamp_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_timestamp_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<SessionCompatibilityCursor>,
    pub limit: usize,
}

/// One compatibility inventory result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionCompatibilityEntry {
    pub session_id: SessionId,
    pub updated_at_ms: u64,
    pub category: SessionCompatibilityCategory,
    pub action: SessionCompatibilityAction,
    pub retryable: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_writer_epoch: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub first_historical_event_schema: Option<u16>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

/// Bounded aggregate for one compatibility category in this page.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionCompatibilityCategorySummary {
    pub count: usize,
    pub action: SessionCompatibilityAction,
    pub retryable: bool,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub samples: Vec<SessionCompatibilitySample>,
}

/// Capped portable sample explaining one aggregate compatibility category.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionCompatibilitySample {
    pub session_id: SessionId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

/// Bounded compatibility inventory page and aggregate counts for that page.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionCompatibilityInventoryResponse {
    pub entries: Vec<SessionCompatibilityEntry>,
    pub counts: BTreeMap<SessionCompatibilityCategory, usize>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub category_summaries:
        BTreeMap<SessionCompatibilityCategory, SessionCompatibilityCategorySummary>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<SessionCompatibilityCursor>,
}

/// Normalized result of explicitly requested optional milestone generation.
/// Failure text is diagnostic presentation, never provider-private data.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionMilestoneGenerationOutcome {
    /// Whether the optional generation operation succeeded (including a no-milestone result).
    pub success: bool,
    /// Bounded human-readable outcome.
    pub text: String,
}

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

    #[test]
    fn milestone_reads_and_generation_have_distinct_wire_contracts() {
        let session_id = SessionId::new();
        for request in [
            Request::SessionMilestones {
                session_id,
                query: bcode_session_models::SessionMilestoneQuery {
                    after_sequence: Some(42),
                    limit: 2,
                },
            },
            Request::GenerateSessionMilestone { session_id },
        ] {
            let bytes = serde_json::to_vec(&request).unwrap();
            assert_eq!(serde_json::from_slice::<Request>(&bytes).unwrap(), request);
        }
        let response = ResponsePayload::SessionMilestoneGenerated {
            result: SessionMilestoneGenerationOutcome {
                success: false,
                text: "generation disabled".to_owned(),
            },
        };
        let bytes = serde_json::to_vec(&response).unwrap();
        assert_eq!(
            serde_json::from_slice::<ResponsePayload>(&bytes).unwrap(),
            response
        );
    }
}

/// Request payload variants for Bcode client/server IPC.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Request {
    Hello {
        client_name: String,
        #[serde(default)]
        runtime_context: Option<ClientRuntimeContext>,
        #[serde(default)]
        daemon_namespace: String,
        #[serde(default)]
        artifact_id: Option<ArtifactId>,
        #[serde(default)]
        build_fingerprint: String,
        /// Identity of the durable state location this client resolved.
        #[serde(default)]
        state_location_id: Option<String>,
    },
    Ping,
    ServerStatus {
        /// Client working directory used to scope repository-local status.
        #[serde(default)]
        working_directory: Option<PathBuf>,
    },
    ServerStop {
        #[serde(default)]
        mode: ServerStopMode,
    },
    CreateSession {
        name: Option<String>,
        working_directory: PathBuf,
    },
    ListSessions {
        working_directory: PathBuf,
    },
    /// Explicit selection; never changes an attached session's cwd.
    ListSessionsWithScope {
        scope: bcode_session_models::SessionDiscoveryScope,
    },
    /// Bounded catalog candidate window with restartable continuation.
    ListSessionsPageWithScope {
        scope: bcode_session_models::SessionDiscoveryScope,
        query: bcode_session_models::SessionCatalogQuery,
        cursor: Option<bcode_session_models::SessionCatalogCursor>,
        limit: usize,
    },
    RenameSession {
        session_id: SessionId,
        name: Option<String>,
    },
    DeleteSession {
        session_id: SessionId,
    },
    /// Explicit complete-history request for export/debug/history commands only.
    ///
    /// This request may force the server to read every canonical event for the session.
    /// Normal runtime flows must use `SessionHistoryPage`, projection-window requests, or
    /// typed read-model endpoints instead.
    SessionHistory {
        session_id: SessionId,
    },
    SessionHistoryPage {
        session_id: SessionId,
        query: SessionHistoryQuery,
    },
    /// Read a bounded projection-selected page without attaching a session.
    SessionProjectionHistoryPage {
        session_id: SessionId,
        request: Box<ProjectionWindowRequest>,
    },
    AttachSession {
        session_id: SessionId,
    },
    AttachSessionRecent {
        session_id: SessionId,
        limit: usize,
    },
    SendUserMessage {
        session_id: SessionId,
        text: String,
    },
    SendUserMessageWithPlacement {
        session_id: SessionId,
        text: String,
        placement: PromptPlacement,
    },
    /// Send an ordinary user message with immutable execution options for its admitted turn.
    SendUserMessageWithExecution {
        session_id: SessionId,
        text: String,
        placement: PromptPlacement,
        execution: bcode_session_models::TurnExecutionOptions,
    },
    /// Accept a conversational submission without promising independent turn execution.
    SubmitPrompt {
        session_id: SessionId,
        submission_id: bcode_session_models::SubmissionId,
        text: String,
        placement: bcode_session_models::SubmissionPlacement,
        admission: bcode_session_models::TurnAdmissionMetadata,
        /// Reject before acceptance unless steering can target a compatible active turn.
        #[serde(default)]
        require_active_turn: bool,
    },
    /// Read one durable submission through a bounded indexed lookup.
    GetSubmission {
        session_id: SessionId,
        submission_id: bcode_session_models::SubmissionId,
    },
    /// Read one bounded acceptance-ordered page of pending input without recovery or replay.
    PendingSubmissions {
        session_id: SessionId,
        /// Exclusive accepted-event sequence. Rescan without this cursor for earlier items.
        after: Option<u64>,
        /// Page size from 1 through 128.
        limit: u32,
    },
    /// Read the oldest queued item and its blocker without recovering or replaying work.
    SubmissionQueueStatus {
        session_id: SessionId,
    },
    /// Prepare a secret-free execution recipe for explicit caller confirmation.
    PrepareSubmissionResume {
        session_id: SessionId,
        submission_id: bcode_session_models::SubmissionId,
        expected_accepted_sequence: u64,
    },
    /// Authorize recovery of this exact pending item using the confirmed recipe.
    ResumeSubmission {
        session_id: SessionId,
        confirmation: bcode_session_models::SubmissionResumeConfirmation,
        operation_id: String,
    },
    /// Cancel only this exact pending item; never cancel active work or clear the queue.
    CancelPendingSubmission {
        session_id: SessionId,
        submission_id: bcode_session_models::SubmissionId,
        expected_accepted_sequence: u64,
        operation_id: String,
    },
    /// Submit an ordinary turn through generic admission metadata.
    SubmitTurn {
        session_id: SessionId,
        text: String,
        admission: bcode_session_models::TurnAdmissionMetadata,
    },
    InvokeSkill {
        session_id: SessionId,
        skill_id: SkillId,
        arguments: String,
        display_text: String,
    },
    /// Invoke a skill with immutable execution options for its admitted turn.
    InvokeSkillWithExecution {
        session_id: SessionId,
        skill_id: SkillId,
        arguments: String,
        display_text: String,
        execution: bcode_session_models::TurnExecutionOptions,
    },
    CancelSessionTurn {
        session_id: SessionId,
        /// When supplied, cancel only this turn; never clear unrelated queued work.
        #[serde(default)]
        expected_turn_id: Option<String>,
        #[serde(default)]
        clear_queue: bool,
    },
    CancelRuntimeWork {
        session_id: SessionId,
        work_id: WorkId,
    },
    /// Atomically create one logical workflow and its initial mutable draft.
    CreateAuthoredWorkflow(CreateAuthoredWorkflowRequest),
    /// Apply one already-lowered portable source through a single daemon-owned application operation.
    ApplyWorkflowSource(ApplyWorkflowSourceRequest),
    /// Cancel one exact in-flight authored-workflow validation or compilation operation.
    CancelWorkflowComputation {
        operation_id: String,
    },
    /// Explicitly reset an incompatible workflow store after destructive confirmation.
    ResetIncompatibleWorkflowStore {
        /// Must equal the documented destructive confirmation token.
        confirm: String,
    },
    /// Apply one atomic semantic edit batch to an exact draft generation.
    ApplyWorkflowDraftEdits(ApplyWorkflowDraftEditsRequest),
    /// Replace one exact draft generation.
    UpdateWorkflowDraft(UpdateWorkflowDraftRequest),
    /// Publish one exact draft generation, optionally activating it atomically.
    PublishWorkflowDraft(PublishWorkflowDraftRequest),
    /// Publish one exact draft and then independently attempt durable run admission.
    PublishAndStartWorkflow(Box<PublishAndStartWorkflowRequest>),
    /// Compare-and-set one immutable revision as active.
    ActivateWorkflowRevision(ActivateWorkflowRevisionRequest),
    /// Archive or unarchive one logical workflow.
    SetAuthoredWorkflowArchived(SetAuthoredWorkflowArchivedRequest),
    /// Discard one exact mutable draft generation.
    DiscardWorkflowDraft(DiscardWorkflowDraftRequest),
    /// Fork one exact draft or revision into a new mutable draft.
    ForkWorkflowDraft(ForkWorkflowDraftRequest),
    /// Create one revision-bound preset.
    CreateWorkflowPreset(CreateWorkflowPresetRequest),
    /// Replace one exact preset generation.
    UpdateWorkflowPreset(UpdateWorkflowPresetRequest),
    /// Delete one exact preset generation.
    DeleteWorkflowPreset(DeleteWorkflowPresetRequest),
    /// Export one exact immutable authored revision.
    ExportWorkflowRevision(ExportWorkflowRevisionRequest),
    /// Preview one portable import without mutation.
    PreviewWorkflowImport(PreviewWorkflowImportRequest),
    /// Import one portable bundle as a new logical workflow and draft.
    ImportWorkflow(ImportWorkflowRequest),
    /// Import one portable bundle as a new draft in an existing logical workflow.
    ImportWorkflowDraft(ImportWorkflowDraftRequest),
    /// Import one portable bundle as the exact next immutable revision of an existing workflow.
    ImportWorkflowRevision(ImportWorkflowRevisionRequest),
    /// Resolve and start one immutable authored-workflow revision.
    StartAuthoredWorkflow(StartAuthoredWorkflowRequest),
    /// Resolve one exact published package export and start its authored revision.
    StartWorkflowPackageExport(StartWorkflowPackageExportRequest),
    /// List bounded logical authored workflows.
    ListAuthoredWorkflows {
        #[serde(default)]
        cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
        limit: usize,
    },
    /// Get one logical authored workflow.
    GetAuthoredWorkflow {
        workflow_id: String,
    },
    /// Return one bounded aggregate authored-workflow inspection snapshot.
    InspectAuthoredWorkflow {
        workflow_id: String,
        limit: usize,
    },
    /// List bounded drafts for one logical workflow.
    ListWorkflowDrafts {
        workflow_id: String,
        #[serde(default)]
        cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
        limit: usize,
    },
    /// Get one exact mutable draft.
    GetWorkflowDraft {
        workflow_id: String,
        draft_id: String,
    },
    /// List bounded immutable published revisions.
    ListWorkflowRevisions {
        workflow_id: String,
        #[serde(default)]
        cursor: Option<bcode_workflow::WorkflowRevisionListCursor>,
        limit: usize,
    },
    /// Get one exact immutable published revision.
    GetWorkflowRevision {
        workflow_id: String,
        revision: u64,
    },
    /// Inspect one immutable revision together with current, non-canonical requirement availability.
    InspectWorkflowRevisionRequirements {
        workflow_id: String,
        revision: u64,
    },
    /// List bounded revision-bound presets.
    ListWorkflowPresets {
        workflow_id: String,
        #[serde(default)]
        cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
        limit: usize,
    },
    /// Get one exact revision-bound preset.
    GetWorkflowPreset {
        workflow_id: String,
        preset_id: String,
    },
    /// Return the portable runtime-workflow authoring catalog.
    WorkflowAuthoringCatalog,
    /// Release a retained discovery continuation; never cancels workflow execution.
    CancelWorkflowDiscovery {
        token: String,
    },
    /// Discover one bounded page of launchable workflow sources for a workspace.
    WorkflowLaunchCatalog(bcode_workflow::WorkflowLaunchCatalogRequest),
    /// Inspect one exact launch source without mutation.
    WorkflowLaunchDetail(bcode_workflow::WorkflowLaunchDetailRequest),
    /// Read one bounded derived package publication receipt without mutation.
    GetWorkflowPackagePublication {
        package_id: String,
    },
    /// Atomically apply one previously validated package plan as canonical package drafts.
    ApplyWorkflowPackage(ApplyWorkflowPackageRequest),
    /// Atomically publish every exact package draft generation.
    PublishWorkflowPackage(PublishWorkflowPackageRequest),
    /// Validate and plan one bounded workflow package through the daemon-owned catalog.
    ValidateWorkflowPackage(WorkflowPackageComputationRequest),
    /// Compile-preview one complete package plan through the daemon-owned catalog.
    PreviewWorkflowPackage(WorkflowPackagePreviewRequest),
    /// Validate and lower one raw workflow source through the daemon-owned catalog.
    ValidateWorkflowSource(WorkflowSourceComputationRequest),
    /// Lower and compile-preview one raw workflow source through the daemon-owned catalog.
    PreviewWorkflowSource(WorkflowSourcePreviewRequest),
    /// Validate one portable workflow authoring document without mutation.
    ValidateWorkflowAuthoring {
        document: bcode_workflow::WorkflowAuthoringDocument,
        #[serde(default)]
        control: WorkflowComputationControl,
    },
    /// Compile and preview one authored workflow without mutation or dispatch.
    PreviewWorkflowCompilation {
        document: bcode_workflow::WorkflowAuthoringDocument,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        configuration: Option<serde_json::Value>,
        #[serde(default)]
        control: WorkflowComputationControl,
    },
    /// List bounded normalized template inspections.
    InspectWorkflowTemplates {
        limit: usize,
    },
    /// Inspect one exact template without plugin loading details.
    InspectWorkflowTemplate {
        owner_plugin_id: String,
        template_id: String,
        template_version: u32,
    },
    /// List bounded plugin-owned workflow templates and requirement diagnostics.
    ListWorkflowTemplates {
        limit: usize,
    },
    /// Describe one exact loaded plugin-owned workflow template.
    DescribeWorkflowTemplate {
        owner_plugin_id: String,
        template_id: String,
        template_version: u32,
    },
    /// Instantiate one external standard authoring-document template as a mutable draft.
    InstantiateWorkflowTemplate(WorkflowTemplateInstantiationRequest),
    /// Register and start one exact loaded template with validated configuration.
    StartWorkflowTemplate(WorkflowTemplateStartRequest),
    /// Register one immutable, structurally validated workflow definition.
    RegisterWorkflowDefinition(WorkflowDefinitionRegistrationRequest),
    /// Register an exact definition and start its bound durable run through one retry-safe request.
    StartWorkflow(WorkflowStartRequest),
    WorkflowContinuationSource {
        run_id: String,
    },
    ContinueWorkflow(bcode_workflow::WorkflowContinuationRequest),
    /// Start one durable workflow from an existing exact definition.
    RequestWorkflowReplacement(bcode_workflow::WorkflowReplacementRequest),
    StartWorkflowRun(WorkflowRunStartRequest),
    /// List bounded, checksum-verified durable workflow definitions.
    ListWorkflowDefinitions {
        limit: usize,
    },
    /// Describe one exact durable workflow definition version.
    DescribeWorkflowDefinition {
        definition_id: String,
        version: u32,
    },
    /// Publish an exact staged edit subject to distinct publication authorization.
    PublishWorkflowRunGraphEdit {
        request: bcode_workflow::WorkflowRunGraphEditBatch,
    },
    /// Persist a candidate edit without publishing or dispatching it.
    StageWorkflowRunGraphEdit {
        request: bcode_workflow::WorkflowRunGraphEditBatch,
    },
    /// Read a bounded run graph page at an expected revision.
    InspectWorkflowRunGraph {
        request: WorkflowRunGraphPageRequest,
    },
    /// Return one bounded aggregate workflow inspection snapshot.
    InspectWorkflowRun {
        run_id: String,
        limit: usize,
    },
    /// Read mixed canonical run identities without legacy-only catalog omission.
    WorkflowIdentityCatalog {
        request: bcode_workflow_view_models::WorkflowIdentityCatalogRequest,
    },
    /// Read one revision-pinned scheduling observation page.
    WorkflowSchedulingPage {
        request: bcode_workflow::WorkflowSchedulingPageRequest,
    },
    /// Return one bounded renderer-neutral workflow run projection.
    WorkflowRunView {
        run_id: String,
        limit: usize,
    },
    /// Return one bounded renderer-neutral workflow run catalog.
    WorkflowCatalogView {
        request: bcode_workflow_view_models::WorkflowCatalogRequest,
    },
    /// Return one bounded durable workflow run summary.
    WorkflowRunStatus {
        run_id: String,
    },
    /// Return the newest run for one exact generic binding key.
    AssociatedWorkflowRun {
        key: WorkflowRunBindingLookup,
    },
    /// Inspect the newest run for one exact generic binding key.
    InspectAssociatedWorkflowRun {
        key: WorkflowRunBindingLookup,
        limit: usize,
    },
    /// Apply one lifecycle transition to the newest run for one exact binding key.
    /// Control one exact run without newest-binding selection.
    ControlWorkflowRun {
        run_id: String,
        action: WorkflowRunControlAction,
    },
    ControlAssociatedWorkflowRun {
        key: WorkflowRunBindingLookup,
        action: WorkflowRunControlAction,
    },
    /// List bounded durable workflow run summaries.
    ListWorkflowRuns {
        limit: usize,
    },
    /// Return bounded canonical validated output values for one workflow run.
    WorkflowRunOutputs {
        run_id: String,
        limit: usize,
    },
    /// Run one bounded non-mutating workflow doctor inspection.
    DoctorWorkflowRun {
        run_id: String,
        limit: usize,
    },
    /// Apply one explicit typed resolution to a repair-required attempt.
    RepairWorkflowAttempt {
        dispatch_identity: String,
        resolution: bcode_workflow::RepairResolution,
    },
    /// Explicitly retry one exact latest failed workflow node attempt.
    RetryWorkflowNode {
        run_id: String,
        node_id: String,
        activation_id: String,
        failed_attempt: u32,
    },
    /// Request durable cancellation of one workflow run.
    CancelWorkflowRun {
        run_id: String,
    },
    /// Explicit maintenance: find nonterminal workflow runs whose recorded coordinator daemon
    /// verifiably ended and, when `apply` is set, reassign each to this daemon and cancel it.
    ///
    /// Runs whose coordinator is still live or unverifiable, and runs that still hold live
    /// attempts on another artifact, are reported as skipped and never mutated.
    ReconcileOrphanedWorkflowRuns {
        apply: bool,
        limit: usize,
    },
    /// Pause one running workflow before further scheduler admission.
    PauseWorkflowRun {
        run_id: String,
    },
    /// Resume one paused workflow for subsequent scheduler admission.
    ResumeWorkflowRun {
        run_id: String,
    },
    /// List bounded durable input/approval waits for one workflow run.
    ListWorkflowWaits {
        run_id: String,
        limit: usize,
    },
    /// Resolve one exact durable input wait with schema-validated JSON.
    ProvideWorkflowInput {
        run_id: String,
        node_id: String,
        activation_id: String,
        value: serde_json::Value,
    },
    /// Resolve one exact durable approval wait.
    ResolveWorkflowApproval {
        run_id: String,
        node_id: String,
        activation_id: String,
        approved: bool,
    },
    /// List bounded pending durable mutation approvals across all workflow runs.
    ListWorkflowMutationApprovalsAll {
        limit: usize,
    },
    /// List bounded pending durable mutation approvals for one workflow run.
    ListWorkflowMutationApprovals {
        run_id: String,
        limit: usize,
    },
    /// Resolve one exact durable mutation approval by stable approval identity.
    ResolveWorkflowMutationApproval {
        approval_id: String,
        decision: bcode_workflow::WorkflowMutationApprovalDecision,
    },
    /// Return one bounded page of durable workflow attempts.
    WorkflowAttemptHistory {
        run_id: String,
        #[serde(default)]
        cursor: Option<bcode_workflow_store::AttemptCursor>,
        limit: usize,
    },
    /// Return one bounded page of durable workflow events.
    WorkflowEventHistory {
        run_id: String,
        #[serde(default)]
        after_sequence: Option<u64>,
        limit: usize,
    },
    /// Bounded versioned attempt semantic history, not legacy opaque events.
    WorkflowCommittedHistory {
        query: bcode_workflow::WorkflowCommittedHistoryQuery,
    },
    WorkflowLifecycleHistory {
        query: bcode_workflow::WorkflowLifecycleHistoryQuery,
    },
    /// Read finite, pinned attempt history, not legacy opaque events.
    /// Read generation-pinned canonical current facts without execution authority.
    WorkflowInventory {
        query: bcode_workflow::WorkflowInventoryQuery,
    },
    WorkflowSemanticHistory {
        query: bcode_workflow::WorkflowSemanticHistoryQuery,
    },
    /// Return finite transactional observations with explicit coverage, not guessed history.
    WorkflowObservations {
        run_id: String,
        after_sequence: Option<u64>,
        through_sequence: Option<u64>,
        limit: usize,
    },
    /// Return bounded workflow live notifications after one canonical sequence.
    WorkflowLiveEventCatchUp {
        after_sequence: u64,
        limit: usize,
    },
    CompactSession {
        session_id: SessionId,
    },
    SetSessionModel {
        session_id: SessionId,
        provider_plugin_id: Option<String>,
        model_id: String,
    },
    SetSessionReasoning {
        session_id: SessionId,
        effort: Option<String>,
        summary: Option<String>,
    },
    SessionModelStatus {
        session_id: SessionId,
    },
    DefaultModelStatus,
    SessionModelList {
        provider_plugin_id: Option<String>,
    },
    /// List models using the effective selection of one attached session.
    SessionModelListForSession {
        session_id: SessionId,
        provider_plugin_id: Option<String>,
    },
    /// Request cancellation; acknowledgment does not establish provider settlement.
    CancelModelCheck {
        check_id: String,
    },
    /// Probe a bounded set of models through the active provider configuration.
    /// Explicit consent is required before consuming provider quota.
    CheckModels {
        check_id: String,
        session_id: Option<SessionId>,
        provider_plugin_id: Option<String>,
        model_ids: Vec<String>,
        timeout_seconds: u64,
        allow_provider_requests: bool,
    },
    /// Inspect enabled judgement models and owned account availability without evaluating.
    JudgementSetup,
    /// Recheck one exact evaluator and account selection against current daemon state.
    JudgementSetupValidate {
        /// Bounded JSON encoding of a portable selection.
        selection_json: String,
    },
    /// Read a disposable bounded observation; never replays or repairs canonical history.
    TurnJudgementObservation {
        session_id: SessionId,
        turn_id: String,
    },
    /// Failure-optional judgement without session mutation.
    OptionalJudge {
        request_json: String,
    },
    /// Select bounded request-only evidence; producers retain admission ownership.
    SelectOptionalContext {
        /// Codec-safe encoding of the portable context selection request.
        request_json: String,
    },
    /// Verify authentication and model readiness for one exact selection.
    JudgementVerify {
        selection_json: String,
    },
    /// Invoke a judgement model without creating or modifying a session turn.
    Judge {
        provider_plugin_id: String,
        auth_profile: String,
        /// Bounded JSON encoding of the portable judgement request (codec-safe).
        request_json: String,
    },
    ListAgents,
    ListSkills,
    DescribeSkill {
        skill_id: SkillId,
    },
    ActivateSkill {
        session_id: SessionId,
        skill_id: SkillId,
    },
    DeactivateSkill {
        session_id: SessionId,
        skill_id: SkillId,
    },
    ActiveSkills {
        session_id: SessionId,
    },
    AgentPolicyStatus,
    SetSessionAgent {
        session_id: SessionId,
        agent_id: String,
    },
    SessionHistoryAround {
        session_id: SessionId,
        query: SessionHistoryAroundQuery,
    },
    ListPermissions,
    ResolvePermission {
        permission_id: String,
        approved: bool,
        #[serde(default)]
        remember: bool,
    },
    AddPermissionRule {
        agent_id: String,
        category: String,
        pattern: String,
        action: String,
    },
    ListPluginServices,
    ListPluginContributions,
    /// Invoke a plugin with daemon-validated workspace context and bounded lifetime.
    InvokePluginServiceContextual {
        context: PluginServiceContext,
        plugin_id: String,
        interface_id: String,
        operation: String,
        payload: Vec<u8>,
    },
    /// Cancel a contextual invocation owned by this connected client.
    CancelPluginServiceInvocation {
        invocation_id: String,
    },
    InvokePluginService {
        plugin_id: String,
        interface_id: String,
        operation: String,
        payload: Vec<u8>,
    },
    CallPluginService {
        interface_id: String,
        operation: String,
        payload: Vec<u8>,
    },
    PublishPluginEvent {
        topic: String,
        payload: Vec<u8>,
    },
    UpdateClientRuntimeContext {
        #[serde(default)]
        runtime_context: Option<ClientRuntimeContext>,
    },
    ChangeSessionWorkingDirectory {
        session_id: SessionId,
        working_directory: PathBuf,
    },
    ListWorktrees(WorktreeListRequest),
    /// Start or recover an idempotent daemon-owned worktree creation operation.
    WorktreeCreateStart {
        operation_id: String,
        request: WorktreeCreateRequest,
    },
    /// Read a worktree creation operation snapshot.
    WorktreeCreateStatus {
        operation_id: String,
    },
    /// Wait for a newer worktree creation operation revision.
    WorktreeCreateWait {
        operation_id: String,
        after_revision: u64,
        timeout_ms: u64,
    },
    RemoveWorktree(WorktreeRemoveRequest),
    RalphStatus(RalphStatusRequest),
    RunRalphLoop(RalphRunRequest),
    CancelRalphLoop(RalphCancelRequest),
    ListRalphRuns(Box<RalphListRunsRequest>),
    ListRalphIterations(Box<RalphListIterationsRequest>),
    ResumeRalphRun(RalphResumeRequest),
    ApproveRalphRun(RalphApproveRequest),
    RalphRunStatus(RalphRunStatusRequest),
    RecordRalphLifecycle(RalphLifecycleRequest),
    /// Classify a session for open and start or join required legacy migration.
    PrepareSessionOpen {
        session_id: SessionId,
    },
    /// Wait for a newer session-open operation snapshot or a bounded timeout.
    WaitSessionOpenProgress {
        session_id: SessionId,
        operation_id: SessionOpenOperationId,
        after_revision: u64,
        timeout_ms: u64,
    },
    ImportExternalSession {
        source_id: String,
        external_session_id: String,
        /// Client working directory used when the imported source has no cwd metadata.
        #[serde(default)]
        working_directory: Option<PathBuf>,
    },
    /// Explicitly invoke configured optional generation; no client prompt is accepted.
    GenerateSessionMilestone {
        session_id: SessionId,
    },
    /// Read one bounded page of canonical milestones without invoking generation.
    SessionMilestones {
        session_id: SessionId,
        query: bcode_session_models::SessionMilestoneQuery,
    },
    /// Capture one bounded source snapshot for generic session derivation.
    SessionDerivationSnapshot {
        session_id: SessionId,
    },
    /// Read one bounded prompt-candidate page pinned to a source generation.
    SessionDerivationPrompts {
        session_id: SessionId,
        query: SessionDerivationPromptQuery,
    },
    /// Execute one versioned generic session derivation request.
    DeriveSession {
        request: Box<SessionDerivationRequest>,
    },
    SessionDerivationStatus {
        operation_id: bcode_session_models::SessionDerivationOperationId,
    },
    CancelSessionDerivation {
        operation_id: bcode_session_models::SessionDerivationOperationId,
    },
    RefreshSessionCatalog {
        #[serde(default)]
        working_directory: Option<PathBuf>,
        #[serde(default)]
        sources: Option<Vec<String>>,
    },
    /// Read a bounded page of locally owned runtime observations.
    RuntimeWorkObservations {
        session_id: SessionId,
        after: Option<WorkId>,
        limit: usize,
    },
    ListRuntimeWork {
        session_id: SessionId,
    },
    RuntimeWorkHistory {
        session_id: SessionId,
        limit: usize,
    },
    SubscribeRuntimeWork {
        session_id: SessionId,
    },
    /// Subscribe this connection to workflow-run change notifications.
    SubscribeWorkflowRuns,
    SubscribeCatalogUpdates,
    AttachSessionProjectionWindow {
        session_id: SessionId,
        request: ProjectionWindowRequest,
    },
    SetComposerDraft {
        scope: ComposerDraftScope,
        text: String,
    },
    ComposerDraft {
        scope: ComposerDraftScope,
    },
    ListPendingToolExchanges,
    ResolveToolExchange {
        exchange_id: String,
        resolution_json: serde_json::Value,
    },
    /// Inspect effective model catalog and refresh state.
    ModelCatalogDiagnostics,
    /// Prepare a pinned request-only context for isolated structured generation.
    PrepareContextGeneration {
        request: bcode_session_models::PrepareContextGeneration,
    },
    /// Explicitly prepare or inspect a bounded mutable session working document.
    SessionWorkingDocument {
        request: bcode_session_models::SessionWorkingDocumentRequest,
    },
    ListWorkflowDiagnosticArtifacts {
        run_id: String,
        dispatch_identity: String,
    },
    /// Explicit sensitive diagnostic access to an immutable workflow dispatch artifact.
    ReadWorkflowDiagnosticArtifact {
        request: bcode_workflow::WorkflowDiagnosticArtifactRequest,
    },
    /// Explicit sensitive diagnostic access to a canonical event's retained trace blob.
    ReadSessionTraceBlob {
        request: bcode_session_models::ReadSessionTraceBlobRequest,
    },
    /// Read a bounded byte range from a generic session artifact reference.
    ReadSessionArtifact {
        session_id: SessionId,
        artifact_id: String,
        reference_key: String,
        #[serde(default)]
        offset: u64,
        length: u32,
    },
    /// Deliver opaque schema-versioned input to an active invocation.
    InvocationInput {
        session_id: SessionId,
        input: bcode_tool::ToolInvocationInput,
    },
    /// Resolve every currently pending checkpoint in one exact authorization batch.
    ResolvePermissionBatch {
        batch_id: String,
        approved: bool,
    },
    /// Submit bounded client-side TUI telemetry to the daemon-owned metrics registry.
    IngestClientMetrics {
        /// Validated-at-ingress metric observations.
        batch: bcode_metrics::ClientMetricBatch,
    },
    /// List portable auth-pool status.
    AuthPoolList,
    /// Persist or clear an interactive preferred auth-pool profile.
    SetAuthPoolPreference {
        pool: String,
        #[serde(default)]
        profile: Option<String>,
    },
    /// Ask this daemon to release runtime ownership when the session is quiescent.
    ReleaseSessionOwnership {
        session_id: SessionId,
    },
    /// Read disposable advisory labels; never replays or modifies history.
    SessionCategorization {
        session_id: SessionId,
    },
    SessionInspection {
        session_id: SessionId,
        query: SessionInspectionQuery,
    },
    SessionSearch {
        request: bcode_session_search::SessionSearchRequest,
        #[serde(default)]
        policy: bcode_session_search::SessionSearchPlanPolicy,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        routes: Vec<bcode_session_search::SessionSearchContentRoute>,
        #[serde(default)]
        hydrate: bool,
    },
    SessionSearchProviders,
    SessionSearchExplain {
        request: bcode_session_search::SessionSearchRequest,
        #[serde(default)]
        policy: bcode_session_search::SessionSearchPlanPolicy,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        routes: Vec<bcode_session_search::SessionSearchContentRoute>,
    },
    /// Explicitly purge one provider's derived session-search state.
    SessionSearchPurge {
        provider_id: String,
        confirmation: String,
    },
    /// Explicitly recreate one provider's empty derived session-search state.
    SessionSearchRebuild {
        provider_id: String,
        confirmation: String,
    },
    /// Start an addressable complete historical backfill operation.
    SessionSearchCompleteBackfillStart {
        request: bcode_session_search::CompleteSessionSearchBackfillRequest,
    },
    /// Start an addressable bounded single-provider historical backfill operation.
    SessionSearchBackfillStart {
        request: bcode_session_search::BackfillSessionSearchRequest,
    },
    /// Read bounded state for an addressable historical backfill operation.
    SessionSearchBackfillStatus {
        operation_id: String,
    },
    /// Wait for a newer revision or timeout for an addressable historical backfill operation.
    SessionSearchBackfillWait {
        operation_id: String,
        after_revision: u64,
        timeout_ms: u64,
    },
    /// Request cancellation of an addressable historical backfill operation.
    SessionSearchBackfillCancel {
        operation_id: String,
    },
    /// Explicitly backfill selected or bounded catalog sessions into one provider.
    SessionSearchBackfill {
        request: bcode_session_search::BackfillSessionSearchRequest,
    },
    /// Read one bounded non-mutating compatibility inventory page.
    SessionCompatibilityInventory {
        request: SessionCompatibilityInventoryRequest,
    },
    /// Explicitly inventory or start bounded canonical migration for selected sessions.
    SessionBulkMigrationStart {
        request: SessionBulkMigrationStartRequest,
    },
    /// Read transient aggregate migration operation state.
    SessionBulkMigrationStatus {
        operation_id: String,
    },
    /// Wait for a newer transient aggregate migration operation revision.
    SessionBulkMigrationWait {
        operation_id: String,
        after_revision: u64,
        timeout_ms: u64,
    },
    /// Request cooperative cancellation between per-session migration units.
    SessionBulkMigrationCancel {
        operation_id: String,
    },
    /// Append one durable, presentation-only note at the current session sequence.
    AppendPresentationNote {
        session_id: SessionId,
        source_id: String,
        note_id: String,
        text: String,
        format: bcode_command::CommandTextFormat,
    },
    /// Explicit maintenance: value a session interval using a supplied catalog snapshot.
    RepriceSession {
        session_id: SessionId,
        range: bcode_session_models::SessionCostRange,
        catalog: Box<bcode_model_catalog_models::CatalogDocument>,
    },
    /// Explicit bounded physical storage measurement; never loads session history.
    // Keep existing positional wire tags stable.
    SessionStorageUsage {
        session_id: SessionId,
        entry_budget: u32,
    },
    /// Bounded query of explicitly collected reporting snapshots.
    UsageReport {
        query: bcode_usage_models::UsageQuery,
    },
    /// Explicitly collect one session accounting page into the disposable reporting index.
    /// An initial request may return an empty acknowledgement: no cursor means its verified
    /// generation is already published; a cursor resumes matching durable staging.
    UsageCollect {
        session_id: SessionId,
        query: bcode_session_models::SessionUsageQuery,
    },
    /// Discover up to 128 native session identities for explicit collection.
    UsageCatalog {
        after: Option<SessionId>,
    },
    /// Explicit bounded manual compression; compatibility follows this protocol.
    SessionCompress {
        request: bcode_session_models::StorageCompressionRequest,
    },
    /// Explicit admission inspection or recovery.
    SessionAdmission {
        session_id: SessionId,
        apply: bool,
    },
    /// Read bounded accounting contributions without collecting or repairing them.
    SessionUsage {
        session_id: SessionId,
        query: bcode_session_models::SessionUsageQuery,
    },
    /// Accept cancellation intents without claiming a committed graph revision.
    AcceptWorkflowRunGraphPublication {
        request: bcode_workflow::WorkflowRunGraphEditBatch,
    },
    /// Inspect configured delegation prerequisites without granting execution authority.
    WorkflowDelegationPreflight {
        plugin_id: String,
    },
    /// Create an ordinary child with caller-retained retry identity.
    CreateChildSession {
        request: bcode_session_models::SessionChildCreateRequest,
    },
    /// Exact non-owning metadata and immutable parent navigation.
    SessionSummary {
        session_id: SessionId,
    },
    /// Bounded non-owning renderer-neutral activity observation.
    SessionActivity {
        session_id: SessionId,
        query: bcode_session_models::SessionActivityQuery,
    },
    /// Bounded direct-child discovery.
    SessionChildren {
        query: bcode_session_models::SessionChildQuery,
    },
    /// Explicit maintenance; never called from discovery or activity reads.
    ReconcileSessionChildren {
        parent_id: SessionId,
        apply: bool,
    },
    /// Read a content-free ephemeral progress marker without acquiring session ownership.
    SessionProgressSnapshot {
        session_id: SessionId,
    },
    /// Bounded inventory through one ephemeral retained workflow snapshot.
    WorkflowActivityPage {
        request: bcode_workflow_view_models::WorkflowActivityPageRequest,
    },
    /// Release a retained read snapshot; does not affect workflow execution.
    WorkflowActivityClose {
        scan_id: String,
    },
}

/// Server stop request policy.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ServerStopMode {
    /// Stop regardless of connected clients or active work.
    #[default]
    Force,
    /// Stop only when the daemon is idle.
    IfIdle,
}

/// Per-client model/provider/auth context supplied at connection time.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClientRuntimeContext {
    /// Canonical working directory of the client process.
    #[serde(default)]
    pub working_directory: Option<PathBuf>,
    /// Non-secret client configuration provenance for reproduction guidance.
    #[serde(default)]
    pub config_launch: Option<Box<bcode_config::ConfigLaunchContext>>,
    /// Fully resolved declarative configuration encoded as TOML by the client.
    ///
    /// The IPC layer treats this as opaque portable data. The daemon's config owner validates and
    /// decodes it before use.
    #[serde(default)]
    pub effective_config_toml: Option<Box<String>>,
    #[serde(default)]
    pub selected_provider_plugin_id: Option<String>,
    #[serde(default)]
    pub selected_model_id: Option<String>,
    /// User-facing model id before alias resolution.
    #[serde(default)]
    pub requested_model_id: Option<String>,
    #[serde(default)]
    pub provider_context: bcode_model::ProviderRequestContext,
    /// Provider-neutral process environment forwarded from the client shell.
    ///
    /// This is the subset of `provider_context.env` that came directly from the client's process
    /// environment (a fixed allowlist of provider configuration variables) rather than from a
    /// resolved auth profile or pool. Unlike `provider_context.auth`, it is not scoped to the
    /// client's selected provider, so the daemon may apply it to a session whose explicitly
    /// selected provider differs from the client's default. Older clients leave it empty, in which
    /// case no cross-provider environment is applied.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub process_env: BTreeMap<String, String>,
    /// Renderer-owned adapters available on this client connection.
    #[serde(default)]
    pub interaction_adapters:
        Vec<bcode_plugin_sdk::interaction::PluginInteractionAdapterCapability>,
    /// Redacted names of transient environment variables included in `provider_context.env`.
    #[serde(default)]
    pub env_keys: BTreeMap<String, bool>,
}

// Compatibility exports; catalog observation contracts belong to the session domain.
pub use bcode_session_models::{SessionCatalogSourceStatus, SessionCatalogStatus};

#[cfg(test)]
mod catalog_contract_tests {
    use super::{SessionCatalogSourceStatus, SessionCatalogStatus};

    #[test]
    fn import_warning_compatibility_preserves_optional_counts() {
        let warning: bcode_session_import::ImportWarning =
            serde_json::from_value::<super::SessionImportWarning>(
                serde_json::json!({"code": "partial", "message": "Some entries unavailable"}),
            )
            .unwrap();
        assert_eq!(warning.count, None);
        for count in [None, Some(0), Some(u64::MAX)] {
            let warning = bcode_session_import::ImportWarning {
                count,
                ..warning.clone()
            };
            let expected = serde_json::json!({"code": "partial", "message": "Some entries unavailable", "count": count});
            assert_eq!(serde_json::to_value(&warning).unwrap(), expected);
            assert_eq!(
                serde_json::from_value::<super::SessionImportWarning>(expected).unwrap(),
                warning
            );
        }
    }

    #[test]
    fn restored_selection_preserves_defaults_and_distinct_model_identities() {
        let empty: bcode_session_models::SessionRuntimeSelection =
            serde_json::from_str::<super::SessionRuntimeSelection>("{}").unwrap();
        assert_eq!(
            empty,
            bcode_session_models::SessionRuntimeSelection::default()
        );
        let payload = serde_json::json!({
            "agent_id": "build", "provider_plugin_id": "provider",
            "requested_model_id": "alias", "effective_model_id": "concrete",
            "model_id": "compatibility", "reasoning_effort": "high",
            "reasoning_summary": "auto"
        });
        let selection: bcode_session_models::SessionRuntimeSelection =
            serde_json::from_value::<super::SessionRuntimeSelection>(payload.clone()).unwrap();
        assert_eq!(selection.requested_model_id.as_deref(), Some("alias"));
        assert_eq!(selection.effective_model_id.as_deref(), Some("concrete"));
        assert_eq!(selection.model_id.as_deref(), Some("compatibility"));
        assert_eq!(serde_json::to_value(selection).unwrap(), payload);
        assert!(
            serde_json::from_value::<super::SessionRuntimeSelection>(
                serde_json::json!({"agent_id": 42})
            )
            .is_err()
        );
    }

    #[test]
    fn catalog_status_wire_values_remain_domain_owned() {
        for (status, json) in [
            (
                SessionCatalogStatus::NotStarted,
                serde_json::json!("not_started"),
            ),
            (SessionCatalogStatus::Loading, serde_json::json!("loading")),
            (SessionCatalogStatus::Loaded, serde_json::json!("loaded")),
            (
                SessionCatalogStatus::Degraded("partial".into()),
                serde_json::json!({"degraded": "partial"}),
            ),
            (
                SessionCatalogStatus::Failed("unavailable".into()),
                serde_json::json!({"failed": "unavailable"}),
            ),
        ] {
            let domain: bcode_session_models::SessionCatalogStatus = status;
            assert_eq!(serde_json::to_value(&domain).unwrap(), json);
            assert_eq!(
                serde_json::from_value::<SessionCatalogStatus>(json).unwrap(),
                domain
            );
        }
        assert!(
            serde_json::from_value::<SessionCatalogStatus>(serde_json::json!("future_status"))
                .is_err()
        );
    }

    #[test]
    fn catalog_source_preserves_legacy_timestamp_default_and_wire_shape() {
        let legacy = serde_json::json!({
            "source_id": "native", "display_name": "Bcode", "status": "loaded"
        });
        let source: bcode_session_models::SessionCatalogSourceStatus =
            serde_json::from_value::<SessionCatalogSourceStatus>(legacy).unwrap();
        assert_eq!(source.updated_at_ms, 0);
        let expected = serde_json::json!({
            "source_id": "native", "display_name": "Bcode", "status": "loaded", "updated_at_ms": 42
        });
        let source = SessionCatalogSourceStatus {
            updated_at_ms: 42,
            ..source
        };
        assert_eq!(serde_json::to_value(&source).unwrap(), expected);
        assert_eq!(
            serde_json::from_value::<SessionCatalogSourceStatus>(expected).unwrap(),
            source
        );
    }
}

/// Local server status summary.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ServerStatus {
    pub connected_client_count: usize,
    pub sessions: Vec<SessionSummary>,
    #[serde(default)]
    pub session_catalog_loaded: bool,
    #[serde(default)]
    pub session_catalog_status: SessionCatalogStatus,
    #[serde(default)]
    pub session_catalog_sources: Vec<SessionCatalogSourceStatus>,
    #[serde(default)]
    pub session_catalog_revision: u64,
    #[serde(default)]
    pub selected_provider_plugin_id: Option<String>,
    #[serde(default)]
    pub selected_model_id: Option<String>,
    #[serde(default)]
    pub plugin_runtime: Vec<bcode_plugin_models::PluginExecutorStatus>,
    /// Server process identity and lifecycle metadata.
    #[serde(default)]
    pub daemon: DaemonStatus,
    /// Lightweight runtime metrics snapshot.
    #[serde(default)]
    pub metrics: MetricsSnapshot,
    /// Rich dashboard-ready metrics report.
    #[serde(default = "default_metrics_report_box")]
    pub metrics_report: Box<bcode_metrics::MetricsReport>,
    /// Why workflow operations are unavailable, without private storage details.
    #[serde(default)]
    pub workflow_unavailable_reason: Option<String>,
    /// Runtime work currently registered as active on this daemon, across all sessions.
    ///
    /// This is the same set the daemon consults when deciding whether it is idle, so a
    /// non-empty list explains why idle shutdown or session ownership release is deferred.
    #[serde(default)]
    pub active_runtime_work: Vec<SessionRuntimeWork>,
    /// Why the daemon currently refuses idle-only shutdown, if anything blocks it.
    #[serde(default)]
    pub idle_shutdown_blocker: Option<String>,
}

/// One active runtime work registration together with its owning session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionRuntimeWork {
    pub session_id: SessionId,
    pub work: RuntimeWorkSnapshot,
}

fn default_metrics_report_box() -> Box<bcode_metrics::MetricsReport> {
    Box::default()
}

/// Server process identity and lifecycle metadata.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct DaemonStatus {
    /// Daemon namespace.
    #[serde(default)]
    pub namespace: String,
    /// IPC protocol version.
    #[serde(default)]
    pub protocol_version: u32,
    /// Exact produced-artifact identity.
    #[serde(default)]
    pub artifact_id: Option<ArtifactId>,
    /// Build fingerprint retained as source/build diagnostic evidence.
    #[serde(default)]
    pub build_fingerprint: String,
    /// SHA-256 digest of the executable bytes running the daemon.
    #[serde(default)]
    pub executable_digest: Option<String>,
    /// Durable session-storage writer epoch supported by this daemon.
    #[serde(default)]
    pub storage_writer_epoch: Option<u32>,
    /// Identity of the durable state location this daemon serves.
    #[serde(default)]
    pub state_location_id: Option<String>,
    /// Highest persisted session-event schema version understood by this daemon.
    #[serde(default)]
    pub session_event_schema_version: Option<u16>,
    /// Process identifier, when available.
    #[serde(default)]
    pub pid: Option<u32>,
    /// Random per-process identity token.
    #[serde(default)]
    pub instance_id: String,
    /// Daemon start time in Unix milliseconds.
    #[serde(default)]
    pub started_at_unix_ms: u64,
}

/// Compatibility export of the session-owned restored selection contract.
pub use bcode_session_models::SessionRuntimeSelection;

/// Compatibility export of the model-owned application status contract.
pub use bcode_model::SessionModelStatus;

#[cfg(test)]
mod contextual_agent_contract_tests {
    #[test]
    fn optional_selected_agent_round_trips_and_legacy_defaults_explicitly() {
        let legacy =
            serde_json::json!({"invocation_id":"search", "workspace":{}, "deadline_ms":1000});
        let mut context: super::PluginServiceContext = serde_json::from_value(legacy).unwrap();
        assert_eq!(context.agent_id, None);
        context.agent_id = Some("plan".into());
        let decoded: super::PluginServiceContext =
            serde_json::from_value(serde_json::to_value(&context).unwrap()).unwrap();
        assert_eq!(decoded.agent_id.as_deref(), Some("plan"));
    }
}

/// Context hints and lifetime for one client-owned plugin invocation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PluginServiceContext {
    /// Selected enabled agent hint. Missing legacy hints use the explicit default policy.
    #[serde(default)]
    pub agent_id: Option<String>,
    /// Unique identity within the connected client; reuse is rejected.
    pub invocation_id: String,
    /// Workspace hints validated by the daemon, never authorization grants.
    pub workspace: bcode_plugin_sdk::WorkspaceContext,
    /// Total admission and execution deadline, between 1 and 30,000 milliseconds.
    pub deadline_ms: u64,
}

/// Compatibility export of the plugin SDK's discovery contract.
pub use bcode_plugin_sdk::PluginContributions;

/// Service interface provided by a loaded plugin.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PluginServiceSummary {
    pub plugin_id: String,
    pub interface_id: String,
    pub name: Option<String>,
    pub description: Option<String>,
    #[serde(default)]
    pub workflow_blocks: Vec<bcode_workflow::WorkflowBlockDefinition>,
}

/// Correlation metadata for one permission checkpoint in a complete tool-call batch.
pub use bcode_session_models::PermissionBatchCorrelation;

/// Compatibility export for session-owned pending permission observations.
pub use bcode_session_models::PermissionSummary;

/// Compatibility export for domain-owned model-catalog diagnostics.
pub use bcode_model_catalog_models::ModelCatalogDiagnostics;

/// Compatibility export for the session-owned artifact range response.
pub use bcode_session_models::SessionArtifactRange;

/// Pending renderer-neutral invocation exchange.
pub use bcode_session_models::PendingToolExchangeSummary;

/// Plugin service invocation result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PluginServiceResponse {
    pub payload: Vec<u8>,
    pub error: Option<PluginServiceError>,
}

/// Plugin service invocation error payload.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PluginServiceError {
    pub code: String,
    pub message: String,
}

/// Compatibility export of the import-domain warning contract.
pub use bcode_session_import::ImportWarning as SessionImportWarning;

pub use bcode_ralph_models::{
    RalphApproveRequest, RalphCancelRequest, RalphCancelResponse, RalphIterationSummary,
    RalphLifecycleRequest, RalphListIterationsRequest, RalphListIterationsResponse,
    RalphListRunsRequest, RalphListRunsResponse, RalphResumeRequest, RalphResumeResponse,
    RalphRunRequest, RalphRunResponse, RalphRunStatusRequest, RalphRunStatusResponse,
    RalphRunSummary, RalphStatusRequest, RalphStatusResponse, RalphStatusSummary,
    RalphValidationSummary,
};

/// One unavailable workflow template requirement.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowTemplateDiagnostic {
    pub code: String,
    pub requirement: String,
    pub message: String,
}

/// Portable template catalog entry with explicit availability diagnostics.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowTemplateDescription {
    pub owner_plugin_id: String,
    pub template: bcode_plugin_models::WorkflowTemplateDescriptor<
        bcode_workflow::ValueSchema,
        bcode_workflow::WorkflowDefinition,
    >,
    /// Normalized standard authoring document when the template uses the maintainable source form.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub authoring_document: Option<bcode_workflow::WorkflowAuthoringDocument>,
    pub identity: bcode_workflow::WorkflowDefinitionIdentity,
    #[serde(default)]
    pub diagnostics: Vec<WorkflowTemplateDiagnostic>,
}

impl WorkflowTemplateDescription {
    /// Normalize a validated catalog entry without exposing plugin loading details.
    ///
    /// # Errors
    /// Returns an error when the entry has no normalized definition or configuration schema.
    pub fn into_inspection(
        self,
    ) -> Result<bcode_workflow::WorkflowTemplateInspection, &'static str> {
        let document = self.authoring_document;
        let (definition, configuration_schema) = if let Some(document) = document {
            (document.definition, document.configuration_schema)
        } else {
            (
                self.template
                    .definition
                    .ok_or("template definition unavailable")?,
                self.template
                    .configuration_schema
                    .ok_or("template configuration unavailable")?,
            )
        };
        Ok(bcode_workflow::WorkflowTemplateInspection {
            owner_plugin_id: self.owner_plugin_id,
            template_id: self.template.template_id,
            template_version: self.template.template_version,
            title: self.template.title,
            description: self.template.description,
            definition,
            configuration_schema,
            identity: self.identity,
            diagnostics: self
                .diagnostics
                .into_iter()
                .map(
                    |diagnostic| bcode_workflow::WorkflowTemplateAvailabilityDiagnostic {
                        code: diagnostic.code,
                        requirement: diagnostic.requirement,
                        message: diagnostic.message,
                    },
                )
                .collect(),
        })
    }
}

/// Compatibility export of the workflow-owned start contract.
pub use bcode_workflow::WorkflowTemplateInstantiationRequest;

/// Compatibility export of the workflow-owned start contract.
pub use bcode_workflow::WorkflowTemplateStartRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowAuthoringPage;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::ApplyWorkflowPackageRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::PublishWorkflowPackageRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowPackageComputationRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowPackagePreviewRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowPackageValidationResult;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowSourceComputationRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowSourcePreviewRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowSourceValidationResult;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowSourcePreviewResult;

/// Compatibility exports of workflow-owned computation controls.
pub use bcode_workflow::{
    DEFAULT_WORKFLOW_COMPUTATION_TIMEOUT_MS, MAX_WORKFLOW_COMPUTATION_TIMEOUT_MS,
    WorkflowComputationControl,
};

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::AuthoredWorkflowSnapshot;

/// Compatibility export of the workflow-owned draft snapshot.
pub use bcode_workflow::WorkflowDraftSnapshot;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowRevisionSnapshot;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowRevisionRequirementInspection;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowAuthoringEventSnapshot;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowAuthoringIssueSnapshot;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowDraftInspectionSummary;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowRevisionInspectionSummary;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowPresetInspectionSummary;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::AuthoredWorkflowInspection;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowPresetSnapshot;

/// Compatibility export of the workflow-owned optimistic conflict.
pub use bcode_workflow::WorkflowAuthoringConflict;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowDraftUpdateResult;

/// Compatibility exports of workflow-owned draft edit contracts.
pub use bcode_workflow::{ApplyWorkflowDraftEditsRequest, WorkflowDraftEditResult};

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::ApplyWorkflowSourceRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::CreateAuthoredWorkflowRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::UpdateWorkflowDraftRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::PublishWorkflowDraftRequest;

/// Compatibility export of the workflow-owned admission contract.
pub use bcode_workflow::PublishAndStartWorkflowRequest;

/// Compatibility export of the workflow-owned admission contract.
pub use bcode_workflow::WorkflowRunAdmissionResult;

/// Compatibility export of the workflow-owned admission contract.
pub use bcode_workflow::WorkflowPublishAndStartResult;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::ActivateWorkflowRevisionRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::SetAuthoredWorkflowArchivedRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::DiscardWorkflowDraftRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowDraftForkSource;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::ForkWorkflowDraftRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowAuthoringMutationResult;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowPublicationResult;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowPresetMutation;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::CreateWorkflowPresetRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::UpdateWorkflowPresetRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::DeleteWorkflowPresetRequest;

/// Compatibility export of the workflow-owned authoring contract.
pub use bcode_workflow::WorkflowPresetUpdateResult;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::ExportWorkflowRevisionRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::PreviewWorkflowImportRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowImportCollisionPolicy;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::ImportWorkflowRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::ImportWorkflowDraftRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::ImportWorkflowRevisionRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowRevisionImportResult;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowDraftImportResult;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::AuthoredWorkflowRunSelection;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::StartAuthoredWorkflowRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::StartWorkflowPackageExportRequest;

/// Compatibility export of the workflow-owned application contract.
pub use bcode_workflow::WorkflowPackageExportRunStartResponse;

/// Compatibility export of the workflow-owned admission contract.
pub use bcode_workflow::AuthoredWorkflowRunStartResponse;

/// Compatibility export of the workflow-owned definition registration request.
pub use bcode_workflow::WorkflowDefinitionRegistrationRequest;

/// Current bounded canonical terminal-output inspection contract version.
pub use bcode_workflow::WORKFLOW_TERMINAL_OUTPUT_INSPECTION_VERSION;

/// Current bounded validated workflow-output inspection contract version.
pub use bcode_workflow::WORKFLOW_OUTPUT_INSPECTION_VERSION;

/// Compatibility export of the workflow-owned output contract.
pub use bcode_workflow::WorkflowOutputInspection;

/// Compatibility export of the workflow-owned output contract.
pub use bcode_workflow::WorkflowTerminalOutputInspection;

/// Compatibility export of the workflow-owned observation/control contract.
pub use bcode_workflow::OrphanedWorkflowRunReport;

/// Compatibility export of the workflow-owned observation/control contract.
pub use bcode_workflow::OrphanedWorkflowRun;

/// Compatibility export of the workflow-owned observation/control contract.
pub use bcode_workflow::SkippedWorkflowRun;

/// Compatibility exports for workflow-owned graph inspection contracts.
pub use bcode_workflow::{
    WorkflowRunGraphEdgeInspection, WorkflowRunGraphInspection, WorkflowRunGraphNodeInspection,
    WorkflowRunGraphPageRequest,
};

/// Compatibility export of the workflow-owned bounded inspection contract.
pub use bcode_workflow::WorkflowRunInspection;

/// Compatibility export of the workflow-owned observation/control contract.
pub use bcode_workflow::WorkflowCoordinatorStatus;

/// Compatibility export of the workflow-owned observation/control contract.
pub use bcode_workflow::WorkflowRunControlAction;

/// Compatibility export of the workflow-owned observation/control contract.
pub use bcode_workflow::WorkflowRunBindingLookup;

/// Compatibility export of the workflow-owned start contract.
pub use bcode_workflow::WorkflowStartRequest;

/// Compatibility export of the workflow-owned start contract.
pub use bcode_workflow::WorkflowRunStartRequest;

/// Compatibility export of the workflow-owned admission contract.
pub use bcode_workflow::WorkflowRunStartResponse;

/// Domain-owned runtime-work snapshot, retained here for source compatibility.
pub use bcode_session_models::RuntimeWorkSnapshot;

/// Successful response payload variants.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ResponsePayload {
    Hello {
        protocol_version: ProtocolVersion,
        client_id: ClientId,
        daemon: DaemonStatus,
    },
    Pong,
    ServerStatus {
        status: ServerStatus,
    },
    ServerStopping,
    SessionCreated {
        session: SessionSummary,
    },
    SessionList {
        sessions: Vec<SessionSummary>,
        #[serde(default)]
        catalog_status: SessionCatalogStatus,
        #[serde(default)]
        catalog_sources: Vec<SessionCatalogSourceStatus>,
        #[serde(default)]
        catalog_revision: u64,
    },
    SessionCatalogPage {
        page: bcode_session_models::SessionCatalogPage,
    },
    SessionRenamed {
        session: SessionSummary,
    },
    SessionDeleted {
        session: SessionSummary,
    },
    SessionHistory {
        session_id: SessionId,
        history: Vec<SessionEvent>,
    },
    SessionHistoryPage {
        page: SessionHistoryPage,
    },
    SessionProjectionHistoryPage {
        page: bcode_session_models::ProjectionHistoryPage,
    },
    Attached {
        session_id: SessionId,
        session: SessionSummary,
        history: Vec<SessionEvent>,
        #[serde(default)]
        input_history: Vec<SessionInputHistoryEntry>,
        #[serde(default)]
        usage_summary: Option<Box<bcode_session_models::SessionUsageSummary>>,
        #[serde(default)]
        import_warnings: Vec<SessionImportWarning>,
        #[serde(default)]
        draft: Option<String>,
        #[serde(default)]
        runtime_selection: SessionRuntimeSelection,
        #[serde(default)]
        projection_window: Option<bcode_session_models::ProjectionWindow>,
    },
    MessageSent,
    TurnCancellationRequested {
        cancelled: bool,
    },
    SessionCompacted {
        compacted: bool,
        message: String,
    },
    SessionModelSet,
    SessionModelStatus {
        status: SessionModelStatus,
    },
    AgentList {
        agents: Vec<AgentInfo>,
    },
    SkillList {
        skills: Box<SkillList>,
    },
    SkillManifest {
        skill: Box<SkillManifest>,
    },
    ActiveSkills {
        skills: Vec<SkillContextResponse>,
    },
    AgentPolicyStatus {
        status: PolicyStatusResponse,
    },
    SessionAgentSet,
    PermissionList {
        permissions: Vec<PermissionSummary>,
    },
    PermissionResolved {
        resolved: bool,
    },
    PermissionRuleAdded {
        config_path: String,
    },
    PluginServices {
        services: Vec<PluginServiceSummary>,
    },
    /// Result of cancellation against the connected client's active invocation.
    PluginServiceInvocationCancelled {
        accepted: bool,
    },
    PluginServiceResult {
        response: PluginServiceResponse,
    },
    PluginEventPublished {
        delivered: usize,
    },
    MessageAccepted {
        queued: bool,
        queue_position: Option<u32>,
    },
    MessageAcceptedWithDisposition {
        queued: bool,
        queue_position: Option<u32>,
        disposition: MessageAcceptanceDisposition,
    },
    /// Exact durable receipt for conversational acceptance, including an idempotent retry.
    SubmissionAccepted {
        receipt: bcode_session_models::SubmissionReceipt,
    },
    /// Bounded pending-queue observation; not a snapshot or event stream.
    PendingSubmissions {
        page: bcode_session_models::PendingSubmissionsPage,
    },
    /// Bounded oldest-item queue observation, not an event stream or durable snapshot.
    SubmissionQueueStatus {
        status: bcode_session_models::SubmissionQueueStatus,
    },
    /// Exact secret-free recipe requiring explicit caller confirmation before recovery.
    SubmissionResumePrepared {
        confirmation: bcode_session_models::SubmissionResumeConfirmation,
    },
    /// Durable targeted recovery operation receipt.
    SubmissionRecovered {
        result: bcode_session_models::SubmissionRecoveryResult,
    },
    /// Bounded authoritative lookup; absence is not an inferred cancellation.
    Submission {
        record: Option<bcode_session_models::SubmissionRecord>,
    },
    TurnAdmission {
        admission: bcode_session_models::TurnAdmission,
    },
    SessionModelList {
        provider_plugin_id: Option<String>,
        models: bcode_model::ModelList,
    },
    /// Cancellation was requested, not necessarily completed.
    ModelCheckCancellationRequested {
        cancellation_requested: bool,
    },
    /// Codec-safe encoding of the normalized bounded model-check report.
    CheckModels {
        result_json: String,
    },
    /// Secret-free, bounded non-conversational evaluator setup catalog.
    JudgementSetup {
        /// Bounded JSON encoding of the portable catalog (codec-safe).
        catalog_json: String,
    },
    /// Whether daemon credentials exist and the exact model still advertises `YesNo`.
    /// Not an upstream key-validity probe or an account reservation.
    JudgementSetupValidated {
        available: bool,
    },
    /// Failure-optional normalized judgement result.
    OptionalJudgement {
        result_json: String,
    },
    /// Codec-safe encoding of the portable context selection response.
    OptionalContextSelection {
        result_json: String,
    },
    /// Normalized authentication and model readiness result.
    JudgementVerified {
        result_json: String,
    },
    /// Normalized judgement answers; no provider context or secret material.
    Judgement {
        /// Bounded JSON encoding of the validated portable judgement response.
        result_json: String,
    },
    ClientRuntimeContextUpdated,
    SessionWorkingDirectoryChanged {
        session: SessionSummary,
        changed: bool,
    },
    WorktreeList(WorktreeListResponse),
    WorktreeRemoved(WorktreeRemoveResponse),
    RalphStatus(RalphStatusResponse),
    RalphRunStarted(RalphRunResponse),
    RalphRunCancelled(RalphCancelResponse),
    RalphRunsListed(RalphListRunsResponse),
    RalphIterationsListed(RalphListIterationsResponse),
    RalphRunResumed(RalphResumeResponse),
    RalphRunApproved(RalphRunResponse),
    RalphRunStatus(RalphRunStatusResponse),
    RalphLifecycleRecorded {
        event: SessionEvent,
    },
    ExternalSessionImported {
        session: SessionSummary,
        warnings: Vec<SessionImportWarning>,
    },
    SessionMilestoneGenerated {
        result: SessionMilestoneGenerationOutcome,
    },
    SessionMilestones {
        page: bcode_session_models::SessionMilestonePage,
    },
    SessionDerivationSnapshot {
        snapshot: SessionDerivationSourceSnapshot,
    },
    SessionDerivationPrompts {
        page: SessionDerivationPromptPage,
    },
    SessionDerived {
        outcome: SessionDerivationTerminalOutcome,
    },
    SessionDerivationStatus {
        snapshot: bcode_session_models::SessionDerivationOperationSnapshot,
    },
    SessionDerivationCancellationRequested {
        accepted: bool,
    },
    SessionCatalogRefreshed {
        sessions: Vec<SessionSummary>,
        #[serde(default)]
        catalog_status: SessionCatalogStatus,
        #[serde(default)]
        catalog_sources: Vec<SessionCatalogSourceStatus>,
        #[serde(default)]
        catalog_revision: u64,
    },
    CatalogUpdatesSubscribed,
    RuntimeWorkObservations {
        page: bcode_session_models::RuntimeWorkObservationPage,
    },
    RuntimeWorkList {
        work: Vec<RuntimeWorkSnapshot>,
    },
    RuntimeWorkCancellationRequested {
        cancelled: bool,
    },
    AuthoredWorkflowCreated {
        workflow: AuthoredWorkflowSnapshot,
        draft: Box<WorkflowDraftSnapshot>,
    },
    WorkflowSourceApplied {
        result: bcode_workflow::WorkflowSourceApplyResult,
    },
    WorkflowComputationCancellationRequested {
        cancelled: bool,
    },
    /// Receipt from a completed explicit incompatible workflow-store reset.
    WorkflowStoreReset {
        receipt: bcode_workflow_store::WorkflowStoreResetReceipt,
    },
    WorkflowDraftEditResult {
        result: WorkflowDraftEditResult,
    },
    WorkflowDraftUpdateResult {
        result: WorkflowDraftUpdateResult,
    },
    WorkflowPublicationResult {
        result: WorkflowPublicationResult,
    },
    WorkflowPublishAndStartResult {
        result: WorkflowPublishAndStartResult,
    },
    WorkflowActivationResult {
        result: WorkflowAuthoringMutationResult,
    },
    AuthoredWorkflowArchived {
        workflow: AuthoredWorkflowSnapshot,
    },
    WorkflowDraftDiscardResult {
        result: WorkflowAuthoringMutationResult,
    },
    WorkflowDraftForked {
        draft: Box<WorkflowDraftSnapshot>,
    },
    WorkflowPresetCreated {
        preset: WorkflowPresetSnapshot,
    },
    WorkflowPresetUpdateResult {
        result: WorkflowPresetUpdateResult,
    },
    WorkflowPresetDeleteResult {
        result: WorkflowAuthoringMutationResult,
    },
    WorkflowRevisionExported {
        bundle: Box<bcode_workflow::WorkflowExportBundle>,
    },
    WorkflowImportPreview {
        preview: Box<bcode_workflow::WorkflowImportPreview>,
    },
    WorkflowImported {
        workflow: AuthoredWorkflowSnapshot,
        draft: Box<WorkflowDraftSnapshot>,
    },
    WorkflowDraftImported {
        result: WorkflowDraftImportResult,
    },
    WorkflowRevisionImported {
        result: WorkflowRevisionImportResult,
    },
    AuthoredWorkflowRunStarted(AuthoredWorkflowRunStartResponse),
    WorkflowPackageExportRunStarted(Box<WorkflowPackageExportRunStartResponse>),
    AuthoredWorkflowList {
        page: WorkflowAuthoringPage<
            AuthoredWorkflowSnapshot,
            bcode_workflow::WorkflowAuthoringListCursor,
        >,
    },
    AuthoredWorkflowDescription {
        workflow: Option<AuthoredWorkflowSnapshot>,
    },
    AuthoredWorkflowInspection {
        inspection: Option<Box<AuthoredWorkflowInspection>>,
    },
    WorkflowDraftList {
        page: WorkflowAuthoringPage<
            WorkflowDraftSnapshot,
            bcode_workflow::WorkflowAuthoringListCursor,
        >,
    },
    WorkflowDraftDescription {
        draft: Option<Box<WorkflowDraftSnapshot>>,
    },
    WorkflowRevisionList {
        page: WorkflowAuthoringPage<
            WorkflowRevisionSnapshot,
            bcode_workflow::WorkflowRevisionListCursor,
        >,
    },
    WorkflowRevisionDescription {
        revision: Option<Box<WorkflowRevisionSnapshot>>,
    },
    WorkflowRevisionRequirementInspection {
        inspection: Option<WorkflowRevisionRequirementInspection>,
    },
    WorkflowPresetList {
        page: WorkflowAuthoringPage<
            WorkflowPresetSnapshot,
            bcode_workflow::WorkflowAuthoringListCursor,
        >,
    },
    WorkflowPresetDescription {
        preset: Option<WorkflowPresetSnapshot>,
    },
    WorkflowAuthoringCatalog {
        catalog: bcode_workflow::WorkflowAuthoringCatalogSnapshot,
    },
    WorkflowDiscoveryCancelled {
        released: bool,
    },
    WorkflowLaunchCatalog {
        page: bcode_workflow::WorkflowLaunchCatalogPage,
    },
    WorkflowLaunchDetail {
        detail: Box<bcode_workflow::WorkflowLaunchDetail>,
    },
    WorkflowPackagePublication {
        receipt: Option<bcode_workflow::WorkflowPackagePublicationReceipt>,
    },
    WorkflowPackageApplied {
        result: Box<bcode_workflow::WorkflowPackageMutationResult>,
    },
    WorkflowPackagePublished {
        result: Box<bcode_workflow::WorkflowPackageMutationResult>,
    },
    WorkflowPackageValidated {
        result: Box<WorkflowPackageValidationResult>,
    },
    WorkflowPackagePreviewed {
        preview: Box<bcode_workflow::WorkflowPackagePreview>,
    },
    WorkflowSourceValidated {
        result: Box<WorkflowSourceValidationResult>,
    },
    WorkflowSourcePreviewed {
        result: Box<WorkflowSourcePreviewResult>,
    },
    WorkflowAuthoringValidated {
        report: bcode_workflow::WorkflowValidationReport,
    },
    WorkflowCompilationPreview {
        preview: Box<bcode_workflow::WorkflowCompilationPreview>,
    },
    /// Bounded normalized template inspections.
    WorkflowTemplateInspections {
        templates: Vec<bcode_workflow::WorkflowTemplateInspection>,
    },
    /// Optional exact normalized template inspection.
    WorkflowTemplateInspection {
        template: Option<Box<bcode_workflow::WorkflowTemplateInspection>>,
    },
    WorkflowTemplateList {
        templates: Vec<WorkflowTemplateDescription>,
    },
    WorkflowTemplateDescription {
        template: Option<Box<WorkflowTemplateDescription>>,
    },
    WorkflowTemplateStarted(WorkflowRunStartResponse),
    WorkflowDefinitionRegistered {
        definition: bcode_workflow::StoredWorkflowDefinition,
    },
    WorkflowReplacementRequested(bcode_workflow::WorkflowReplacementResponse),
    WorkflowRunStarted(WorkflowRunStartResponse),
    WorkflowContinuationSource(Box<bcode_workflow::WorkflowContinuationSource>),
    WorkflowDefinitionList {
        definitions: Vec<bcode_workflow::StoredWorkflowDefinition>,
    },
    WorkflowDefinitionDescription {
        definition: Option<bcode_workflow::StoredWorkflowDefinition>,
    },
    /// Committed graph revision, including for identical duplicate publication.
    WorkflowRunGraphEditPublished {
        revision: u64,
    },
    /// Candidate persisted (true) or identical candidate already present (false).
    WorkflowRunGraphEditStaged {
        created: bool,
    },
    WorkflowRunGraphInspection {
        graph: WorkflowRunGraphInspection,
    },
    WorkflowRunInspection {
        inspection: Box<WorkflowRunInspection>,
    },
    WorkflowIdentityCatalog {
        page: bcode_workflow_view_models::WorkflowIdentityCatalogPage,
    },
    WorkflowSchedulingPage {
        page: bcode_workflow::WorkflowSchedulingPage<
            bcode_workflow_view_models::WorkflowSchedulingView,
        >,
    },
    WorkflowRunView {
        view: Box<bcode_workflow_view_models::WorkflowRunView>,
    },
    WorkflowCatalogView {
        view: bcode_workflow_view_models::WorkflowCatalogView,
    },
    WorkflowRunStatus {
        run: Option<bcode_workflow_store::WorkflowRunSummary>,
    },
    AssociatedWorkflowRun {
        run: Option<bcode_workflow_store::WorkflowRunSummary>,
    },
    AssociatedWorkflowRunInspection {
        inspection: Option<Box<WorkflowRunInspection>>,
    },
    AssociatedWorkflowRunControlled {
        run: Option<bcode_workflow_store::WorkflowRunSummary>,
        changed: bool,
    },
    WorkflowRunList {
        runs: Vec<bcode_workflow_store::WorkflowRunSummary>,
    },
    WorkflowRunOutputs {
        outputs: Vec<WorkflowOutputInspection>,
    },
    WorkflowDoctorReport {
        report: bcode_workflow::WorkflowDoctorReport,
    },
    WorkflowAttemptRepaired {
        result: bcode_workflow::RepairResult,
    },
    WorkflowRunCancellationRequested {
        recorded: bool,
    },
    OrphanedWorkflowRunsReconciled {
        report: OrphanedWorkflowRunReport,
    },
    WorkflowRunPaused {
        changed: bool,
    },
    WorkflowRunResumed {
        changed: bool,
    },
    WorkflowNodeRetried {
        result: bcode_workflow_store::WorkflowNodeRetryResult,
    },
    WorkflowWaitList {
        waits: Vec<bcode_workflow::WaitingActivation>,
    },
    WorkflowWaitResolved {
        result: bcode_workflow_store::WaitingResolutionResult,
    },
    WorkflowMutationApprovalList {
        approvals: Vec<bcode_workflow::WorkflowMutationApprovalInspection>,
    },
    WorkflowMutationApprovalResolved {
        result: bcode_workflow::WorkflowMutationApprovalResolution,
    },
    WorkflowAttemptHistory {
        attempts: Vec<bcode_workflow::AttemptSummary>,
    },
    WorkflowEventHistory {
        events: Vec<bcode_workflow::WorkflowHistoryEvent>,
    },
    WorkflowCommittedHistory {
        page: bcode_workflow::WorkflowCommittedHistoryPage,
    },
    WorkflowLifecycleHistory {
        page: bcode_workflow::WorkflowLifecycleHistoryPage,
    },
    WorkflowInventory {
        page: bcode_workflow::WorkflowInventoryPage,
    },
    WorkflowSemanticHistory {
        page: bcode_workflow::WorkflowSemanticHistoryPage,
    },
    WorkflowObservations {
        page: bcode_workflow::WorkflowObservationPage,
    },
    WorkflowLiveEventCatchUp {
        page: bcode_workflow_view_models::WorkflowLiveEventPage,
    },
    RuntimeWorkHistory {
        events: Vec<SessionEvent>,
    },
    RuntimeWorkSubscribed,
    /// Subscription watermark. No event with a sequence at or below this value will be published
    /// by this connection's live subscription; clients use it to detect the first gap.
    WorkflowRunsSubscribed {
        after_sequence: u64,
    },
    ComposerDraft {
        draft: Option<String>,
    },
    ComposerDraftSet,
    PluginContributions {
        contributions: PluginContributions,
    },
    PendingToolExchangeList {
        exchanges: Vec<PendingToolExchangeSummary>,
    },
    ToolExchangeResolved {
        resolved: bool,
    },
    /// Effective model catalog diagnostics.
    ModelCatalogDiagnostics {
        diagnostics: ModelCatalogDiagnostics,
    },
    PreparedContextGeneration {
        prepared: bcode_session_models::PreparedContextGeneration,
    },
    SessionWorkingDocument {
        document: Option<bcode_session_models::SessionWorkingDocument>,
    },
    WorkflowDiagnosticArtifacts {
        artifacts: bcode_workflow::WorkflowDiagnosticArtifacts,
    },
    WorkflowDiagnosticArtifactRange {
        range: bcode_workflow::WorkflowDiagnosticArtifactRange,
    },
    SessionTraceBlobRange {
        range: bcode_session_models::SessionTraceBlobRange,
    },
    SessionArtifactRange {
        range: SessionArtifactRange,
    },
    SessionHistoryAround {
        window: SessionHistoryWindow,
    },
    InvocationInputAccepted,
    PermissionBatchResolved {
        resolved: usize,
    },
    /// Acknowledges that a bounded client metrics batch was accepted.
    ClientMetricsIngested {
        /// Number of observations merged into the daemon registry.
        accepted: usize,
    },
    /// Portable auth-pool status list.
    AuthPoolList {
        pools: Vec<bcode_provider_auth_models::AuthPoolSummary>,
    },
    /// Interactive auth-pool preference was updated.
    AuthPoolPreferenceSet,
    /// Current snapshot after preparing or waiting for session open.
    SessionOpenPrepared {
        snapshot: SessionOpenOperationSnapshot,
    },
    SessionOwnershipReleased {
        session_id: SessionId,
        outcome: SessionOwnershipReleaseOutcome,
    },
    /// Current disposable advisory labels, absent after eviction or daemon restart.
    /// Codec-safe portable snapshot; absence is unknown after restart/eviction/legacy execution.
    TurnJudgementObservation {
        observation_json: Option<String>,
    },
    SessionCategorization {
        categorization: Option<bcode_session_models::SessionCategorization>,
    },
    SessionInspection {
        page: SessionInspectionPage,
    },
    SessionSearch {
        response: bcode_session_search::FederatedSessionSearchResponse,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        hydrated_hits: Vec<bcode_session_search::HydratedSessionSearchHit>,
    },
    SessionSearchProviders {
        response: bcode_session_search::ListSessionSearchProvidersResponse,
    },
    SessionSearchPlan {
        plan: bcode_session_search::SessionSearchPlan,
    },
    SessionSearchMaintenance {
        response: bcode_session_search::SessionSearchMaintenanceResponse,
    },
    SessionSearchBackfillStarted {
        response: bcode_session_search::StartSessionSearchBackfillResponse,
    },
    SessionSearchCompleteBackfillStarted {
        response: bcode_session_search::StartSessionSearchBackfillResponse,
    },
    SessionSearchBackfillOperation {
        status: bcode_session_search::SessionSearchBackfillOperationStatus,
    },
    SessionSearchBackfill {
        response: bcode_session_search::SessionSearchBackfillResponse,
    },
    SessionCompatibilityInventory {
        response: SessionCompatibilityInventoryResponse,
    },
    SessionBulkMigrationOperation {
        status: SessionBulkMigrationOperationStatus,
    },
    WorktreeCreateOperation {
        status: bcode_worktree_models::WorktreeCreateOperationStatus,
    },
    PresentationNoteAppended,
    /// Compaction has entered the session queue; a terminal response follows on this request.
    SessionCompactionAccepted,
    SessionRepriced {
        report: Box<bcode_session_models::SessionRepriceReport>,
    },
    // Binary enum tags are positional: append new responses to preserve v1 tags.
    SessionStorageUsage {
        usage: bcode_session_models::SessionStorageUsage,
    },
    UsageReport {
        report: bcode_usage_models::UsageReport,
    },
    UsageCollected {
        page: bcode_session_models::SessionUsagePage,
    },
    UsageCatalog {
        session_ids: Vec<SessionId>,
    },
    SessionCompressed {
        result: bcode_session_models::StorageCompressionResult,
    },
    SessionAdmission {
        report: bcode_session_models::StorageAdmissionReport,
    },
    SessionUsage {
        page: bcode_session_models::SessionUsagePage,
    },
    /// Lifecycle outcome of an accepted graph publication.
    WorkflowRunGraphPublicationAccepted {
        status: bcode_workflow::WorkflowRunGraphPublicationStatus,
    },
    /// Configured prerequisites only; not an execution grant.
    WorkflowDelegationPreflight {
        result: bcode_workflow::WorkflowDelegationPreflight,
    },
    /// Exact canonical metadata, including immutable parent navigation.
    SessionSummary {
        session: SessionSummary,
    },
    /// Renderer-neutral non-owning activity observation.
    SessionActivity {
        page: bcode_session_models::SessionActivityPage,
    },
    /// Bounded discovery; incomplete pages do not imply a complete tree.
    SessionChildren {
        page: bcode_session_models::SessionChildPage,
    },
    /// Explicit relationship maintenance result.
    SessionChildrenReconciled {
        result: bcode_session_models::SessionChildReconciliation,
    },
    /// Own-session progress only; unavailable coverage is not silence or idleness.
    SessionProgressSnapshot {
        snapshot: SessionProgressSnapshot,
    },
    /// Inventory is as-of a retained snapshot, not current-idle authority.
    WorkflowActivityPage {
        page: bcode_workflow_view_models::WorkflowActivityPage,
    },
    /// Ephemeral read resources released; no execution effects.
    WorkflowActivityClosed,
}

/// Stable reasons runtime ownership cannot currently be released.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionOwnershipBlocker {
    AttachedClient,
    PendingAttach,
    QueuedCommand,
    ActiveRuntime,
    RuntimeWork,
    PluginInvocation,
    Migration,
    /// The session database connection could not be proven closed during release.
    ///
    /// The session was otherwise quiescent, so this reports a retained backend handle rather than
    /// user-visible activity. Release fails closed in this case instead of claiming success,
    /// because a retained handle keeps process-level file locks that would block a later owner.
    DatabaseHandleRetained,
}

/// Typed outcome of an explicit session ownership release request.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionOwnershipReleaseOutcome {
    Released,
    AlreadyUnowned,
    Blocked {
        blockers: Vec<SessionOwnershipBlocker>,
    },
}

/// Structured error response.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ErrorResponse {
    pub code: String,
    pub message: String,
}

impl ErrorResponse {
    /// Create a new error response.
    #[must_use]
    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
        Self {
            code: code.into(),
            message: message.into(),
        }
    }
}

/// Top-level response message.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Response {
    Ok(ResponsePayload),
    Err(ErrorResponse),
}

/// Server-to-client event payloads.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Event {
    Session(SessionEvent),
    SessionLive(SessionLiveEvent),
    RuntimeWork(SessionEvent),
    /// Notification that canonical state for one workflow run changed.
    Workflow(bcode_workflow_view_models::WorkflowLiveEvent),
    /// Local/client signal that a reattached session view needs one bounded snapshot refresh.
    SessionViewResyncRequired {
        session_id: SessionId,
    },
    SessionCatalogUpdated {
        #[serde(default)]
        revision: u64,
    },
}

/// Errors returned by Bcode IPC encoding/decoding.
#[derive(Debug, Error)]
pub enum CodecError {
    #[error("frame payload exceeds max size ({actual} bytes > {max} bytes)")]
    PayloadTooLarge { actual: usize, max: usize },
    #[error("invalid IPC chunk: {0}")]
    InvalidChunk(String),
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("serialization failed: {0}")]
    Serialize(#[source] bmux_codec::Error),
    #[error("deserialization failed: {0}")]
    Deserialize(#[source] bmux_codec::Error),
    #[error("event conversion failed: {0}")]
    EventConversion(String),
    #[error("unsupported protocol version {actual}; expected {expected}")]
    UnsupportedVersion { actual: u16, expected: u16 },
}

/// Encode a value with the positional Bcode wire codec.
///
/// Positional encoding is intended for fixed, hot-path IPC framing structures.
/// Prefer [`encode_typed_stable`] for logical IPC payloads with evolving schemas.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn encode_positional<T: Serialize>(value: &T) -> Result<Vec<u8>, CodecError> {
    bmux_codec::to_positional_vec(value).map_err(CodecError::Serialize)
}

/// Decode a value with the positional Bcode wire codec.
///
/// # Errors
///
/// Returns an error when deserialization fails.
pub fn decode_positional<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CodecError> {
    bmux_codec::from_positional_bytes(bytes).map_err(CodecError::Deserialize)
}

/// Encode a value with the typed-stable Bcode wire codec.
///
/// Typed-stable encoding is intended for logical IPC payloads whose schemas
/// evolve more frequently than the outer frame protocol.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn encode_typed_stable<T: Serialize>(value: &T) -> Result<Vec<u8>, CodecError> {
    bmux_codec::to_typed_vec(value).map_err(CodecError::Serialize)
}

/// Decode a value with the typed-stable Bcode wire codec.
///
/// # Errors
///
/// Returns an error when deserialization fails.
pub fn decode_typed_stable<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CodecError> {
    bmux_codec::from_typed_bytes(bytes).map_err(CodecError::Deserialize)
}

/// Encode a serializable value with the fixed-frame Bcode wire codec.
///
/// This compatibility helper remains positional for envelope and chunk framing.
/// New logical IPC payload code should use the explicit typed-stable helpers.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn encode<T: Serialize>(value: &T) -> Result<Vec<u8>, CodecError> {
    encode_positional(value)
}

/// Encode a request with the Bcode wire codec.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn encode_request(request: &Request) -> Result<Vec<u8>, CodecError> {
    encode_typed_stable(request)
}

/// Decode a request with the Bcode wire codec.
///
/// # Errors
///
/// Returns an error when deserialization fails.
pub fn decode_request(bytes: &[u8]) -> Result<Request, CodecError> {
    decode_typed_stable(bytes)
}

/// Encode a server event with the Bcode wire codec.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn encode_event(event: &Event) -> Result<Vec<u8>, CodecError> {
    encode_typed_stable(event)
}

/// Decode a deserializable value with the fixed-frame Bcode wire codec.
///
/// This compatibility helper remains positional for envelope and chunk framing.
/// New logical IPC payload code should use the explicit typed-stable helpers.
///
/// # Errors
///
/// Returns an error when deserialization fails.
pub fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CodecError> {
    decode_positional(bytes)
}

/// Encode a response with the Bcode wire codec.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn encode_response(response: &Response) -> Result<Vec<u8>, CodecError> {
    encode_typed_stable(response)
}

/// Decode a response with the Bcode wire codec.
///
/// # Errors
///
/// Returns an error when deserialization or domain conversion fails.
pub fn decode_response(bytes: &[u8]) -> Result<Response, CodecError> {
    decode_typed_stable(bytes)
}

/// Decode a server event with the Bcode wire codec.
///
/// # Errors
///
/// Returns an error when deserialization or domain conversion fails.
pub fn decode_event(bytes: &[u8]) -> Result<Event, CodecError> {
    decode_typed_stable(bytes)
}

const COMPRESSION_MIN_BYTES: usize = 256 * 1024;
const COMPRESSION_LEVEL: i32 = 1;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct CompressedEnvelopePayload {
    kind: EnvelopeKind,
    request_id: u64,
    algorithm_wire_id: u8,
    uncompressed_len: u64,
    data: Vec<u8>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct ChunkPayload {
    chunk_index: u32,
    chunk_count: u32,
    total_len: u64,
    data: Vec<u8>,
}

/// Send one logical envelope.
///
/// Logical envelopes larger than [`MAX_FRAME_PAYLOAD_SIZE`] are transparently
/// fragmented into multiple physical IPC frames and reassembled by
/// [`recv_envelope`].
///
/// # Errors
///
/// Returns an error when serialization or writing fails.
pub async fn send_envelope<W>(writer: &mut W, envelope: &Envelope) -> Result<(), CodecError>
where
    W: AsyncWrite + Unpin,
{
    let frames = encode_envelope_frames(envelope)?;
    write_encoded_envelope_frames(writer, &frames).await
}

/// Encode one logical envelope into physical wire frames.
///
/// Pre-encoding lets callers perform expensive serialization/chunk construction outside writer
/// critical sections while preserving the exact same wire protocol.
///
/// # Errors
///
/// Returns an error when serialization fails or the logical envelope is too large to chunk.
pub fn encode_envelope_frames(envelope: &Envelope) -> Result<Vec<Vec<u8>>, CodecError> {
    let physical_envelope = maybe_compress_envelope(envelope)?;
    let payload = encode(&physical_envelope)?;
    if payload.len() <= MAX_FRAME_PAYLOAD_SIZE {
        return Ok(vec![encode_envelope_frame(&physical_envelope)?]);
    }
    encode_chunked_envelope_frames(physical_envelope.request_id, &payload)
}

fn maybe_compress_envelope(envelope: &Envelope) -> Result<Envelope, CodecError> {
    if !matches!(envelope.kind, EnvelopeKind::Response | EnvelopeKind::Event) {
        return Ok(envelope.clone());
    }
    let policy = bmux_codec::compression::CompressionPolicy::new(
        bmux_codec::compression::CompressionAlgorithm::Lz4,
    )
    .min_bytes(COMPRESSION_MIN_BYTES)
    .level(COMPRESSION_LEVEL);
    match bmux_codec::compression::maybe_compress_bytes(envelope.payload.clone(), policy)
        .map_err(CodecError::Serialize)?
    {
        bmux_codec::compression::CompressionDecision::Plain { .. } => Ok(envelope.clone()),
        bmux_codec::compression::CompressionDecision::Compressed {
            algorithm,
            uncompressed_len,
            bytes,
        } => Ok(Envelope::new(
            envelope.request_id,
            EnvelopeKind::Compressed,
            encode(&CompressedEnvelopePayload {
                kind: envelope.kind,
                request_id: envelope.request_id,
                algorithm_wire_id: algorithm.wire_id(),
                uncompressed_len: u64::try_from(uncompressed_len).map_err(|_| {
                    CodecError::PayloadTooLarge {
                        actual: uncompressed_len,
                        max: usize::MAX,
                    }
                })?,
                data: bytes,
            })?,
        )),
    }
}

fn encode_chunked_envelope_frames(
    request_id: u64,
    payload: &[u8],
) -> Result<Vec<Vec<u8>>, CodecError> {
    let chunk_count = payload.len().div_ceil(MAX_CHUNK_DATA_SIZE);
    let chunk_count = u32::try_from(chunk_count).map_err(|_| CodecError::PayloadTooLarge {
        actual: payload.len(),
        max: MAX_FRAME_PAYLOAD_SIZE,
    })?;
    let total_len = u64::try_from(payload.len()).map_err(|_| CodecError::PayloadTooLarge {
        actual: payload.len(),
        max: MAX_FRAME_PAYLOAD_SIZE,
    })?;

    payload
        .chunks(MAX_CHUNK_DATA_SIZE)
        .enumerate()
        .map(|(chunk_index, data)| {
            let chunk_payload = ChunkPayload {
                chunk_index: u32::try_from(chunk_index).map_err(|_| {
                    CodecError::PayloadTooLarge {
                        actual: payload.len(),
                        max: MAX_FRAME_PAYLOAD_SIZE,
                    }
                })?,
                chunk_count,
                total_len,
                data: data.to_vec(),
            };
            let chunk_envelope =
                Envelope::new(request_id, EnvelopeKind::Chunk, encode(&chunk_payload)?);
            encode_envelope_frame(&chunk_envelope)
        })
        .collect()
}

/// Write pre-encoded physical envelope frames.
///
/// # Errors
///
/// Returns an error when the underlying writer fails.
pub async fn write_encoded_envelope_frames<W>(
    writer: &mut W,
    frames: &[Vec<u8>],
) -> Result<(), CodecError>
where
    W: AsyncWrite + Unpin,
{
    for frame in frames {
        writer.write_all(frame).await?;
    }
    writer.flush().await?;
    Ok(())
}

/// Return how many physical frames would be written for a logical envelope.
///
/// # Errors
///
/// Returns an error when envelope serialization fails or the frame count exceeds wire limits.
pub fn envelope_frame_count(envelope: &Envelope) -> Result<u32, CodecError> {
    u32::try_from(encode_envelope_frames(envelope)?.len()).map_err(|_| {
        CodecError::PayloadTooLarge {
            actual: usize::MAX,
            max: MAX_FRAME_PAYLOAD_SIZE,
        }
    })
}

fn encode_envelope_frame(envelope: &Envelope) -> Result<Vec<u8>, CodecError> {
    let payload = encode(envelope)?;
    if payload.len() > MAX_FRAME_PAYLOAD_SIZE {
        return Err(CodecError::PayloadTooLarge {
            actual: payload.len(),
            max: MAX_FRAME_PAYLOAD_SIZE,
        });
    }
    let payload_len = u32::try_from(payload.len()).map_err(|_| CodecError::PayloadTooLarge {
        actual: payload.len(),
        max: MAX_FRAME_PAYLOAD_SIZE,
    })?;
    let mut frame = Vec::with_capacity(FRAME_LEN_BYTES.saturating_add(payload.len()));
    frame.extend_from_slice(&payload_len.to_le_bytes());
    frame.extend_from_slice(&payload);
    Ok(frame)
}

/// Receive one logical envelope.
///
/// If the sender fragmented a large logical envelope into continuation frames,
/// this function reassembles those frames before returning.
///
/// # Errors
///
/// Returns an error when reading, reassembly, or deserialization fails.
pub async fn recv_envelope<R>(reader: &mut R) -> Result<Envelope, CodecError>
where
    R: AsyncRead + Unpin,
{
    let envelope = read_envelope_frame(reader).await?;
    if envelope.kind == EnvelopeKind::Chunk {
        recv_chunked_envelope(reader, envelope).await
    } else {
        unwrap_compressed_envelope(envelope)
    }
}

async fn read_envelope_frame<R>(reader: &mut R) -> Result<Envelope, CodecError>
where
    R: AsyncRead + Unpin,
{
    let mut len_bytes = [0_u8; FRAME_LEN_BYTES];
    reader.read_exact(&mut len_bytes).await?;
    let payload_len = u32::from_le_bytes(len_bytes) as usize;
    if payload_len > MAX_FRAME_PAYLOAD_SIZE {
        return Err(CodecError::PayloadTooLarge {
            actual: payload_len,
            max: MAX_FRAME_PAYLOAD_SIZE,
        });
    }
    let mut payload = vec![0_u8; payload_len];
    reader.read_exact(&mut payload).await?;
    let envelope: Envelope = decode(&payload)?;
    if envelope.version != ProtocolVersion::current() {
        return Err(CodecError::UnsupportedVersion {
            actual: envelope.version.0,
            expected: ProtocolVersion::current().0,
        });
    }
    Ok(envelope)
}

async fn recv_chunked_envelope<R>(
    reader: &mut R,
    first_envelope: Envelope,
) -> Result<Envelope, CodecError>
where
    R: AsyncRead + Unpin,
{
    let first = decode_chunk_payload(&first_envelope)?;
    validate_first_chunk(&first)?;

    let mut assembled = Vec::new();
    let chunk_count = first.chunk_count;
    let total_len = first.total_len;
    assembled.extend_from_slice(&first.data);

    for expected_index in 1..chunk_count {
        let envelope = read_envelope_frame(reader).await?;
        if envelope.kind != EnvelopeKind::Chunk {
            return Err(CodecError::InvalidChunk(format!(
                "expected chunk {expected_index}, got {:?}",
                envelope.kind
            )));
        }
        let chunk = decode_chunk_payload(&envelope)?;
        validate_next_chunk(&chunk, expected_index, chunk_count, total_len)?;
        assembled.extend_from_slice(&chunk.data);
    }

    let actual_len = u64::try_from(assembled.len()).map_err(|_| {
        CodecError::InvalidChunk("assembled payload length does not fit in u64".to_string())
    })?;
    if actual_len != total_len {
        return Err(CodecError::InvalidChunk(format!(
            "assembled payload length {actual_len} does not match expected {total_len}"
        )));
    }

    let envelope: Envelope = decode(&assembled)?;
    if envelope.kind == EnvelopeKind::Chunk {
        return Err(CodecError::InvalidChunk(
            "nested chunk envelope is not allowed".to_string(),
        ));
    }
    if envelope.version != ProtocolVersion::current() {
        return Err(CodecError::UnsupportedVersion {
            actual: envelope.version.0,
            expected: ProtocolVersion::current().0,
        });
    }
    unwrap_compressed_envelope(envelope)
}

fn unwrap_compressed_envelope(envelope: Envelope) -> Result<Envelope, CodecError> {
    if envelope.kind != EnvelopeKind::Compressed {
        return Ok(envelope);
    }
    let compressed: CompressedEnvelopePayload = decode(&envelope.payload)?;
    if matches!(
        compressed.kind,
        EnvelopeKind::Chunk | EnvelopeKind::Compressed
    ) {
        return Err(CodecError::InvalidChunk(
            "compressed envelope cannot unwrap to chunk or compressed envelope".to_string(),
        ));
    }
    let algorithm =
        bmux_codec::compression::CompressionAlgorithm::from_wire_id(compressed.algorithm_wire_id)
            .map_err(CodecError::Deserialize)?;
    let expected_len =
        usize::try_from(compressed.uncompressed_len).map_err(|_| CodecError::PayloadTooLarge {
            actual: usize::MAX,
            max: MAX_FRAME_PAYLOAD_SIZE,
        })?;
    let payload =
        bmux_codec::compression::decompress_bytes(algorithm, &compressed.data, expected_len)
            .map_err(CodecError::Deserialize)?;
    Ok(Envelope::new(
        compressed.request_id,
        compressed.kind,
        payload,
    ))
}

fn decode_chunk_payload(envelope: &Envelope) -> Result<ChunkPayload, CodecError> {
    decode(&envelope.payload)
}

fn validate_first_chunk(chunk: &ChunkPayload) -> Result<(), CodecError> {
    if chunk.chunk_count == 0 {
        return Err(CodecError::InvalidChunk(
            "chunk count must be greater than zero".to_string(),
        ));
    }
    if chunk.chunk_index != 0 {
        return Err(CodecError::InvalidChunk(format!(
            "first chunk index must be 0, got {}",
            chunk.chunk_index
        )));
    }
    validate_next_chunk(chunk, 0, chunk.chunk_count, chunk.total_len)
}

fn validate_next_chunk(
    chunk: &ChunkPayload,
    expected_index: u32,
    chunk_count: u32,
    total_len: u64,
) -> Result<(), CodecError> {
    if chunk.chunk_index != expected_index {
        return Err(CodecError::InvalidChunk(format!(
            "expected chunk index {expected_index}, got {}",
            chunk.chunk_index
        )));
    }
    if chunk.chunk_count != chunk_count {
        return Err(CodecError::InvalidChunk(format!(
            "chunk count changed from {chunk_count} to {}",
            chunk.chunk_count
        )));
    }
    if chunk.total_len != total_len {
        return Err(CodecError::InvalidChunk(format!(
            "total length changed from {total_len} to {}",
            chunk.total_len
        )));
    }
    if chunk.data.len() > MAX_CHUNK_DATA_SIZE {
        return Err(CodecError::InvalidChunk(format!(
            "chunk data exceeds max size ({} bytes > {MAX_CHUNK_DATA_SIZE} bytes)",
            chunk.data.len()
        )));
    }
    Ok(())
}

/// Build a request envelope.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn request_envelope(request_id: u64, request: &Request) -> Result<Envelope, CodecError> {
    Ok(Envelope::new(
        request_id,
        EnvelopeKind::Request,
        encode_request(request)?,
    ))
}

/// Build a response envelope.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn response_envelope(request_id: u64, response: &Response) -> Result<Envelope, CodecError> {
    Ok(Envelope::new(
        request_id,
        EnvelopeKind::Response,
        encode_response(response)?,
    ))
}

/// Build an event envelope.
///
/// # Errors
///
/// Returns an error when serialization fails.
pub fn event_envelope(event: &Event) -> Result<Envelope, CodecError> {
    Ok(Envelope::new(0, EnvelopeKind::Event, encode_event(event)?))
}

/// Return the normalized current working directory for session scoping.
#[must_use]
pub fn current_working_directory() -> PathBuf {
    env::current_dir().map_or_else(|_| PathBuf::from("."), |path| normalize_path(&path))
}

fn normalize_path(path: &Path) -> PathBuf {
    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
}

#[cfg(unix)]
fn prepare_endpoint_for_bind(endpoint: &IpcEndpoint) -> Result<(), IpcTransportError> {
    if let Some(path) = endpoint.as_unix_socket() {
        prepare_unix_socket_path_for_bind(path)?;
    }
    Ok(())
}

#[cfg(unix)]
fn prepare_unix_socket_path_for_bind(path: &Path) -> Result<(), IpcTransportError> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    if !path.exists() {
        return Ok(());
    }
    match std::os::unix::net::UnixStream::connect(path) {
        Ok(_stream) => Err(std::io::Error::new(
            std::io::ErrorKind::AddrInUse,
            format!(
                "refusing to replace live IPC socket {}; another bcode daemon is listening",
                display_from_current_dir(path)
            ),
        )
        .into()),
        Err(error)
            if matches!(
                error.kind(),
                std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
            ) =>
        {
            fs::remove_file(path)?;
            Ok(())
        }
        Err(error) => Err(error.into()),
    }
}

/// Environment variable carrying an encoded local IPC endpoint for child processes.
pub const BCODE_IPC_ENDPOINT_ENV: &str = "BCODE_IPC_ENDPOINT";

/// Environment variable identifying the build namespace that supplied an IPC endpoint override.
pub const BCODE_IPC_ENDPOINT_NAMESPACE_ENV: &str = "BCODE_IPC_ENDPOINT_NAMESPACE";

/// Serialize an IPC endpoint for process environment propagation.
///
/// # Errors
///
/// Returns an error when endpoint serialization fails.
pub fn endpoint_env_value(endpoint: &IpcEndpoint) -> Result<String, serde_json::Error> {
    serde_json::to_string(endpoint)
}

/// Return the environment pair used to propagate an exact IPC endpoint.
///
/// # Errors
///
/// Returns an error when endpoint serialization fails.
pub fn endpoint_env_pair(
    endpoint: &IpcEndpoint,
) -> Result<(&'static str, String), serde_json::Error> {
    Ok((BCODE_IPC_ENDPOINT_ENV, endpoint_env_value(endpoint)?))
}

/// Parse an IPC endpoint previously produced by [`endpoint_env_value`].
///
/// # Errors
///
/// Returns an error when the encoded endpoint is invalid.
pub fn endpoint_from_env_value(value: &str) -> Result<IpcEndpoint, serde_json::Error> {
    serde_json::from_str(value)
}

/// Return the daemon namespace for this produced artifact and IPC protocol version.
#[must_use]
pub fn daemon_namespace() -> String {
    daemon_namespace_for(&ArtifactId::current())
}

/// Return the daemon namespace for one exact produced artifact.
#[must_use]
pub fn daemon_namespace_for(artifact_id: &ArtifactId) -> String {
    format!("ipc-v{CURRENT_PROTOCOL_VERSION}-{artifact_id}")
}

/// Return the stable identity of the durable runtime scope this process uses.
///
/// The scope is the pair of resolved locations that determine a daemon's behavior:
/// the durable **state root** (which canonical session storage it owns) and the
/// **config directory** (which plugins, permissions, and model policy it applies).
/// Endpoint routing, daemon registry scoping, and handshake verification all use
/// this identity, so a client reaches only a daemon serving the same pair it
/// resolved. Opening a session under a different state root or a different config
/// directory therefore starts a separate daemon rather than reusing this one.
///
/// Including the config directory matters for correctness, not just tidiness: two
/// config directories can select different plugins and permission policy, so a
/// shared daemon would apply the wrong policy to one of them.
///
/// It is derived only from those resolved paths, never from writer, build, or
/// process identity.
#[must_use]
pub fn state_location_id() -> String {
    bcode_config::runtime_scope_id(
        &bcode_config::default_state_dir(),
        &bcode_config::default_config_dir(),
    )
}

/// Return the default local IPC endpoint.
#[must_use]
pub fn default_endpoint() -> IpcEndpoint {
    let endpoint_override_allowed = endpoint_override_allowed_for_current_process();
    if endpoint_override_allowed
        && let Ok(value) = env::var(BCODE_IPC_ENDPOINT_ENV)
        && let Ok(endpoint) = endpoint_from_env_value(&value)
    {
        return endpoint;
    }
    #[cfg(unix)]
    {
        IpcEndpoint::unix_socket(default_socket_path(endpoint_override_allowed))
    }
    #[cfg(windows)]
    {
        let user_scope = current_windows_user_scope();
        IpcEndpoint::windows_named_pipe(format!(
            r"\\.\pipe\bcode-{user_scope}-{}-{}",
            daemon_namespace(),
            state_location_id()
        ))
    }
}

#[cfg(windows)]
fn current_windows_user_scope() -> String {
    windows_token_user_scope().unwrap_or_else(|_| {
        windows_user_scope(
            env::var("USERDOMAIN").ok().as_deref(),
            env::var("USERNAME").ok().as_deref(),
            env::var_os("USERPROFILE").as_deref().map(Path::new),
        )
    })
}

#[cfg(windows)]
fn windows_token_user_scope() -> std::io::Result<String> {
    use windows_sys::Win32::Foundation::{CloseHandle, HANDLE};
    use windows_sys::Win32::Security::{
        GetLengthSid, GetTokenInformation, TOKEN_QUERY, TOKEN_USER, TokenUser,
    };
    use windows_sys::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken};

    struct Token(HANDLE);

    impl Drop for Token {
        fn drop(&mut self) {
            // SAFETY: `self.0` is an owned token handle returned by `OpenProcessToken`.
            unsafe {
                CloseHandle(self.0);
            }
        }
    }

    let mut token = std::ptr::null_mut();
    // SAFETY: The pseudo process handle is always valid and `token` points to writable storage.
    if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &raw mut token) } == 0 {
        return Err(std::io::Error::last_os_error());
    }
    let token = Token(token);
    let mut required = 0;
    // SAFETY: A null output buffer with zero length is the documented size query.
    unsafe {
        GetTokenInformation(
            token.0,
            TokenUser,
            std::ptr::null_mut(),
            0,
            &raw mut required,
        );
    }
    if required == 0 {
        return Err(std::io::Error::last_os_error());
    }
    let word_size = std::mem::size_of::<usize>();
    let words = usize::try_from(required)
        .map_err(|_| std::io::Error::other("Windows token information length overflow"))?
        .div_ceil(word_size);
    let mut buffer = vec![0_usize; words];
    // SAFETY: `buffer` is aligned for `TOKEN_USER`, has at least `required` writable bytes, and the
    // token remains open for the duration of the query and SID read.
    if unsafe {
        GetTokenInformation(
            token.0,
            TokenUser,
            buffer.as_mut_ptr().cast(),
            required,
            &raw mut required,
        )
    } == 0
    {
        return Err(std::io::Error::last_os_error());
    }
    // SAFETY: A successful `TokenUser` query writes a `TOKEN_USER` at the start of the buffer.
    let token_user = unsafe { &*buffer.as_ptr().cast::<TOKEN_USER>() };
    // SAFETY: `token_user.User.Sid` is valid while `buffer` remains alive.
    let sid_length = unsafe { GetLengthSid(token_user.User.Sid) };
    if sid_length == 0 {
        return Err(std::io::Error::last_os_error());
    }
    // SAFETY: `GetLengthSid` returned the exact readable byte length for this SID.
    let sid = unsafe {
        std::slice::from_raw_parts(
            token_user.User.Sid.cast::<u8>(),
            usize::try_from(sid_length)
                .map_err(|_| std::io::Error::other("Windows SID length overflow"))?,
        )
    };
    Ok(short_scope_digest(&[sid]))
}

#[cfg(any(windows, test))]
fn windows_user_scope(
    domain: Option<&str>,
    username: Option<&str>,
    profile: Option<&Path>,
) -> String {
    let domain = domain.unwrap_or_default().trim().to_lowercase();
    let username = username.unwrap_or_default().trim().to_lowercase();
    let profile = profile
        .map(|path| path.as_os_str().to_string_lossy().to_lowercase())
        .unwrap_or_default();
    short_scope_digest(&[domain.as_bytes(), username.as_bytes(), profile.as_bytes()])
}

#[cfg(any(windows, test))]
fn short_scope_digest(components: &[&[u8]]) -> String {
    let mut hasher = Sha256::new();
    for component in components {
        hasher.update(component);
        hasher.update([0]);
    }
    let digest = hex::encode(hasher.finalize());
    digest[..16].to_owned()
}

fn endpoint_override_allowed_for_current_process() -> bool {
    let current_namespace = daemon_namespace();
    let inherited_namespace = env::var(BCODE_IPC_ENDPOINT_NAMESPACE_ENV).ok();
    endpoint_override_allowed(
        inherited_namespace.as_deref(),
        env::var_os("BCODE_DAEMON_LOG").is_some(),
        &current_namespace,
    )
}

fn endpoint_override_allowed(
    inherited_namespace: Option<&str>,
    daemon_context: bool,
    current_namespace: &str,
) -> bool {
    inherited_namespace.map_or(!daemon_context, |namespace| namespace == current_namespace)
}

#[cfg(unix)]
const MAX_PORTABLE_UNIX_SOCKET_PATH_BYTES: usize = 103;
#[cfg(unix)]
const SOCKET_SCOPE_HEX_BYTES: usize = 32;

#[cfg(unix)]
fn default_socket_path(endpoint_override_allowed: bool) -> PathBuf {
    if endpoint_override_allowed && let Ok(path) = env::var("BCODE_SOCKET") {
        return PathBuf::from(path);
    }
    let user = env::var("USER").unwrap_or_else(|_| "user".to_string());
    let namespace = daemon_namespace();
    let digest = socket_scope_digest(&user, &namespace, &state_location_id());
    // Endpoint discovery must be identical for every client of this artifact/state,
    // including launchers with different temporary-directory environments.
    // The scope digest isolates users/artifacts/state; /tmp keeps Unix paths bounded.
    socket_path_with_digest(Path::new("/tmp"), &digest).expect("/tmp can hold a bcode socket path")
}

#[cfg(unix)]
fn socket_scope_digest(user: &str, namespace: &str, state_location_id: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(user.as_bytes());
    hasher.update([0]);
    hasher.update(namespace.as_bytes());
    hasher.update([0]);
    hasher.update(state_location_id.as_bytes());
    let digest = hex::encode(hasher.finalize());
    digest[..SOCKET_SCOPE_HEX_BYTES].to_owned()
}

#[cfg(unix)]
fn socket_path_with_digest(base: &Path, digest: &str) -> Option<PathBuf> {
    use std::os::unix::ffi::OsStrExt as _;

    let path = base.join(format!("bcode-{digest}.sock"));
    (path.as_os_str().as_bytes().len() <= MAX_PORTABLE_UNIX_SOCKET_PATH_BYTES).then_some(path)
}

#[cfg(test)]
mod child_wire_tests;

#[cfg(test)]
mod tests {
    #[test]
    fn committed_history_query_roundtrip_retains_pin_and_version() {
        let query = bcode_workflow::WorkflowCommittedHistoryQuery {
            version: 1,
            run_id: "run".into(),
            after_sequence: 4,
            through_sequence: Some(200),
            limit: 25,
        };
        let request = super::Request::WorkflowCommittedHistory {
            query: query.clone(),
        };
        let decoded =
            serde_json::from_slice::<super::Request>(&serde_json::to_vec(&request).unwrap())
                .unwrap();
        let super::Request::WorkflowCommittedHistory { query: actual } = decoded else {
            panic!("committed history request");
        };
        assert_eq!(actual, query);
        actual.validate().unwrap();
    }
    #[test]
    fn lifecycle_history_query_and_page_roundtrip_without_legacy_conversion() {
        use bcode_workflow::*;
        let query = WorkflowLifecycleHistoryQuery {
            version: WORKFLOW_LIFECYCLE_HISTORY_VERSION,
            run_id: "run".into(),
            scope: WorkflowLifecycleHistoryScope::RunLifecycleV1,
            cursor: None,
            limit: 7,
        };
        let request = super::Request::WorkflowLifecycleHistory {
            query: query.clone(),
        };
        let encoded = serde_json::to_vec(&request).unwrap();
        let decoded: super::Request = serde_json::from_slice(&encoded).unwrap();
        let super::Request::WorkflowLifecycleHistory { query: actual } = decoded else {
            panic!("typed request");
        };
        assert_eq!(actual, query);
        let page = WorkflowLifecycleHistoryPage {
            cursor: WorkflowLifecycleHistoryCursor {
                version: query.version,
                run_id: query.run_id.clone(),
                scope: query.scope,
                coverage: WorkflowSemanticHistoryCoverage::HistoricalGap,
                through_sequence: 0,
                after_sequence: 0,
            },
            events: Vec::new(),
            next_cursor: None,
        };
        let response = super::ResponsePayload::WorkflowLifecycleHistory { page: page.clone() };
        let encoded = serde_json::to_vec(&response).unwrap();
        let decoded: super::ResponsePayload = serde_json::from_slice(&encoded).unwrap();
        let super::ResponsePayload::WorkflowLifecycleHistory { page: actual } = decoded else {
            panic!("typed page");
        };
        assert_eq!(actual, page);
        assert_eq!(query.validate_page(&actual), Ok(()));
    }
    #[test]
    fn semantic_history_query_and_page_roundtrip_without_legacy_conversion() {
        use bcode_workflow::*;
        let query = WorkflowSemanticHistoryQuery {
            version: WORKFLOW_SEMANTIC_HISTORY_VERSION,
            run_id: "run".into(),
            scope: WorkflowSemanticHistoryScope::AttemptsV1,
            cursor: None,
            limit: 7,
        };
        let request = super::Request::WorkflowSemanticHistory {
            query: query.clone(),
        };
        let encoded = serde_json::to_vec(&request).unwrap();
        let decoded: super::Request = serde_json::from_slice(&encoded).unwrap();
        let super::Request::WorkflowSemanticHistory { query: actual } = decoded else {
            panic!("typed request");
        };
        assert_eq!(actual, query);
        let page = WorkflowSemanticHistoryPage {
            cursor: WorkflowSemanticHistoryCursor {
                version: query.version,
                run_id: query.run_id.clone(),
                scope: query.scope,
                coverage: WorkflowSemanticHistoryCoverage::HistoricalGap,
                through_sequence: 0,
                after_sequence: 0,
            },
            events: Vec::new(),
            next_cursor: None,
        };
        let response = super::ResponsePayload::WorkflowSemanticHistory { page: page.clone() };
        let encoded = serde_json::to_vec(&response).unwrap();
        let decoded: super::ResponsePayload = serde_json::from_slice(&encoded).unwrap();
        let super::ResponsePayload::WorkflowSemanticHistory { page: actual } = decoded else {
            panic!("typed page");
        };
        assert_eq!(actual, page);
        assert_eq!(query.validate_page(&actual), Ok(()));
    }
    use super::*;
    use bcode_session_models::{
        CURRENT_SESSION_EVENT_SCHEMA_VERSION, ModelTurnOutcome, RuntimeWorkKind, RuntimeWorkStatus,
        SessionEventKind, SessionId, SessionSummary, SessionTraceEvent, ToolInvocationResult,
    };
    use bcode_skill_models::SkillActivationMode;
    use std::collections::BTreeSet;

    #[test]
    fn compression_wire_round_trip() {
        let request = Request::SessionCompress {
            request: bcode_session_models::StorageCompressionRequest {
                session_id: SessionId::new(),
                as_of_ms: 123,
                minimum_age_ms: None,
                tier: bcode_session_models::StorageCompressionTier::Deep,
                dry_run: false,
                cursor: Some(bcode_session_models::StorageCompressionCursor::History {
                    start: 5,
                    through: 9,
                }),
            },
        };
        let bytes = serde_json::to_vec(&request).expect("encode");
        assert_eq!(
            serde_json::from_slice::<Request>(&bytes).expect("decode"),
            request
        );
    }

    #[test]
    fn usage_catalog_round_trip_preserves_cursor_and_id_order() {
        let id = SessionId::new();
        let request = Request::UsageCatalog { after: Some(id) };
        let decoded: Request =
            serde_json::from_slice(&serde_json::to_vec(&request).unwrap()).unwrap();
        assert!(matches!(decoded, Request::UsageCatalog { after: Some(actual) } if actual == id));
        let response = ResponsePayload::UsageCatalog {
            session_ids: vec![id],
        };
        let decoded: ResponsePayload =
            serde_json::from_slice(&serde_json::to_vec(&response).unwrap()).unwrap();
        assert!(
            matches!(decoded, ResponsePayload::UsageCatalog { session_ids } if session_ids == vec![id])
        );
    }

    #[test]
    fn storage_usage_round_trip_preserves_partial_accounting() {
        let session_id = SessionId::new();
        let request = Request::SessionStorageUsage {
            session_id,
            entry_budget: 7,
        };
        let decoded: Request =
            serde_json::from_slice(&serde_json::to_vec(&request).expect("encode request"))
                .expect("decode request");
        assert!(
            matches!(decoded, Request::SessionStorageUsage { session_id: id, entry_budget: 7 } if id == session_id)
        );
        let usage = bcode_session_models::SessionStorageUsage {
            visited_entries: 7,
            skipped_entries: 2,
            budget_exhausted: true,
            database: bcode_session_models::SessionStorageBytes {
                files: 1,
                file_bytes: 9000,
                allocated_bytes: Some(4096),
            },
            ..Default::default()
        };
        let response = ResponsePayload::SessionStorageUsage {
            usage: usage.clone(),
        };
        let decoded: ResponsePayload =
            serde_json::from_slice(&serde_json::to_vec(&response).expect("encode response"))
                .expect("decode response");
        let ResponsePayload::SessionStorageUsage { usage: actual } = decoded else {
            panic!("unexpected response")
        };
        assert_eq!(actual, usage);
    }

    #[test]
    fn plugin_discovery_contract_preserves_populated_wire_shape() {
        let json = serde_json::json!({
            "commands": [{"plugin_id":"example", "command": {
                "id":"inspect", "title":"Inspect", "description":null,
                "category":"tools", "surface":"slash"
            }}],
            "command_contributions": [],
            "composer_contributions": [],
            "config_extensions": [{"plugin_id":"example", "section":"example",
                "aliases":[{"section":"legacy", "reason":"compatibility"}],
                "categories":["tools"], "schema_version":1, "schema_file":"schema.json"
            }]
        });
        let wire: PluginContributions = serde_json::from_value(json.clone()).unwrap();
        let domain: bcode_plugin_sdk::PluginContributions = wire;
        assert_eq!(
            domain.config_extensions[0].sections(),
            ["example", "legacy"]
        );
        assert!(domain.composer_contributions.is_empty());
        let mut legacy = json.clone();
        legacy
            .as_object_mut()
            .unwrap()
            .remove("composer_contributions");
        let legacy: PluginContributions = serde_json::from_value(legacy).unwrap();
        assert_eq!(legacy, domain);
        assert_eq!(serde_json::to_value(domain).unwrap(), json);
        let empty: PluginContributions = serde_json::from_str("{}").unwrap();
        assert!(empty.commands.is_empty());
        assert!(empty.command_contributions.is_empty());
        assert!(empty.config_extensions.is_empty());
        assert!(empty.composer_contributions.is_empty());
    }

    #[test]
    fn graph_contracts_preserve_domain_identity_and_wire_shape() {
        let request_json = serde_json::json!({
            "run_id": "run-1", "expected_revision": 7,
            "after_node_id": "node-2", "after_edge_id": 9, "limit": 12,
        });
        let wire: WorkflowRunGraphPageRequest =
            serde_json::from_value(request_json.clone()).unwrap();
        let domain: bcode_workflow::WorkflowRunGraphPageRequest = wire;
        assert_eq!(serde_json::to_value(domain).unwrap(), request_json);
        let page_json = serde_json::json!({
            "revision": 7, "nodes": [], "edges": [],
            "nodes_complete": true, "edges_complete": false,
        });
        let wire: WorkflowRunGraphInspection = serde_json::from_value(page_json.clone()).unwrap();
        let domain: bcode_workflow::WorkflowRunGraphInspection = wire;
        assert_eq!(serde_json::to_value(domain).unwrap(), page_json);
        let mut unknown_request = request_json;
        unknown_request["future_field"] = serde_json::json!(true);
        assert!(serde_json::from_value::<WorkflowRunGraphPageRequest>(unknown_request).is_err());
        let mut unknown_page = page_json;
        unknown_page["future_field"] = serde_json::json!(true);
        assert!(serde_json::from_value::<WorkflowRunGraphInspection>(unknown_page).is_err());
    }

    #[test]
    fn model_status_preserves_domain_identity_and_wire_shape() {
        let wire = serde_json::json!({
            "provider_plugin_id": "provider", "requested_model_id": "alias",
            "effective_model_id": "concrete", "model_id": "concrete",
            "display_name": "Model", "context_window": 128_000,
            "context_occupancy": null, "request_context_error": null,
            "auth_profile": "work", "context_format_version": 1,
            "compatibility_key": "format", "max_output_tokens": 8192,
            "reasoning": null, "reasoning_effort": "high", "reasoning_summary": "auto",
            "prompt_cache_mode": "auto", "conversation_reuse_mode": "auto",
            "compaction_mode": "auto", "compaction_backend": "local",
            "proactive_compaction_threshold_percent": 80,
            "proactive_compaction_threshold_tokens": 100_000,
            "proactive_compaction_effective_threshold_tokens": 96_000,
            "proactive_compaction_threshold_source": "tokens",
            "cache": null, "metadata_source": null, "pricing": null
        });
        let ipc: SessionModelStatus = serde_json::from_value(wire.clone()).unwrap();
        let domain: bcode_model::SessionModelStatus = ipc;
        assert_eq!(serde_json::to_value(&domain).unwrap(), wire);
        let empty: bcode_model::SessionModelStatus =
            serde_json::from_value(serde_json::json!({})).unwrap();
        let encoded = serde_json::to_value(&empty).unwrap();
        assert_eq!(
            encoded.as_object().unwrap().len(),
            wire.as_object().unwrap().len()
        );
        assert!(
            encoded
                .as_object()
                .unwrap()
                .values()
                .all(serde_json::Value::is_null)
        );
        let mut invalid = wire;
        invalid["context_window"] = serde_json::json!("unknown");
        assert!(serde_json::from_value::<SessionModelStatus>(invalid).is_err());
    }

    #[test]
    fn runtime_work_snapshot_preserves_domain_identity_and_wire_shape() {
        let mut wire = serde_json::json!({
            "work_id": "work-1", "kind": "tool", "label": "inspect λ",
            "tool_call_id": "call-1", "status": "cancelling", "cancellable": true,
        });
        let ipc: RuntimeWorkSnapshot = serde_json::from_value(wire.clone()).unwrap();
        let domain: bcode_session_models::RuntimeWorkSnapshot = ipc;
        assert_eq!(domain.status, RuntimeWorkStatus::Cancelling);
        assert_eq!(serde_json::to_value(&domain).unwrap(), wire);
        wire.as_object_mut().unwrap().remove("tool_call_id");
        let domain: bcode_session_models::RuntimeWorkSnapshot =
            serde_json::from_value(wire.clone()).unwrap();
        assert_eq!(domain.tool_call_id, None);
        wire["status"] = serde_json::json!("future_status");
        assert!(serde_json::from_value::<RuntimeWorkSnapshot>(wire).is_err());
    }

    #[test]
    fn pending_exchange_summary_preserves_domain_identity_and_wire_shape() {
        let wire = serde_json::json!({
            "session_id": "00000000-0000-0000-0000-000000000001",
            "request": {
                "invocation_id": "invocation",
                "exchange_id": "exchange",
                "producer_id": "producer",
                "schema": "producer.exchange",
                "schema_version": 42,
                "payload": { "opaque": [null, true, "λ\ntext", 7] },
                "response_policy": "required"
            }
        });
        let ipc: PendingToolExchangeSummary =
            serde_json::from_value(wire.clone()).expect("existing pending exchange wire shape");
        // Assignment verifies the compatibility export is the domain type, not a copy.
        let domain: bcode_session_models::PendingToolExchangeSummary = ipc;
        assert_eq!(domain.request.schema_version, 42);
        assert_eq!(serde_json::to_value(&domain).unwrap(), wire);
        let decoded: bcode_session_models::PendingToolExchangeSummary =
            serde_json::from_value(wire).unwrap();
        assert_eq!(decoded, domain);
    }

    #[test]
    fn raw_artifact_identity_tracks_final_bytes_and_survives_copy() {
        let first = b"final target a with cached dependency label";
        let second = b"final target b with cached dependency label";
        let identity = ArtifactId::from_executable(&mut first.as_slice()).unwrap();
        assert_eq!(
            identity,
            ArtifactId::from_executable(&mut first.to_vec().as_slice()).unwrap()
        );
        assert_ne!(
            identity,
            ArtifactId::from_executable(&mut second.as_slice()).unwrap()
        );
        assert_eq!(ArtifactId::parse(identity.as_str()).unwrap(), identity);
    }

    #[test]
    fn raw_artifact_identity_rejects_unreadable_bytes() {
        struct Unreadable;
        impl std::io::Read for Unreadable {
            fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
                Err(std::io::Error::other("unreadable artifact"))
            }
        }
        assert!(ArtifactId::from_executable(&mut Unreadable).is_err());
    }

    #[test]
    fn artifact_id_validates_portable_identity() {
        let artifact_id = ArtifactId::parse("artifact_123-test").expect("valid artifact identity");
        assert_eq!(artifact_id.as_str(), "artifact_123-test");
        assert!(ArtifactId::parse("").is_err());
        assert!(ArtifactId::parse("invalid/value").is_err());
    }

    #[test]
    fn daemon_namespace_uses_exact_artifact_identity() {
        let artifact_id = ArtifactId::parse("artifact-a").expect("artifact identity");
        assert_eq!(
            daemon_namespace_for(&artifact_id),
            format!("ipc-v{CURRENT_PROTOCOL_VERSION}-artifact-a")
        );
        assert_eq!(
            daemon_namespace(),
            daemon_namespace_for(&ArtifactId::current())
        );
    }

    #[cfg(unix)]
    #[test]
    fn socket_endpoint_environment_probe() {
        if env::var_os("BCODE_SOCKET_ENDPOINT_PROBE").is_some() {
            println!("endpoint-probe:{}", default_socket_path(false).display());
        }
    }

    #[cfg(unix)]
    #[test]
    fn socket_endpoint_is_independent_of_temporary_directory() {
        let executable = env::current_exe().expect("test executable");
        let mut paths = Vec::new();
        for directory in [
            "/tmp",
            "/var/tmp",
            "/tmp/a-deliberately-different-temporary-directory",
        ] {
            let output = std::process::Command::new(&executable)
                .args([
                    "--exact",
                    "tests::socket_endpoint_environment_probe",
                    "--nocapture",
                ])
                .env("BCODE_SOCKET_ENDPOINT_PROBE", "1")
                .env("TMPDIR", directory)
                .env("TMP", directory)
                .env("TEMP", directory)
                .output()
                .expect("endpoint probe");
            assert!(output.status.success());
            let stdout = String::from_utf8(output.stdout).expect("probe text");
            let path = stdout
                .lines()
                .find_map(|line| line.strip_prefix("endpoint-probe:"))
                .expect("observed endpoint")
                .to_owned();
            paths.push(path);
        }
        assert!(paths.windows(2).all(|pair| pair[0] == pair[1]));
    }

    #[cfg(unix)]
    #[test]
    fn generated_socket_paths_are_bounded_and_namespace_scoped() {
        use std::os::unix::ffi::OsStrExt as _;

        let first_digest =
            socket_scope_digest("example-user", "ipc-v26-artifact-a", "state-location-a");
        let second_digest =
            socket_scope_digest("example-user", "ipc-v26-artifact-b", "state-location-a");
        let path = socket_path_with_digest(Path::new("/tmp"), &first_digest)
            .expect("representable socket path");

        assert_eq!(first_digest.len(), SOCKET_SCOPE_HEX_BYTES);
        assert_ne!(first_digest, second_digest);
        assert!(path.starts_with("/tmp"));
        assert!(path.as_os_str().as_bytes().len() <= MAX_PORTABLE_UNIX_SOCKET_PATH_BYTES);
    }

    #[cfg(unix)]
    #[test]
    fn socket_scope_separates_distinct_state_locations() {
        let first = socket_scope_digest("example-user", "ipc-v30-artifact-a", "state-location-a");
        let second = socket_scope_digest("example-user", "ipc-v30-artifact-a", "state-location-b");

        assert_ne!(
            first, second,
            "two state locations must not share one daemon endpoint"
        );
        assert_eq!(
            first,
            socket_scope_digest("example-user", "ipc-v30-artifact-a", "state-location-a"),
            "endpoint scoping must be deterministic for one location"
        );
    }

    #[cfg(unix)]
    #[test]
    fn generated_socket_path_rejects_an_overlong_base() {
        let digest = socket_scope_digest("user", "namespace", "state-location");
        let base = PathBuf::from("/").join("x".repeat(MAX_PORTABLE_UNIX_SOCKET_PATH_BYTES));
        assert!(socket_path_with_digest(&base, &digest).is_none());
    }

    #[test]
    fn windows_user_scope_is_normalized_bounded_and_collision_resistant() {
        let first = windows_user_scope(
            Some("Example"),
            Some("Alice"),
            Some(Path::new(r"C:\\Users\\Alice")),
        );
        let equivalent = windows_user_scope(
            Some("example"),
            Some(" alice "),
            Some(Path::new(r"c:\\users\\alice")),
        );
        let other_domain = windows_user_scope(
            Some("Other"),
            Some("Alice"),
            Some(Path::new(r"C:\\Users\\Alice")),
        );
        let other_profile = windows_user_scope(
            Some("Example"),
            Some("Alice"),
            Some(Path::new(r"D:\\Profiles\\Alice")),
        );

        assert_eq!(first, equivalent);
        assert_ne!(first, other_domain);
        assert_ne!(first, other_profile);
        assert_eq!(first.len(), 16);
        assert!(first.bytes().all(|byte| byte.is_ascii_hexdigit()));
        assert!(!first.contains("alice"));
    }

    #[test]
    #[cfg(windows)]
    fn default_windows_endpoint_is_a_namespaced_named_pipe() {
        let endpoint = default_endpoint();
        let name = endpoint
            .as_windows_named_pipe()
            .expect("Windows default endpoint must be a named pipe");
        assert!(name.starts_with(r"\\.\pipe\bcode-"));
        assert!(name.ends_with(&format!("{}-{}", daemon_namespace(), state_location_id())));
        let username = env::var("USERNAME")
            .unwrap_or_default()
            .trim()
            .to_lowercase();
        if !username.is_empty() {
            assert!(!name.to_lowercase().contains(&username));
        }
    }

    #[test]
    fn permission_batch_correlation_round_trips_and_defaults_when_absent() {
        let summary = PermissionSummary {
            interaction_route: None,
            permission_id: "perm-1".to_string(),
            session_id: SessionId::new(),
            tool_call_id: "call-2".to_string(),
            tool_name: "example.tool".to_string(),
            arguments_json: "{}".to_string(),
            batch: Some(PermissionBatchCorrelation {
                batch_id: "permission-batch-7".to_string(),
                call_index: 1,
                call_count: 3,
            }),
            agent_id: "build".to_string(),
            policy_source: None,
            policy_reason: None,
            can_remember_policy: false,
        };
        let value = serde_json::to_value(&summary).expect("permission summary should encode");
        let decoded: PermissionSummary =
            serde_json::from_value(value).expect("permission summary should decode");
        assert_eq!(decoded, summary);

        let mut value = serde_json::to_value(summary).expect("permission summary should encode");
        value
            .as_object_mut()
            .expect("permission summary JSON object")
            .remove("batch");
        let decoded: PermissionSummary =
            serde_json::from_value(value).expect("summary without batch should decode");
        assert_eq!(decoded.batch, None);
    }

    #[test]
    #[allow(clippy::too_many_lines)]
    fn workflow_requests_and_responses_round_trip_renderer_neutrally() {
        let definition = bcode_workflow::WorkflowBuilder::new(
            "review",
            bcode_workflow::Step::map("review", |value: u32| Ok(value)),
        )
        .build()
        .expect("workflow")
        .definition()
        .clone();
        let authoring_document = bcode_workflow::WorkflowAuthoringDocument {
            schema_version: bcode_workflow::WORKFLOW_AUTHORING_DOCUMENT_VERSION,
            workflow_id: "authored/review".to_string(),
            metadata: bcode_workflow::WorkflowAuthoringMetadata {
                title: "Review".to_string(),
                description: None,
                labels: std::collections::BTreeMap::new(),
            },
            configuration_schema: bcode_workflow::ValueSchema::of::<serde_json::Value>(),
            configuration_defaults: None,
            plugin_input_defaults: std::collections::BTreeMap::new(),
            definition: definition.clone(),
            bindings: Vec::new(),
            requirements: bcode_workflow::WorkflowRequirementSummary::default(),
            run_limits: bcode_workflow::WorkflowRunLimitPolicy::default(),
            producer: bcode_workflow::WorkflowProducerProvenance {
                kind: bcode_workflow::WorkflowProducerKind::Sdk,
                producer_id: Some("ipc-test".to_string()),
                source_revision: None,
            },
            presentation: None,
        };
        let template_description = WorkflowTemplateDescription {
            owner_plugin_id: "owner".to_string(),
            template: bcode_plugin_models::WorkflowTemplateDescriptor {
                contribution_version: 1,
                template_id: "review".to_string(),
                template_version: 1,
                title: "Review".to_string(),
                description: "Review source".to_string(),
                configuration_schema: Some(authoring_document.configuration_schema.clone()),
                definition: Some(definition.clone()),
                document_source: None,
                required_plugins: Vec::new(),
                required_capabilities: Vec::new(),
                presentation: std::collections::BTreeMap::new(),
            },
            authoring_document: None,
            identity: bcode_workflow::WorkflowDefinitionIdentity {
                kind: "review".to_string(),
                definition_id: "review-definition".to_string(),
                definition_version: definition.schema_version,
            },
            diagnostics: vec![WorkflowTemplateDiagnostic {
                code: "missing-plugin".to_string(),
                requirement: "optional-plugin".to_string(),
                message: "Plugin unavailable".to_string(),
            }],
        };
        let inline = template_description
            .clone()
            .into_inspection()
            .expect("inline inspection");
        assert_eq!(inline.definition, definition);
        assert_eq!(inline.diagnostics[0].requirement, "optional-plugin");
        let mut external = template_description;
        external.template.definition = None;
        external.template.configuration_schema = None;
        external.authoring_document = Some(authoring_document.clone());
        assert_eq!(
            external
                .clone()
                .into_inspection()
                .expect("external inspection"),
            inline
        );
        external.authoring_document = None;
        assert!(external.into_inspection().is_err());
        let encoded = serde_json::to_value(&inline).expect("inspection JSON");
        assert!(encoded.get("template").is_none());
        assert!(encoded.get("document_source").is_none());
        let requests = vec![
            Request::InspectWorkflowTemplates { limit: 10 },
            Request::InspectWorkflowTemplate {
                owner_plugin_id: "owner".to_string(),
                template_id: "review".to_string(),
                template_version: 1,
            },
            Request::CreateAuthoredWorkflow(CreateAuthoredWorkflowRequest {
                document: authoring_document.clone(),
                draft_id: "draft-1".to_string(),
            }),
            Request::ApplyWorkflowDraftEdits(ApplyWorkflowDraftEditsRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                draft_id: "draft-1".to_string(),
                batch: bcode_workflow::WorkflowAuthoringEditBatch {
                    version: bcode_workflow::WORKFLOW_AUTHORING_EDIT_VERSION,
                    expected_generation: 1,
                    edits: vec![bcode_workflow::WorkflowAuthoringEdit::UpdateMetadata {
                        metadata: authoring_document.metadata.clone(),
                    }],
                },
                producer: authoring_document.producer.clone(),
            }),
            Request::UpdateWorkflowDraft(UpdateWorkflowDraftRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                draft_id: "draft-1".to_string(),
                expected_generation: 1,
                document: authoring_document.clone(),
                producer: authoring_document.producer.clone(),
            }),
            Request::PublishWorkflowDraft(PublishWorkflowDraftRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                draft_id: "draft-1".to_string(),
                expected_generation: 1,
                configuration: None,
                activate: true,
                expected_active_revision: None,
                control: WorkflowComputationControl::default(),
            }),
            Request::PublishAndStartWorkflow(Box::new(PublishAndStartWorkflowRequest {
                publication: PublishWorkflowDraftRequest {
                    workflow_id: authoring_document.workflow_id.clone(),
                    draft_id: "draft-1".to_string(),
                    expected_generation: 1,
                    configuration: None,
                    activate: true,
                    expected_active_revision: None,
                    control: WorkflowComputationControl::default(),
                },
                run_id: Some("run-1".to_string()),
                parent_session_id: bcode_session_models::SessionId::new(),
                workspace_snapshot: Some("snapshot-1".to_string()),
            })),
            Request::ImportWorkflowDraft(ImportWorkflowDraftRequest {
                bundle: bcode_workflow::WorkflowExportBundle {
                    version: bcode_workflow::WORKFLOW_EXPORT_BUNDLE_VERSION,
                    revision: bcode_workflow::WorkflowPortableRevision {
                        identity: bcode_workflow::WorkflowRevisionIdentity {
                            workflow_id: authoring_document.workflow_id.clone(),
                            revision: 1,
                        },
                        source_checksum_sha256: authoring_document
                            .source_digest_sha256()
                            .expect("source digest"),
                        executable_source_checksum_sha256: authoring_document
                            .executable_source_digest_sha256()
                            .expect("executable digest"),
                        definition_identity:
                            bcode_workflow::WorkflowDefinitionIdentity::for_definition(
                                authoring_document.workflow_id.clone(),
                                &definition,
                            )
                            .expect("identity"),
                        document: authoring_document.clone(),
                        producer: authoring_document.producer.clone(),
                        published_at_ms: 1,
                    },
                    dependencies: Vec::new(),
                },
                workflow_id: authoring_document.workflow_id.clone(),
                draft_id: "imported-draft".to_string(),
                collision_policy: WorkflowImportCollisionPolicy::RequireExistingWorkflowNewDraft,
                control: WorkflowComputationControl::default(),
            }),
            Request::ImportWorkflowRevision(ImportWorkflowRevisionRequest {
                bundle: bcode_workflow::WorkflowExportBundle {
                    version: bcode_workflow::WORKFLOW_EXPORT_BUNDLE_VERSION,
                    revision: bcode_workflow::WorkflowPortableRevision {
                        identity: bcode_workflow::WorkflowRevisionIdentity {
                            workflow_id: authoring_document.workflow_id.clone(),
                            revision: 1,
                        },
                        source_checksum_sha256: authoring_document
                            .source_digest_sha256()
                            .expect("source digest"),
                        executable_source_checksum_sha256: authoring_document
                            .executable_source_digest_sha256()
                            .expect("executable digest"),
                        definition_identity:
                            bcode_workflow::WorkflowDefinitionIdentity::for_definition(
                                authoring_document.workflow_id.clone(),
                                &definition,
                            )
                            .expect("identity"),
                        document: authoring_document.clone(),
                        producer: authoring_document.producer.clone(),
                        published_at_ms: 1,
                    },
                    dependencies: Vec::new(),
                },
                workflow_id: authoring_document.workflow_id.clone(),
                revision: 2,
                activate: true,
                expected_active_revision: Some(1),
                collision_policy:
                    WorkflowImportCollisionPolicy::RequireExistingWorkflowNextRevision,
                control: WorkflowComputationControl::default(),
            }),
            Request::ActivateWorkflowRevision(ActivateWorkflowRevisionRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                revision: 1,
                expected_active_revision: None,
            }),
            Request::SetAuthoredWorkflowArchived(SetAuthoredWorkflowArchivedRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                archived: true,
            }),
            Request::DiscardWorkflowDraft(DiscardWorkflowDraftRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                draft_id: "draft-1".to_string(),
                expected_generation: 1,
            }),
            Request::ForkWorkflowDraft(ForkWorkflowDraftRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                source: WorkflowDraftForkSource::Revision { revision: 1 },
                draft_id: "draft-2".to_string(),
                producer: authoring_document.producer.clone(),
            }),
            Request::CreateWorkflowPreset(CreateWorkflowPresetRequest {
                preset: WorkflowPresetMutation {
                    workflow_id: authoring_document.workflow_id.clone(),
                    preset_id: "default".to_string(),
                    revision: 1,
                    name: "Default".to_string(),
                    configuration: serde_json::json!({}),
                    run_limits: None,
                    producer: authoring_document.producer.clone(),
                },
            }),
            Request::UpdateWorkflowPreset(UpdateWorkflowPresetRequest {
                expected_generation: 1,
                preset: WorkflowPresetMutation {
                    workflow_id: authoring_document.workflow_id.clone(),
                    preset_id: "default".to_string(),
                    revision: 1,
                    name: "Updated".to_string(),
                    configuration: serde_json::json!({}),
                    run_limits: None,
                    producer: authoring_document.producer.clone(),
                },
            }),
            Request::DeleteWorkflowPreset(DeleteWorkflowPresetRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                preset_id: "default".to_string(),
                expected_generation: 2,
            }),
            Request::ExportWorkflowRevision(ExportWorkflowRevisionRequest {
                workflow_id: authoring_document.workflow_id.clone(),
                revision: 1,
            }),
            Request::PreviewWorkflowImport(PreviewWorkflowImportRequest {
                bundle: bcode_workflow::WorkflowExportBundle {
                    version: bcode_workflow::WORKFLOW_EXPORT_BUNDLE_VERSION,
                    revision: bcode_workflow::WorkflowPortableRevision {
                        identity: bcode_workflow::WorkflowRevisionIdentity {
                            workflow_id: authoring_document.workflow_id.clone(),
                            revision: 1,
                        },
                        source_checksum_sha256: authoring_document
                            .source_digest_sha256()
                            .expect("source digest"),
                        executable_source_checksum_sha256: authoring_document
                            .executable_source_digest_sha256()
                            .expect("executable digest"),
                        definition_identity:
                            bcode_workflow::WorkflowDefinitionIdentity::for_definition(
                                "authored/review",
                                &authoring_document.definition,
                            )
                            .expect("definition identity"),
                        document: authoring_document.clone(),
                        producer: authoring_document.producer.clone(),
                        published_at_ms: 1,
                    },
                    dependencies: Vec::new(),
                },
                target_workflow_id: "authored/imported".to_string(),
                control: WorkflowComputationControl::default(),
            }),
            Request::StartAuthoredWorkflow(StartAuthoredWorkflowRequest {
                selection: AuthoredWorkflowRunSelection::Revision {
                    workflow_id: authoring_document.workflow_id.clone(),
                    revision: 1,
                },
                run_id: Some("authored-run".to_string()),
                parent_session_id: SessionId::new(),
                workspace_snapshot: Some("snapshot".to_string()),
                parent_session_generation: None,
                configuration: Some(serde_json::json!({})),
                input: Some(serde_json::json!({"subject": "typed"})),
            }),
            Request::ListAuthoredWorkflows {
                cursor: None,
                limit: 25,
            },
            Request::InspectAuthoredWorkflow {
                workflow_id: "authored/review".to_string(),
                limit: 25,
            },
            Request::GetAuthoredWorkflow {
                workflow_id: "authored/review".to_string(),
            },
            Request::ListWorkflowDrafts {
                workflow_id: "authored/review".to_string(),
                cursor: None,
                limit: 25,
            },
            Request::GetWorkflowDraft {
                workflow_id: "authored/review".to_string(),
                draft_id: "draft-1".to_string(),
            },
            Request::ListWorkflowRevisions {
                workflow_id: "authored/review".to_string(),
                cursor: None,
                limit: 25,
            },
            Request::InspectWorkflowRevisionRequirements {
                workflow_id: authoring_document.workflow_id.clone(),
                revision: 1,
            },
            Request::GetWorkflowRevision {
                workflow_id: "authored/review".to_string(),
                revision: 1,
            },
            Request::ListWorkflowPresets {
                workflow_id: "authored/review".to_string(),
                cursor: None,
                limit: 25,
            },
            Request::GetWorkflowPreset {
                workflow_id: "authored/review".to_string(),
                preset_id: "default".to_string(),
            },
            Request::WorkflowAuthoringCatalog,
            Request::ValidateWorkflowAuthoring {
                document: authoring_document.clone(),
                control: WorkflowComputationControl::default(),
            },
            Request::PreviewWorkflowCompilation {
                document: authoring_document,
                configuration: Some(serde_json::json!({})),
                control: WorkflowComputationControl::default(),
            },
            Request::RegisterWorkflowDefinition(WorkflowDefinitionRegistrationRequest {
                definition_id: "review".to_string(),
                version: 1,
                definition: definition.clone(),
            }),
            Request::StartWorkflow(WorkflowStartRequest {
                identity: bcode_workflow::WorkflowDefinitionIdentity::for_definition(
                    "review",
                    &definition,
                )
                .expect("identity"),
                definition: definition.clone(),
                run_id: Some("bound-review-run".to_string()),
                workspace_snapshot: None,
                parent_session_id: SessionId::new(),
                input: serde_json::json!(1),
                binding: bcode_workflow_store::WorkflowRunBinding {
                    owner_plugin_id: "bcode.review".to_string(),
                    workflow_kind: "review".to_string(),
                    scope_key: "session-1".to_string(),
                    display_label: Some("Review".to_string()),
                    single_active: true,
                },
                limits: bcode_workflow_store::WorkflowRunLimits::default(),
            }),
            Request::StartWorkflowRun(WorkflowRunStartRequest {
                definition_id: "review".to_string(),
                definition_version: 1,
                run_id: Some("review-run".to_string()),
                workspace_snapshot: "snapshot-1".to_string(),
                parent_session_id: SessionId::new(),
                parent_session_generation: None,
                binding: None,
                input: Some(serde_json::json!(1)),
                limits: bcode_workflow_store::WorkflowRunLimits::default(),
            }),
            Request::ListWorkflowDefinitions { limit: 25 },
            Request::DescribeWorkflowDefinition {
                definition_id: "review".to_string(),
                version: 1,
            },
            Request::InspectWorkflowRun {
                run_id: "run-1".to_string(),
                limit: 25,
            },
            Request::WorkflowRunView {
                run_id: "run-1".to_string(),
                limit: 25,
            },
            Request::WorkflowCatalogView {
                request: bcode_workflow_view_models::WorkflowCatalogRequest {
                    limit: 25,
                    cursor: None,
                    filter: bcode_workflow_view_models::WorkflowCatalogFilter::All,
                    sort: bcode_workflow_view_models::WorkflowCatalogSort::UpdatedAt,
                    group: bcode_workflow_view_models::WorkflowCatalogGroup::None,
                    search: None,
                },
            },
            Request::WorkflowRunStatus {
                run_id: "run-1".to_string(),
            },
            Request::AssociatedWorkflowRun {
                key: WorkflowRunBindingLookup {
                    owner_plugin_id: "bcode.review".to_string(),
                    workflow_kind: "review".to_string(),
                    scope_key: "session-1".to_string(),
                },
            },
            Request::InspectAssociatedWorkflowRun {
                key: WorkflowRunBindingLookup {
                    owner_plugin_id: "bcode.review".to_string(),
                    workflow_kind: "review".to_string(),
                    scope_key: "session-1".to_string(),
                },
                limit: 25,
            },
            Request::ControlAssociatedWorkflowRun {
                key: WorkflowRunBindingLookup {
                    owner_plugin_id: "bcode.review".to_string(),
                    workflow_kind: "review".to_string(),
                    scope_key: "session-1".to_string(),
                },
                action: WorkflowRunControlAction::Pause,
            },
            Request::ListWorkflowRuns { limit: 25 },
            Request::WorkflowRunOutputs {
                run_id: "run-1".to_string(),
                limit: 25,
            },
            Request::CancelWorkflowRun {
                run_id: "run-1".to_string(),
            },
            Request::PauseWorkflowRun {
                run_id: "run-1".to_string(),
            },
            Request::ResumeWorkflowRun {
                run_id: "run-1".to_string(),
            },
            Request::RetryWorkflowNode {
                run_id: "run-1".to_string(),
                node_id: "review".to_string(),
                activation_id: "activation-1".to_string(),
                failed_attempt: 1,
            },
            Request::ListWorkflowWaits {
                run_id: "run-1".to_string(),
                limit: 25,
            },
            Request::ProvideWorkflowInput {
                run_id: "run-1".to_string(),
                node_id: "input".to_string(),
                activation_id: "activation-1".to_string(),
                value: serde_json::json!({"answer": 42}),
            },
            Request::ResolveWorkflowApproval {
                run_id: "run-1".to_string(),
                node_id: "approve".to_string(),
                activation_id: "activation-2".to_string(),
                approved: true,
            },
            Request::ListWorkflowMutationApprovalsAll { limit: 25 },
            Request::ListWorkflowMutationApprovals {
                run_id: "run-1".to_string(),
                limit: 25,
            },
            Request::ResolveWorkflowMutationApproval {
                approval_id: "approval-1".to_string(),
                decision: bcode_workflow::WorkflowMutationApprovalDecision::Approve,
            },
            Request::WorkflowAttemptHistory {
                run_id: "run-1".to_string(),
                cursor: Some(bcode_workflow_store::AttemptCursor {
                    prepared_at_ms: 10,
                    dispatch_identity: "dispatch-1".to_string(),
                }),
                limit: 25,
            },
            Request::WorkflowEventHistory {
                run_id: "run-1".to_string(),
                after_sequence: Some(10),
                limit: 25,
            },
            Request::WorkflowLiveEventCatchUp {
                after_sequence: 10,
                limit: 25,
            },
            Request::SubscribeWorkflowRuns,
        ];
        for request in requests {
            let encoded = encode_request(&request).expect("encode request");
            assert_eq!(decode_request(&encoded).expect("decode request"), request);
        }

        let response = Response::Ok(ResponsePayload::WorkflowAuthoringCatalog {
            catalog: bcode_workflow::WorkflowAuthoringCatalogSnapshot {
                version: bcode_workflow::WORKFLOW_AUTHORING_CATALOG_VERSION,
                capabilities: bcode_workflow::WorkflowAuthoringCapabilitySummary::from(
                    &bcode_workflow::WorkflowProductionCapabilities::current(),
                ),
                plugins: std::collections::BTreeSet::new(),
                blocks: std::collections::BTreeMap::new(),
                node_configuration_schemas: bcode_workflow::workflow_node_configuration_schemas(),
                workflow_definitions: BTreeMap::new(),
                agent_profiles: std::collections::BTreeSet::from(["build".to_string()]),
                authoring_actions: BTreeMap::new(),
            },
        });
        let encoded = encode_response(&response).expect("encode authoring catalog");
        assert_eq!(
            decode_response(&encoded).expect("decode authoring catalog"),
            response
        );

        let response = Response::Ok(ResponsePayload::WorkflowAuthoringValidated {
            report: bcode_workflow::WorkflowValidationReport {
                authoring_version: bcode_workflow::WORKFLOW_AUTHORING_DOCUMENT_VERSION,
                valid: false,
                source_digest_sha256: None,
                executable_source_digest_sha256: None,
                diagnostics: vec![bcode_workflow::WorkflowValidationDiagnostic {
                    code: "invalid_schema".to_string(),
                    severity: bcode_workflow::WorkflowValidationSeverity::Error,
                    document_path: "configuration_schema".to_string(),
                    message: "invalid schema".to_string(),
                    remediation: "correct the schema".to_string(),
                }],
            },
        });
        let encoded = encode_response(&response).expect("encode validation report");
        assert_eq!(
            decode_response(&encoded).expect("decode validation report"),
            response
        );

        let response = Response::Ok(ResponsePayload::WorkflowDefinitionRegistered {
            definition: bcode_workflow::StoredWorkflowDefinition {
                definition_id: "review".to_string(),
                version: 1,
                checksum_sha256: "checksum".to_string(),
                definition_json: serde_json::to_string(&definition).expect("definition"),
            },
        });
        let encoded = encode_response(&response).expect("encode response");
        assert_eq!(
            decode_response(&encoded).expect("decode response"),
            response
        );

        let response = Response::Ok(ResponsePayload::WorkflowRunList {
            runs: vec![bcode_workflow_store::WorkflowRunSummary {
                run_id: "run-1".to_string(),
                definition_id: "review".to_string(),
                definition_version: 1,
                workspace_snapshot: "snapshot-1".to_string(),
                parent_session_id: None,
                parent_session_generation: None,
                binding: None,
                authored_provenance: None,
                terminal_output_id: None,
                terminal_output_checksum_sha256: None,
                authorization_profile: bcode_workflow::WorkflowAuthorizationProfileIdentity {
                    version: 1,
                    provider_id: "test-policy".to_string(),
                    profile_id: "build".to_string(),
                    policy_digest_sha256: "a".repeat(64),
                },
                authorization_ceiling: bcode_workflow::WorkflowToolCapability::Mutating,
                status: bcode_workflow_store::RunStatus::Running,
                cancellation_requested_at_ms: None,
                created_at_ms: 1,
                updated_at_ms: 2,
            }],
        });
        let encoded = encode_response(&response).expect("encode response");
        assert_eq!(
            decode_response(&encoded).expect("decode response"),
            response
        );
    }

    #[test]
    fn session_search_contract_round_trips_portably() {
        use bcode_session_search::{
            FederatedProviderReport, FederatedSessionSearchResponse, HydratedSessionSearchHit,
            ProviderSearchOutcome, SearchContentKind, SearchField, SearchHitHydrationOutcome,
            SessionSearchContentRoute, SessionSearchFilters, SessionSearchHit,
            SessionSearchLocator, SessionSearchQuery, SessionSearchRequest, SessionSearchRouteMode,
            SessionSearchSort, TextMatchMode,
        };
        use std::collections::BTreeSet;

        let session_id = SessionId::new();
        let search_request = SessionSearchRequest {
            query: SessionSearchQuery::Text {
                text: "needle".to_owned(),
                mode: TextMatchMode::Terms,
                fields: BTreeSet::new(),
            },
            filters: SessionSearchFilters {
                content_kinds: std::iter::once(SearchContentKind::UserMessage).collect(),
                ..SessionSearchFilters::default()
            },
            sort: SessionSearchSort::ProviderRelevance,
            limit: 10,
            cursor: None,
            deadline_ms: Some(1000),
        };
        let request = Request::SessionSearch {
            request: search_request,
            policy: bcode_session_search::SessionSearchPlanPolicy::default(),
            routes: vec![SessionSearchContentRoute {
                content_kinds: std::iter::once(SearchContentKind::UserMessage).collect(),
                mode: SessionSearchRouteMode::Primary,
                provider_ids: vec!["provider".to_owned()],
            }],
            hydrate: true,
        };
        let decoded: Request =
            decode_typed_stable(&encode_typed_stable(&request).expect("encode")).expect("decode");
        assert_eq!(decoded, request);

        let hit = SessionSearchHit {
            locator: SessionSearchLocator {
                session_id,
                sequence: 1,
                record_id: Some("1:user-message:0".to_owned()),
            },
            content_kind: SearchContentKind::UserMessage,
            matched_field: SearchField::Text,
            provider_id: "provider".to_owned(),
            provider_rank: 0,
            provider_score: Some("bm25:1.25".to_owned()),
            preview: Some("needle".to_owned()),
            preview_truncated: false,
        };
        let response = Response::Ok(ResponsePayload::SessionSearch {
            response: FederatedSessionSearchResponse {
                hits: vec![hit.clone()],
                query_complete: true,
                coverage_complete: true,
                providers: vec![FederatedProviderReport {
                    provider_id: "provider".to_owned(),
                    outcome: ProviderSearchOutcome::Complete,
                    elapsed_ms: 2,
                    query_complete: true,
                    coverage_complete: true,
                    next_cursor: None,
                    searched_content: vec![SearchContentKind::UserMessage],
                    excluded_content: Vec::new(),
                }],
                failures: Vec::new(),
            },
            hydrated_hits: vec![HydratedSessionSearchHit {
                hit,
                outcome: SearchHitHydrationOutcome::StaleLocator,
                event: None,
                message: Some("canonical event locator is no longer present".to_owned()),
            }],
        });
        let decoded: Response =
            decode_typed_stable(&encode_typed_stable(&response).expect("encode")).expect("decode");
        assert_eq!(decoded, response);
    }

    #[test]
    fn session_compatibility_inventory_contract_round_trips_portably() {
        use std::collections::{BTreeMap, BTreeSet};

        let session_id = SessionId::new();
        let request = Request::SessionCompatibilityInventory {
            request: SessionCompatibilityInventoryRequest {
                session_ids: BTreeSet::from([session_id]),
                after_timestamp_ms: Some(10),
                before_timestamp_ms: Some(20),
                cursor: Some(SessionCompatibilityCursor {
                    updated_at_ms: 12,
                    session_id,
                }),
                limit: 25,
            },
        };
        let decoded: Request =
            decode_typed_stable(&encode_typed_stable(&request).expect("encode inventory request"))
                .expect("decode inventory request");
        assert_eq!(decoded, request);

        let response = Response::Ok(ResponsePayload::SessionCompatibilityInventory {
            response: SessionCompatibilityInventoryResponse {
                entries: vec![SessionCompatibilityEntry {
                    session_id,
                    updated_at_ms: 12,
                    category: SessionCompatibilityCategory::MigrationRequired,
                    action: SessionCompatibilityAction::Migrate,
                    retryable: false,
                    source_writer_epoch: Some(5),
                    first_historical_event_schema: Some(1),
                    message: Some("explicit migration required".to_owned()),
                }],
                counts: BTreeMap::from([(SessionCompatibilityCategory::MigrationRequired, 1)]),
                category_summaries: BTreeMap::from([(
                    SessionCompatibilityCategory::MigrationRequired,
                    SessionCompatibilityCategorySummary {
                        count: 1,
                        action: SessionCompatibilityAction::Migrate,
                        retryable: false,
                        samples: vec![SessionCompatibilitySample {
                            session_id,
                            message: Some("explicit migration required".to_owned()),
                        }],
                    },
                )]),
                next_cursor: Some(SessionCompatibilityCursor {
                    updated_at_ms: 12,
                    session_id,
                }),
            },
        });
        let decoded: Response = decode_typed_stable(
            &encode_typed_stable(&response).expect("encode inventory response"),
        )
        .expect("decode inventory response");
        assert_eq!(decoded, response);
    }

    #[test]
    fn session_bulk_migration_contract_round_trips_portably() {
        use std::collections::BTreeSet;

        let session_id = SessionId::new();
        let request = Request::SessionBulkMigrationStart {
            request: SessionBulkMigrationStartRequest {
                mode: SessionBulkMigrationMode::Migrate,
                session_ids: BTreeSet::from([session_id]),
                after_timestamp_ms: Some(10),
                before_timestamp_ms: Some(20),
                confirmation: Some(SESSION_BULK_MIGRATION_CONFIRMATION.to_owned()),
            },
        };
        let decoded: Request = decode_typed_stable(
            &encode_typed_stable(&request).expect("encode bulk migration request"),
        )
        .expect("decode bulk migration request");
        assert_eq!(decoded, request);

        let response = Response::Ok(ResponsePayload::SessionBulkMigrationOperation {
            status: SessionBulkMigrationOperationStatus {
                operation_id: "transient-operation".to_owned(),
                revision: 2,
                state: SessionBulkMigrationState::NeedsAttention,
                mode: SessionBulkMigrationMode::Migrate,
                selected: 2,
                visited: 3,
                migrated: 1,
                blocked: 1,
                failed: 0,
                current_session_id: None,
                outcomes: vec![SessionBulkMigrationOutcome {
                    session_id,
                    category: SessionCompatibilityCategory::OwnerBlocked,
                    action: SessionCompatibilityAction::Retry,
                    message: Some("owner blocked".to_owned()),
                }],
            },
        });
        let decoded: Response = decode_typed_stable(
            &encode_typed_stable(&response).expect("encode bulk migration response"),
        )
        .expect("decode bulk migration response");
        assert_eq!(decoded, response);
    }

    #[test]
    fn workflow_store_reset_contract_round_trips_portably() {
        let request = Request::ResetIncompatibleWorkflowStore {
            confirm: bcode_workflow_store::WORKFLOW_STORE_RESET_CONFIRMATION.to_string(),
        };
        let decoded: Request =
            decode_typed_stable(&encode_typed_stable(&request).expect("encode reset request"))
                .expect("decode reset request");
        assert_eq!(decoded, request);
        let receipt = bcode_workflow_store::WorkflowStoreResetReceipt {
            version: bcode_workflow_store::WORKFLOW_STORE_RESET_RECEIPT_VERSION,
            previous_schema_version: Some(9),
            new_schema_version: bcode_workflow_store::WORKFLOW_STORE_SCHEMA_VERSION,
            backup_path: PathBuf::from("workflows/reset-backups/workflow-42"),
            backup_sha256: "a".repeat(64),
            removed_artifact_entries: 2,
            reset_at_ms: 42,
        };
        let response = Response::Ok(ResponsePayload::WorkflowStoreReset { receipt });
        let decoded: Response =
            decode_typed_stable(&encode_typed_stable(&response).expect("encode reset response"))
                .expect("decode reset response");
        assert_eq!(decoded, response);
    }

    #[test]
    fn session_search_maintenance_contract_round_trips_portably() {
        let purge = Request::SessionSearchPurge {
            provider_id: "bcode.tantivy-session-search".to_owned(),
            confirmation: "purge-bcode.tantivy-session-search".to_owned(),
        };
        let decoded: Request =
            decode_typed_stable(&encode_typed_stable(&purge).expect("encode")).expect("decode");
        assert_eq!(decoded, purge);
        let rebuild = Request::SessionSearchRebuild {
            provider_id: "bcode.tantivy-session-search".to_owned(),
            confirmation: "rebuild-bcode.tantivy-session-search".to_owned(),
        };
        let decoded: Request =
            decode_typed_stable(&encode_typed_stable(&rebuild).expect("encode rebuild"))
                .expect("decode rebuild");
        assert_eq!(decoded, rebuild);
        let backfill = Request::SessionSearchBackfill {
            request: bcode_session_search::BackfillSessionSearchRequest {
                provider_id: "bcode.tantivy-session-search".to_owned(),
                session_ids: std::iter::once(SessionId::new()).collect(),
                after_timestamp_ms: None,
                before_timestamp_ms: None,
                cursor: None,
                deadline_ms: 30_000,
            },
        };
        let decoded: Request =
            decode_typed_stable(&encode_typed_stable(&backfill).expect("encode backfill"))
                .expect("decode backfill");
        assert_eq!(decoded, backfill);

        let maintenance = Response::Ok(ResponsePayload::SessionSearchMaintenance {
            response: bcode_session_search::SessionSearchMaintenanceResponse {
                provider_id: "bcode.tantivy-session-search".to_owned(),
                operation: bcode_session_search::OP_PURGE.to_owned(),
                status: bcode_session_search::SessionSearchStatus {
                    provider_id: "bcode.tantivy-session-search".to_owned(),
                    state: bcode_session_search::SearchProviderState::Disabled,
                    record_schema_version: bcode_session_search::CURRENT_SEARCH_RECORD_VERSION,
                    normalization_version: bcode_session_search::CURRENT_NORMALIZATION_VERSION,
                    policy_version: bcode_session_search::CURRENT_SEARCH_POLICY_VERSION,
                    index_bytes: 0,
                    quota_bytes: 1024,
                    document_count: 0,
                    pending_sessions: 0,
                    coverage: Vec::new(),
                    degraded_reason: None,
                },
            },
        });
        let decoded: Response =
            decode_typed_stable(&encode_typed_stable(&maintenance).expect("encode maintenance"))
                .expect("decode maintenance");
        assert_eq!(decoded, maintenance);
    }

    #[test]
    fn session_inspection_contract_round_trips() {
        let session_id = SessionId::new();
        let query = bcode_session_models::SessionInspectionQuery {
            category: bcode_session_models::SessionInspectionCategory::FailedToolCalls,
            cursor: Some(bcode_session_models::SessionHistoryCursor { sequence: 10 }),
            limit: 25,
            direction: bcode_session_models::SessionHistoryDirection::Forward,
        };
        let request = Request::SessionInspection { session_id, query };
        let encoded = encode_request(&request).expect("encode request");
        assert_eq!(decode_request(&encoded).expect("decode request"), request);

        let response = Response::Ok(ResponsePayload::SessionInspection {
            page: bcode_session_models::SessionInspectionPage {
                session_id,
                category: query.category,
                events: Vec::new(),
                scanned_events: 0,
                compatibility_issues: Vec::new(),
                next_cursor: None,
                has_more: false,
            },
        });
        let encoded = encode_response(&response).expect("encode response");
        assert_eq!(
            decode_response(&encoded).expect("decode response"),
            response
        );
    }

    #[test]
    #[allow(clippy::too_many_lines)]
    fn workflow_package_computation_contract_round_trips() {
        let manifest = bcode_workflow::WorkflowPackageManifest {
            version: bcode_workflow::WORKFLOW_PACKAGE_MANIFEST_VERSION,
            package_id: "example/package".to_string(),
            exports: std::collections::BTreeMap::from([("main".to_string(), "main".to_string())]),
            external_dependencies: std::collections::BTreeMap::new(),
            imports: Vec::new(),
            members: vec![bcode_workflow::WorkflowPackageMember {
                member_id: "main".to_string(),
                source_name: "main.json".to_string(),
                format: bcode_workflow::WorkflowSourceFormat::Json,
                source: "{}".to_string(),
                dependencies: Vec::new(),
                external_dependencies: Vec::new(),
            }],
        };
        let request = Request::ValidateWorkflowPackage(WorkflowPackageComputationRequest {
            closure: bcode_workflow::WorkflowPackageClosure {
                version: bcode_workflow::WORKFLOW_PACKAGE_CLOSURE_VERSION,
                entry_package_id: manifest.package_id.clone(),
                packages: vec![bcode_workflow::WorkflowPackageClosureSource {
                    package_id: manifest.package_id.clone(),
                    source_name: Some("package.workflow-package.json".to_string()),
                    manifest,
                }],
            },
            control: WorkflowComputationControl::default(),
        });
        let encoded = encode_request(&request).expect("encode package request");
        assert_eq!(
            decode_request(&encoded).expect("decode package request"),
            request
        );

        let preview = Request::PreviewWorkflowPackage(WorkflowPackagePreviewRequest {
            plan: bcode_workflow::WorkflowPackagePlan {
                version: bcode_workflow::WORKFLOW_PACKAGE_PLAN_VERSION,
                package_id: "example/package".to_string(),
                members: Vec::new(),
                lock: bcode_workflow::WorkflowPackageLock {
                    version: bcode_workflow::WORKFLOW_PACKAGE_LOCK_VERSION,
                    package_id: "example/package".to_string(),
                    imports: Vec::new(),
                    exports: Vec::new(),
                    package_source_digest_sha256: "a".repeat(64),
                    members: Vec::new(),
                },
            },
            dependency_plans: Vec::new(),
            configurations: std::collections::BTreeMap::new(),
            control: WorkflowComputationControl::default(),
        });
        let encoded = encode_request(&preview).expect("encode package preview");
        assert_eq!(
            decode_request(&encoded).expect("decode package preview"),
            preview
        );

        let apply = Request::ApplyWorkflowPackage(ApplyWorkflowPackageRequest {
            request: bcode_workflow::WorkflowPackageApplyRequest {
                version: bcode_workflow::WORKFLOW_PACKAGE_MUTATION_VERSION,
                plan: bcode_workflow::WorkflowPackagePlan {
                    version: bcode_workflow::WORKFLOW_PACKAGE_PLAN_VERSION,
                    package_id: "example/package".to_string(),
                    members: Vec::new(),
                    lock: bcode_workflow::WorkflowPackageLock {
                        version: bcode_workflow::WORKFLOW_PACKAGE_LOCK_VERSION,
                        package_id: "example/package".to_string(),
                        imports: Vec::new(),
                        exports: Vec::new(),
                        package_source_digest_sha256: "a".repeat(64),
                        members: Vec::new(),
                    },
                },
                expected_generations: Vec::new(),
            },
            applied_at_ms: 10,
        });
        let encoded = encode_request(&apply).expect("encode package apply");
        assert_eq!(
            decode_request(&encoded).expect("decode package apply"),
            apply
        );

        let publish = Request::PublishWorkflowPackage(PublishWorkflowPackageRequest {
            request: bcode_workflow::WorkflowPackagePublishRequest {
                version: bcode_workflow::WORKFLOW_PACKAGE_MUTATION_VERSION,
                package_id: "example/package".to_string(),
                expected_lock: bcode_workflow::WorkflowPackageLock {
                    version: bcode_workflow::WORKFLOW_PACKAGE_LOCK_VERSION,
                    package_id: "example/package".to_string(),
                    imports: Vec::new(),
                    exports: Vec::new(),
                    package_source_digest_sha256: "a".repeat(64),
                    members: Vec::new(),
                },
                expected_generations: Vec::new(),
            },
            published_at_ms: 11,
        });
        let encoded = encode_request(&publish).expect("encode package publish");
        assert_eq!(
            decode_request(&encoded).expect("decode package publish"),
            publish
        );
        let publication = Request::GetWorkflowPackagePublication {
            package_id: "example/package".to_string(),
        };
        let encoded = encode_request(&publication).expect("encode package publication read");
        assert_eq!(
            decode_request(&encoded).expect("decode package publication read"),
            publication
        );
        let receipt = bcode_workflow::WorkflowPackagePublicationReceipt {
            version: bcode_workflow::WORKFLOW_PACKAGE_PUBLICATION_RECEIPT_VERSION,
            package_id: "example/package".to_string(),
            package_lock_digest_sha256: "a".repeat(64),
            published_at_ms: 11,
            exports: Vec::new(),
        };
        let response = Response::Ok(ResponsePayload::WorkflowPackagePublication {
            receipt: Some(receipt),
        });
        let encoded = encode_response(&response).expect("encode package publication response");
        assert_eq!(
            decode_response(&encoded).expect("decode package publication response"),
            response
        );

        let start = Request::StartWorkflowPackageExport(StartWorkflowPackageExportRequest {
            package_export: bcode_workflow::WorkflowPackageExportIdentity {
                package_id: "example/package".to_string(),
                export: "main".to_string(),
                package_lock_digest_sha256: Some("a".repeat(64)),
            },
            run_id: Some("package-run".to_string()),
            parent_session_id: SessionId::new(),
            workspace_snapshot: Some("workspace".to_string()),
            parent_session_generation: Some(1),
            configuration: Some(serde_json::json!({"mode": "safe"})),
            input: Some(serde_json::json!({"subject": "change"})),
        });
        let encoded = encode_request(&start).expect("encode package export start");
        assert_eq!(
            decode_request(&encoded).expect("decode package export start"),
            start
        );
    }

    #[test]
    fn workflow_source_apply_contract_round_trips() {
        let request = Request::ApplyWorkflowSource(ApplyWorkflowSourceRequest {
            source_format: bcode_workflow::WorkflowSourceFormat::Yaml,
            source: "workflow_source_version: 3\nworkflow_id: example/check\ntitle: Check\nsteps:\n  - run: true\n".to_string(),
            draft_id: bcode_workflow::DEFAULT_WORKFLOW_SOURCE_DRAFT_ID.to_string(),
        });
        let encoded = encode_request(&request).expect("encode source apply request");
        assert_eq!(
            decode_request(&encoded).expect("decode source apply request"),
            request
        );

        let result = bcode_workflow::WorkflowSourceApplyResult {
            version: bcode_workflow::WORKFLOW_SOURCE_APPLY_RESULT_VERSION,
            source_format: bcode_workflow::WorkflowSourceFormat::Yaml,
            source_profile: bcode_workflow::WorkflowSourceProfile::Structured,
            workflow_id: "example/check".to_string(),
            draft_id: bcode_workflow::DEFAULT_WORKFLOW_SOURCE_DRAFT_ID.to_string(),
            generation: 1,
            canonical_digest_sha256: "digest".to_string(),
            validation: bcode_workflow::WorkflowValidationReport {
                authoring_version: bcode_workflow::WORKFLOW_AUTHORING_DOCUMENT_VERSION,
                valid: true,
                source_digest_sha256: Some("source-digest".to_string()),
                executable_source_digest_sha256: Some("executable-digest".to_string()),
                diagnostics: Vec::new(),
            },
            source_map: bcode_workflow::WorkflowSourceMap {
                version: bcode_workflow::WORKFLOW_SOURCE_MAP_VERSION,
                entries: Vec::new(),
            },
            requirements: bcode_workflow::WorkflowRequirementSummary::default(),
            effects: bcode_workflow::WorkflowEffectSummary::default(),
            permissions: bcode_workflow::WorkflowPermissionPreview::default(),
            outcome: bcode_workflow::WorkflowSourceApplyOutcome::Created,
        };
        let response = Response::Ok(ResponsePayload::WorkflowSourceApplied { result });
        let encoded = encode_response(&response).expect("encode source apply response");
        assert_eq!(
            decode_response(&encoded).expect("decode source apply response"),
            response
        );
    }

    #[test]
    fn session_history_around_contract_round_trips() {
        let session_id = SessionId::new();
        let query = bcode_session_models::SessionHistoryAroundQuery {
            sequence: 42,
            before: 5,
            after: 7,
        };
        let request = Request::SessionHistoryAround { session_id, query };
        let encoded = encode_request(&request).expect("encode request");
        assert_eq!(decode_request(&encoded).expect("decode request"), request);

        let response = Response::Ok(ResponsePayload::SessionHistoryAround {
            window: bcode_session_models::SessionHistoryWindow {
                session_id,
                requested_sequence: 42,
                events: Vec::new(),
                anchor_present: false,
                first_available_sequence: Some(1),
                last_available_sequence: Some(100),
                compatibility_issues: Vec::new(),
            },
        });
        let encoded = encode_response(&response).expect("encode response");
        assert_eq!(
            decode_response(&encoded).expect("decode response"),
            response
        );
    }

    #[test]
    fn permission_batch_resolution_request_round_trips() {
        let request = Request::ResolvePermissionBatch {
            batch_id: "permission-batch-4".to_string(),
            approved: true,
        };
        let encoded = encode(&request).expect("batch resolution request should encode");
        let decoded: Request = decode(&encoded).expect("batch resolution request should decode");
        assert_eq!(decoded, request);
    }

    #[test]
    fn invocation_input_request_round_trips_with_opaque_payload() {
        let request = Request::InvocationInput {
            session_id: SessionId::new(),
            input: bcode_tool::ToolInvocationInput {
                invocation_id: "call-2".to_owned(),
                input_id: "resize-132x40".to_owned(),
                producer_id: "bcode.shell".to_owned(),
                schema: "bcode.shell.invocation-input".to_owned(),
                schema_version: 1,
                payload: serde_json::json!({
                    "unknown": {"nested": [1, 2, 3]},
                }),
            },
        };
        let encoded = encode_request(&request).expect("invocation input request should encode");
        let decoded = decode_request(&encoded).expect("invocation input request should decode");
        assert_eq!(decoded, request);
    }

    #[test]
    fn endpoint_override_rejects_stale_daemon_context() {
        assert!(!endpoint_override_allowed(None, true, "current"));
        assert!(!endpoint_override_allowed(
            Some("previous"),
            true,
            "current"
        ));
        assert!(endpoint_override_allowed(Some("current"), true, "current"));
        assert!(endpoint_override_allowed(None, false, "current"));
    }

    #[test]
    fn current_protocol_message_responses_and_request_round_trip() {
        let responses = [
            Response::Ok(ResponsePayload::MessageSent),
            Response::Ok(ResponsePayload::TurnCancellationRequested { cancelled: true }),
            Response::Ok(ResponsePayload::MessageAccepted {
                queued: true,
                queue_position: Some(2),
            }),
        ];
        for response in responses {
            let encoded = encode_response(&response).expect("response should encode");
            assert_eq!(
                decode_response(&encoded).expect("response should decode"),
                response
            );
        }

        let request = Request::SendUserMessage {
            session_id: "00000000-0000-0000-0000-000000000001"
                .parse()
                .expect("fixture session id should parse"),
            text: "hello".to_string(),
        };
        let encoded = encode_request(&request).expect("request should encode");
        assert_eq!(
            decode_request(&encoded).expect("request should decode"),
            request
        );
    }

    #[tokio::test]
    async fn historical_payloads_are_rejected_before_decoding() {
        for (version, fixture) in [
            (37, "fixtures/ipc/v37/response_message_sent.hex"),
            (40, "fixtures/ipc/v40/response_message_sent.hex"),
            (41, "fixtures/ipc/v40/response_message_sent.hex"),
        ] {
            let envelope = Envelope {
                version: ProtocolVersion(version),
                request_id: 1,
                kind: EnvelopeKind::Response,
                payload: fixture_bytes(fixture),
            };
            let encoded = encode(&envelope).expect("envelope");
            let mut frame = u32::try_from(encoded.len())
                .expect("frame length")
                .to_le_bytes()
                .to_vec();
            frame.extend_from_slice(&encoded);
            let error = read_envelope_frame(&mut std::io::Cursor::new(frame))
                .await
                .expect_err("historical protocol rejected");
            assert!(matches!(
                error,
                CodecError::UnsupportedVersion {
                    actual,
                    expected: CURRENT_PROTOCOL_VERSION
                } if actual == version
            ));
        }
    }

    #[test]
    fn strict_prompt_admission_round_trips_and_legacy_defaults_to_fallback() {
        let request = Request::SubmitPrompt {
            session_id: SessionId::new(),
            submission_id: bcode_session_models::SubmissionId("strict-retry".to_owned()),
            text: "correction".to_owned(),
            placement: bcode_session_models::SubmissionPlacement::Steer,
            admission: bcode_session_models::TurnAdmissionMetadata::default(),
            require_active_turn: true,
        };
        let encoded = encode_request(&request).expect("encode strict steering");
        assert_eq!(
            decode_request(&encoded).expect("decode strict steering"),
            request
        );
        let mut legacy = serde_json::to_value(&request).expect("request value");
        legacy["submit_prompt"]
            .as_object_mut()
            .expect("prompt payload")
            .remove("require_active_turn");
        assert!(matches!(
            serde_json::from_value::<Request>(legacy).expect("legacy prompt"),
            Request::SubmitPrompt {
                require_active_turn: false,
                ..
            }
        ));
    }

    #[test]
    fn submission_receipt_and_bounded_lookup_round_trip() {
        let id = bcode_session_models::SubmissionId("retry-1".to_owned());
        let request = Request::GetSubmission {
            session_id: SessionId::new(),
            submission_id: id.clone(),
        };
        let encoded = encode(&request).unwrap();
        let decoded: Request = decode(&encoded).unwrap();
        assert!(
            matches!(decoded, Request::GetSubmission { submission_id, .. } if submission_id == id)
        );
        let response = Response::Ok(ResponsePayload::SubmissionAccepted {
            receipt: bcode_session_models::SubmissionReceipt {
                submission_id: id.clone(),
                accepted_event_sequence: 42,
            },
        });
        let encoded = encode(&response).unwrap();
        let decoded: Response = decode(&encoded).unwrap();
        assert!(
            matches!(decoded, Response::Ok(ResponsePayload::SubmissionAccepted { receipt }) if receipt.submission_id == id && receipt.accepted_event_sequence == 42)
        );
    }

    #[test]
    fn targeted_cancellation_preserves_identity_and_legacy_defaults() {
        let session_id = SessionId::new();
        let request = Request::CancelSessionTurn {
            session_id,
            expected_turn_id: Some("turn-具体-1".to_owned()),
            clear_queue: false,
        };
        let encoded = encode_request(&request).expect("encode cancellation");
        assert_eq!(
            decode_request(&encoded).expect("decode cancellation"),
            request
        );

        let mut legacy = serde_json::to_value(&request).expect("request value");
        legacy["cancel_session_turn"]
            .as_object_mut()
            .expect("request payload")
            .remove("expected_turn_id");
        assert_eq!(
            serde_json::from_value::<Request>(legacy.clone()).expect("legacy cancellation"),
            Request::CancelSessionTurn {
                session_id,
                expected_turn_id: None,
                clear_queue: false
            }
        );
        legacy["cancel_session_turn"]
            .as_object_mut()
            .expect("request payload")
            .insert("expected_turn_id".to_owned(), serde_json::json!(42));
        assert!(serde_json::from_value::<Request>(legacy).is_err());
    }

    #[test]
    fn generic_turn_admission_request_and_response_round_trip() {
        let session_id: SessionId = "00000000-0000-0000-0000-000000000001"
            .parse()
            .expect("session id should parse");
        let admission = bcode_session_models::TurnAdmissionMetadata {
            origin: Some(bcode_session_models::TurnOrigin {
                producer: "test.producer".to_string(),
                correlation_id: Some("operation-1".to_string()),
                display_label: Some("Background pass 1".to_string()),
            }),
            idempotency_key: Some("operation-1".to_string()),
            ..bcode_session_models::TurnAdmissionMetadata::default()
        };
        let request = Request::SubmitTurn {
            session_id,
            text: "continue".to_string(),
            admission,
        };
        let encoded = encode_request(&request).expect("request should encode");
        let decoded = decode_request(&encoded).expect("request should decode");
        assert_eq!(decoded, request);

        let response = Response::Ok(ResponsePayload::TurnAdmission {
            admission: bcode_session_models::TurnAdmission::Accepted(
                bcode_session_models::TurnReceipt::from_accepted_event(session_id, 42),
            ),
        });
        let encoded = encode_response(&response).expect("response should encode");
        let decoded = decode_response(&encoded).expect("response should decode");
        assert_eq!(decoded, response);
    }

    #[test]
    fn prompt_placement_domain_contract_preserves_ipc_encoding() {
        let session_id = SessionId::new();
        for placement in [
            bcode_session_models::PromptPlacement::Steering,
            bcode_session_models::PromptPlacement::FollowUp,
        ] {
            let request = Request::SendUserMessageWithPlacement {
                session_id,
                text: "continue".to_owned(),
                placement,
            };
            let encoded = encode_request(&request).expect("encode request");
            assert_eq!(decode_request(&encoded).expect("decode request"), request);
            let mut future = serde_json::to_value(&request).expect("request JSON");
            future["placement"] = serde_json::json!("future_placement");
            assert!(serde_json::from_value::<Request>(future).is_err());
        }
    }

    #[test]
    fn user_message_execution_request_round_trips_reasoning_options() {
        let request = Request::SendUserMessageWithExecution {
            session_id: SessionId::new(),
            text: "continue".to_owned(),
            placement: PromptPlacement::FollowUp,
            execution: bcode_session_models::TurnExecutionOptions {
                reasoning: Some(Box::new(bcode_session_models::TurnReasoningOptions {
                    effort: Some("high".to_owned()),
                    summary: Some("detailed".to_owned()),
                })),
                ..bcode_session_models::TurnExecutionOptions::default()
            },
        };

        let encoded = encode_request(&request).expect("request should encode");
        let decoded = decode_request(&encoded).expect("request should decode");
        assert_eq!(decoded, request);
    }

    #[test]
    fn skill_invocation_execution_request_round_trips_reasoning_options() {
        let request = Request::InvokeSkillWithExecution {
            session_id: SessionId::new(),
            skill_id: SkillId::new("review"),
            arguments: "staged".to_owned(),
            display_text: "Invoke skill review: staged".to_owned(),
            execution: bcode_session_models::TurnExecutionOptions {
                reasoning: Some(Box::new(bcode_session_models::TurnReasoningOptions {
                    effort: Some("high".to_owned()),
                    summary: Some("detailed".to_owned()),
                })),
                ..bcode_session_models::TurnExecutionOptions::default()
            },
        };

        let encoded = encode_request(&request).expect("request should encode");
        let decoded = decode_request(&encoded).expect("request should decode");
        assert_eq!(decoded, request);
    }

    #[test]
    fn attached_projection_window_round_trips() {
        let session_id = SessionId::new();
        let response = Response::Ok(ResponsePayload::Attached {
            session_id,
            session: SessionSummary {
                parent_id: None,
                id: session_id,
                name: None,
                explicit_name: None,
                derived_title: None,
                automatic_title: None,
                automatic_title_source_sequence: None,
                title_source: bcode_session_models::SessionTitleSource::EmptyDraft,
                client_count: 1,
                created_at_ms: 10,
                updated_at_ms: 20,
                working_directory: "/tmp/bcode-window-test".into(),
                import: None,
                execution: None,
                location: None,
            },
            history: Vec::new(),
            input_history: Vec::new(),
            usage_summary: None,
            import_warnings: Vec::new(),
            draft: None,
            runtime_selection: SessionRuntimeSelection::default(),
            projection_window: Some(bcode_session_models::ProjectionWindow {
                projection: bcode_session_models::SessionProjectionKind::Transcript,
                transcript_items: Vec::new(),
                source_range: None,
                has_older: true,
                has_newer: false,
                scanned_events: 4,
            }),
        });

        let encoded = encode(&response).expect("response should encode");
        let decoded: Response = decode(&encoded).expect("response should decode");

        assert_eq!(decoded, response);
    }

    #[test]
    fn attached_response_carries_canonical_session_summary() {
        let session_id: SessionId = "00000000-0000-0000-0000-000000000001"
            .parse()
            .expect("session id should parse");
        let summary = SessionSummary {
            parent_id: None,
            id: session_id,
            name: Some("Canonical title".to_owned()),
            explicit_name: Some("Canonical title".to_owned()),
            derived_title: None,
            automatic_title: None,
            automatic_title_source_sequence: None,
            title_source: bcode_session_models::SessionTitleSource::Explicit,
            client_count: 1,
            created_at_ms: 10,
            updated_at_ms: 20,
            working_directory: "/tmp/bcode-ipc-test".into(),
            import: None,
            execution: None,
            location: None,
        };
        let response = Response::Ok(ResponsePayload::Attached {
            session_id,
            session: summary.clone(),
            history: Vec::new(),
            input_history: Vec::new(),
            usage_summary: None,
            import_warnings: Vec::new(),
            draft: Some("draft text".to_owned()),
            runtime_selection: SessionRuntimeSelection::default(),
            projection_window: None,
        });

        let encoded = encode(&response).expect("response should encode");
        let decoded: Response = decode(&encoded).expect("response should decode");

        assert_eq!(decoded, response);
        let Response::Ok(ResponsePayload::Attached { session, .. }) = decoded else {
            panic!("decoded response should be attached");
        };
        assert_eq!(session, summary);
    }

    #[test]
    fn turn_observation_roundtrips_as_bounded_codec_safe_snapshot() {
        let request = Request::TurnJudgementObservation {
            session_id: SessionId::new(),
            turn_id: "exact-turn".into(),
        };
        let encoded = serde_json::to_vec(&request).unwrap();
        let decoded: Request = serde_json::from_slice(&encoded).unwrap();
        assert_eq!(decoded, request);
        let response = Response::Ok(ResponsePayload::TurnJudgementObservation {
            observation_json: None,
        });
        let encoded = serde_json::to_vec(&response).unwrap();
        let decoded: Response = serde_json::from_slice(&encoded).unwrap();
        assert_eq!(decoded, response);
    }

    #[test]
    fn optional_judgement_and_verification_round_trip() {
        for request in [
            Request::OptionalJudge {
                request_json: "{}".into(),
            },
            Request::JudgementVerify {
                selection_json: "{}".into(),
            },
        ] {
            assert_eq!(
                decode_request(&encode_request(&request).unwrap()).unwrap(),
                request
            );
        }
        for response in [
            Response::Ok(ResponsePayload::OptionalJudgement {
                result_json: "{\"response\":null}".into(),
            }),
            Response::Ok(ResponsePayload::JudgementVerified {
                result_json: "{}".into(),
            }),
        ] {
            assert_eq!(
                decode_response(&encode_response(&response).unwrap()).unwrap(),
                response
            );
        }
    }

    #[test]
    fn judgement_setup_validation_round_trips_as_bounded_json() {
        let selection = bcode_model::judgement::SetupSelection {
            provider_plugin_id: "bcode.jev".into(),
            model_id: "jev-1.13.0".into(),
            auth_profile: "jev".into(),
        };
        let request = Request::JudgementSetupValidate {
            selection_json: serde_json::to_string(&selection).unwrap(),
        };
        assert_eq!(
            decode_request(&encode_request(&request).unwrap()).unwrap(),
            request
        );
        let response = Response::Ok(ResponsePayload::JudgementSetupValidated { available: true });
        assert_eq!(
            decode_response(&encode_response(&response).unwrap()).unwrap(),
            response
        );
    }

    #[test]
    fn judgement_setup_round_trips_as_bounded_json() {
        let request = Request::JudgementSetup;
        assert_eq!(
            decode_request(&encode_request(&request).unwrap()).unwrap(),
            request
        );
        let setup = bcode_model::judgement::SetupCatalog { providers: vec![] };
        let response = Response::Ok(ResponsePayload::JudgementSetup {
            catalog_json: serde_json::to_string(&setup).unwrap(),
        });
        assert_eq!(
            decode_response(&encode_response(&response).unwrap()).unwrap(),
            response
        );
    }

    #[test]
    fn judgement_request_and_response_round_trip_without_provider_secrets() {
        let request = Request::Judge {
            provider_plugin_id: "example.provider".into(),
            auth_profile: "example".into(),
            request_json: serde_json::to_string(&bcode_model::judgement::Request {
                model_id: "example.model".into(),
                state: bcode_model::judgement::State::Text("state".into()),
                questions: std::collections::BTreeMap::from([(
                    "check".into(),
                    bcode_model::judgement::Question::YesNo {
                        instructions: "Is it relevant?".into(),
                    },
                )]),
            })
            .expect("serialize judgement request"),
        };
        let encoded = encode(&request).expect("encode judgement request");
        let decoded: Request = decode(&encoded).expect("decode judgement request");
        assert_eq!(decoded, request);
        let result = Response::Ok(ResponsePayload::Judgement {
            result_json:
                r#"{"answers":{"check":{"kind":"yes_no","probability":0.8}},"usage":null}"#.into(),
        });
        let encoded = encode(&result).expect("encode judgement response");
        let decoded: Response = decode(&encoded).expect("decode judgement response");
        assert_eq!(decoded, result);
    }

    #[test]
    fn session_model_list_with_pricing_round_trips() {
        let response = Response::Ok(ResponsePayload::SessionModelList {
            provider_plugin_id: Some("provider".to_string()),
            models: bcode_model::ModelList {
                models: vec![bcode_model::ModelInfo {
                    model_id: "model".to_string(),
                    display_name: "Model".to_string(),
                    is_default: true,
                    context_window: Some(128_000),
                    max_output_tokens: Some(16_000),
                    max_image_input_base64_bytes: None,
                    capabilities: BTreeSet::new(),
                    feature_support: bcode_model::ModelFeatureSupport::default(),
                    reasoning: None,
                    cache: bcode_model::ModelCacheInfo::default(),
                    metadata_source: None,
                    pricing: Some(bcode_model::ModelPricingInfo {
                        currency: "USD".to_string(),
                        unit: bcode_model::ModelPricingUnit::PerMillionTokens,
                        input: Some(bcode_model::ModelTokenPrice::from_micros(1_250_000)),
                        cached_input: Some(bcode_model::ModelTokenPrice::from_micros(125_000)),
                        cache_write_input: None,
                        output: Some(bcode_model::ModelTokenPrice::from_micros(10_000_000)),
                        context_threshold_tokens: None,
                        rules: Vec::new(),
                        revision: None,
                        source: bcode_model::ModelPricingSource::PatternMatch,
                    }),
                    api_surface: None,
                    visibility: bcode_model::ModelVisibility::Visible,
                }],
                catalog: bcode_model::ModelCatalogHints {
                    availability: BTreeMap::new(),
                    availability_scope: None,
                    policy: bcode_model::ModelCatalogPolicy::ExpandSupported {
                        provider_id: "openai".to_string(),
                        target: bcode_model::ModelCatalogSupportHint {
                            provider: "openai".to_string(),
                            auth_mode: "api_key".to_string(),
                            api_surface: "responses".to_string(),
                            integration: None,
                        },
                        authority: bcode_model::ModelListAuthority::Partial,
                    },
                },
            },
        });

        let encoded = encode_response(&response).expect("response should encode");
        let decoded = decode_response(&encoded).expect("response should decode");

        assert_eq!(decoded, response);
    }

    #[test]
    fn session_model_list_with_populated_feature_support_round_trips() {
        // Guards against the TypedStable codec regression where enum-keyed `BTreeMap`s in
        // `ModelFeatureSupport` failed to decode ("unexpected type tag: expected String, got
        // U32"). Every enum-keyed map is populated so all variant-identifier decode paths run.
        use bcode_model::{
            CapabilitySource, CapabilitySupport, MediaInputFeature, ModelFeatureSupport,
            ModelParameterKey, PromptCacheFeature, StructuredOutputMode, ToolChoiceMode,
        };

        let supported = || CapabilitySupport::supported(CapabilitySource::BundledCatalog);
        let unsupported = || CapabilitySupport::Unsupported {
            source: CapabilitySource::Probe,
            reason: "not available".to_string(),
        };

        let feature_support = ModelFeatureSupport {
            parameters: [
                (ModelParameterKey::Temperature, supported()),
                (ModelParameterKey::ReasoningEffort, unsupported()),
            ]
            .into_iter()
            .collect(),
            structured_output: [
                (StructuredOutputMode::JsonSchema, supported()),
                (StructuredOutputMode::StrictJsonSchema, unsupported()),
            ]
            .into_iter()
            .collect(),
            tool_choice: [
                (ToolChoiceMode::Auto, supported()),
                (ToolChoiceMode::Parallel, supported()),
            ]
            .into_iter()
            .collect(),
            tool_schema: std::collections::BTreeMap::new(),
            prompt_cache: std::iter::once((PromptCacheFeature::ConversationPrefix, supported()))
                .collect(),
            media_input: std::iter::once((MediaInputFeature::UserImage, supported())).collect(),
        };

        let response = Response::Ok(ResponsePayload::SessionModelList {
            provider_plugin_id: Some("bcode.bedrock".to_string()),
            models: bcode_model::ModelList {
                models: vec![bcode_model::ModelInfo {
                    model_id: "anthropic.claude".to_string(),
                    display_name: "Claude".to_string(),
                    is_default: true,
                    context_window: Some(200_000),
                    max_output_tokens: Some(64_000),
                    max_image_input_base64_bytes: None,
                    capabilities: BTreeSet::new(),
                    feature_support,
                    reasoning: None,
                    cache: bcode_model::ModelCacheInfo::default(),
                    metadata_source: None,
                    pricing: None,
                    api_surface: None,
                    visibility: bcode_model::ModelVisibility::Visible,
                }],
                catalog: bcode_model::ModelCatalogHints {
                    availability: BTreeMap::new(),
                    availability_scope: None,
                    policy: bcode_model::ModelCatalogPolicy::EnrichOnly {
                        provider_id: "bedrock".to_string(),
                        target: None,
                        authority: bcode_model::ModelListAuthority::Partial,
                    },
                },
            },
        });

        let encoded = encode_response(&response).expect("response should encode");
        let decoded = decode_response(&encoded).expect("response should decode");

        assert_eq!(decoded, response);
    }

    #[test]
    fn response_envelope_uses_current_protocol_version() {
        let envelope = response_envelope(7, &Response::Ok(ResponsePayload::MessageSent))
            .expect("response envelope should encode");

        assert_eq!(envelope.version, ProtocolVersion::current());
        assert_eq!(ProtocolVersion::current().0, CURRENT_PROTOCOL_VERSION);
    }

    #[test]
    fn reprice_snapshot_and_report_round_trip() {
        let session_id = SessionId::new();
        let range = bcode_session_models::SessionCostRange {
            from_timestamp_ms: 10,
            to_timestamp_ms: 20,
        };
        let request = Request::RepriceSession {
            session_id,
            range,
            catalog: Box::new(bcode_model_catalog_models::CatalogDocument::empty(
                "snapshot",
                "2026-09-07T00:00:00Z",
            )),
        };
        assert_eq!(
            decode_request(&encode_request(&request).unwrap()).unwrap(),
            request
        );
        let response = Response::Ok(ResponsePayload::SessionRepriced {
            report: Box::new(bcode_session_models::SessionRepriceReport {
                session_id,
                range,
                repriced_requests: 0,
                catalog_revision: "snapshot".into(),
                catalog_digest: "digest".into(),
                summary: bcode_session_models::SessionUsageSummary::default(),
            }),
        });
        assert_eq!(
            decode_response(&encode_response(&response).unwrap()).unwrap(),
            response
        );
    }

    #[test]
    fn projection_history_page_round_trips_through_positional_ipc() {
        use bcode_session_models::{
            ProjectionWindow, ProjectionWindowAnchor, ProjectionWindowDirection,
            ProjectionWindowLimits, ProjectionWindowRequest, ProjectionWindowTarget,
            SessionProjectionKind,
        };
        let session_id = SessionId::new();
        let request = Request::SessionProjectionHistoryPage {
            session_id,
            request: Box::new(ProjectionWindowRequest {
                projection: SessionProjectionKind::Transcript,
                anchor: ProjectionWindowAnchor::BeforeSequence(23),
                direction: ProjectionWindowDirection::Backward,
                target: ProjectionWindowTarget {
                    min_items: Some(3),
                    min_estimated_rows: Some(12),
                    width_columns: Some(80),
                    ..ProjectionWindowTarget::default()
                },
                limits: ProjectionWindowLimits {
                    max_items: 10,
                    max_events_scanned: 256,
                    max_bytes: 65_536,
                },
            }),
        };
        assert_eq!(
            decode_request(&encode_request(&request).expect("encode request"))
                .expect("decode request"),
            request
        );
        let response = Response::Ok(ResponsePayload::SessionProjectionHistoryPage {
            page: bcode_session_models::ProjectionHistoryPage {
                session_id,
                window: ProjectionWindow {
                    projection: SessionProjectionKind::Transcript,
                    transcript_items: Vec::new(),
                    source_range: Some(bcode_session_models::ProjectionSourceRange {
                        start_sequence: 8,
                        end_sequence: 16,
                    }),
                    has_older: true,
                    has_newer: true,
                    scanned_events: 9,
                },
                events: Vec::new(),
            },
        });
        assert_eq!(
            decode_response(&encode_response(&response).expect("encode response"))
                .expect("decode response"),
            response
        );
    }

    #[test]
    fn request_envelope_payload_decodes_with_request_codec() {
        let request = Request::Ping;
        let envelope = request_envelope(7, &request).expect("request envelope should encode");

        assert_eq!(envelope.kind, EnvelopeKind::Request);
        assert_eq!(
            decode_request(&envelope.payload).expect("request should decode"),
            request
        );
    }

    #[test]
    fn response_envelope_payload_decodes_with_response_codec() {
        let response = Response::Ok(ResponsePayload::Pong);
        let envelope = response_envelope(7, &response).expect("response envelope should encode");

        assert_eq!(envelope.kind, EnvelopeKind::Response);
        assert_eq!(
            decode_response(&envelope.payload).expect("response should decode"),
            response
        );
    }

    #[test]
    fn session_ownership_release_payloads_round_trip() {
        let session_id = SessionId::new();
        let request = Request::ReleaseSessionOwnership { session_id };
        let encoded = encode_request(&request).expect("request should encode");
        assert_eq!(
            decode_request(&encoded).expect("request should decode"),
            request
        );

        let response = Response::Ok(ResponsePayload::SessionOwnershipReleased {
            session_id,
            outcome: SessionOwnershipReleaseOutcome::Blocked {
                blockers: vec![
                    SessionOwnershipBlocker::AttachedClient,
                    SessionOwnershipBlocker::RuntimeWork,
                ],
            },
        });
        let encoded = encode_response(&response).expect("response should encode");
        assert_eq!(
            decode_response(&encoded).expect("response should decode"),
            response
        );
    }

    #[test]
    fn client_metrics_request_round_trips() {
        let request = Request::IngestClientMetrics {
            batch: bcode_metrics::ClientMetricBatch {
                observations: vec![bcode_metrics::ClientMetricObservation::CounterDelta {
                    name: "tui.frame.total".to_owned(),
                    value: 16,
                    labels: bcode_metrics::MetricLabels::new(),
                }],
            },
        };
        let encoded = encode_request(&request).expect("request should encode");

        assert_eq!(
            decode_request(&encoded).expect("request should decode"),
            request
        );
    }

    #[tokio::test]
    async fn unsupported_protocol_version_is_rejected() {
        let envelope = Envelope {
            version: ProtocolVersion(1),
            request_id: 1,
            kind: EnvelopeKind::Response,
            payload: encode(&Response::Ok(ResponsePayload::MessageSent))
                .expect("response should encode"),
        };
        let encoded = encode(&envelope).expect("envelope should encode");
        let mut frame = Vec::new();
        frame.extend_from_slice(
            &u32::try_from(encoded.len())
                .expect("encoded envelope should fit u32")
                .to_le_bytes(),
        );
        frame.extend_from_slice(&encoded);
        let mut cursor = std::io::Cursor::new(frame);

        let error = read_envelope_frame(&mut cursor)
            .await
            .expect_err("old protocol version should fail");

        assert!(matches!(
            error,
            CodecError::UnsupportedVersion {
                actual: 1,
                expected: CURRENT_PROTOCOL_VERSION
            }
        ));
    }

    #[tokio::test]
    async fn unsupported_future_protocol_version_is_rejected() {
        let actual = CURRENT_PROTOCOL_VERSION.saturating_add(1);
        let envelope = Envelope {
            version: ProtocolVersion(actual),
            request_id: 1,
            kind: EnvelopeKind::Response,
            payload: encode(&Response::Ok(ResponsePayload::MessageSent))
                .expect("response should encode"),
        };
        let encoded = encode(&envelope).expect("envelope should encode");
        let mut frame = Vec::new();
        frame.extend_from_slice(
            &u32::try_from(encoded.len())
                .expect("encoded envelope should fit u32")
                .to_le_bytes(),
        );
        frame.extend_from_slice(&encoded);
        let mut cursor = std::io::Cursor::new(frame);

        let error = read_envelope_frame(&mut cursor)
            .await
            .expect_err("future protocol version should fail");

        assert!(matches!(
            error,
            CodecError::UnsupportedVersion {
                actual: rejected,
                expected: CURRENT_PROTOCOL_VERSION
            } if rejected == actual
        ));
    }

    fn fixture_bytes(path: &str) -> Vec<u8> {
        let hex = std::fs::read_to_string(
            std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
                .join("../..")
                .join(path),
        )
        .expect("fixture should be readable");
        decode_hex(hex.trim()).expect("fixture should contain hex")
    }

    fn test_session_summary(name: &str) -> SessionSummary {
        let session_id: SessionId = "00000000-0000-0000-0000-000000000002"
            .parse()
            .expect("session id should parse");
        SessionSummary {
            parent_id: None,
            id: session_id,
            name: Some(name.to_owned()),
            explicit_name: Some(name.to_owned()),
            derived_title: None,
            automatic_title: None,
            automatic_title_source_sequence: None,
            title_source: bcode_session_models::SessionTitleSource::Explicit,
            client_count: 0,
            created_at_ms: 10,
            updated_at_ms: 20,
            working_directory: "/tmp/bcode-ipc-test".into(),
            import: None,
            execution: None,
            location: None,
        }
    }

    fn decode_hex(hex: &str) -> Result<Vec<u8>, String> {
        if !hex.len().is_multiple_of(2) {
            return Err("hex fixture has odd length".to_string());
        }
        (0..hex.len())
            .step_by(2)
            .map(|index| {
                u8::from_str_radix(&hex[index..index + 2], 16).map_err(|error| error.to_string())
            })
            .collect()
    }

    #[test]
    fn session_runtime_selection_round_trips_agent_identity() {
        let selection = SessionRuntimeSelection {
            agent_id: Some("build".to_owned()),
            provider_plugin_id: Some("provider".to_owned()),
            requested_model_id: Some("requested".to_owned()),
            effective_model_id: Some("effective".to_owned()),
            model_id: Some("requested".to_owned()),
            reasoning_effort: Some("high".to_owned()),
            reasoning_summary: Some("detailed".to_owned()),
        };

        let encoded = encode(&selection).expect("selection should encode");
        let decoded: SessionRuntimeSelection = decode(&encoded).expect("selection should decode");

        assert_eq!(decoded, selection);
    }

    #[test]
    fn runtime_context_with_semantic_auth_round_trips() {
        let request = Request::Hello {
            client_name: "test".to_string(),
            daemon_namespace: daemon_namespace(),
            artifact_id: Some(ArtifactId::current()),
            build_fingerprint: BUILD_FINGERPRINT.to_owned(),
            state_location_id: Some(state_location_id()),
            runtime_context: Some(ClientRuntimeContext {
                working_directory: Some(PathBuf::from("/tmp/client")),
                config_launch: None,
                effective_config_toml: None,
                selected_provider_plugin_id: Some("bcode.openai-compatible".to_string()),
                selected_model_id: Some("model".to_string()),
                requested_model_id: None,
                provider_context: bcode_model::ProviderRequestContext {
                    auth_profile: Some("openrouter".to_string()),
                    auth: Some(bcode_model::ProviderAuthContext {
                        profile: Some("openrouter".to_string()),
                        backend: Some("sshenv".to_string()),
                        scheme: Some("api_key".to_string()),
                        credentials: BTreeMap::from([(
                            "api_key".to_string(),
                            bcode_model::ProviderAuthCredential {
                                value: "secret".to_string(),
                                source: Some("OPENROUTER_API_KEY".to_string()),
                            },
                        )]),
                        attributes: BTreeMap::from([(
                            "base_url".to_string(),
                            "https://openrouter.ai/api/v1".to_string(),
                        )]),
                        storage: BTreeMap::from([(
                            "api_key".to_string(),
                            bcode_model::ProviderAuthStorageRef {
                                backend: "sshenv".to_string(),
                                profile: "openrouter".to_string(),
                                key: "OPENROUTER_API_KEY".to_string(),
                                vault: Some("/tmp/vault".to_string()),
                            },
                        )]),
                        diagnostics: Vec::new(),
                    }),
                    ..bcode_model::ProviderRequestContext::default()
                },
                interaction_adapters: vec![
                    bcode_plugin_sdk::interaction::PluginInteractionAdapterCapability {
                        producer_id: "example.plugin".to_owned(),
                        exchange_schema: "example.request".to_owned(),
                        min_schema_version: 2,
                        max_schema_version: 4,
                        platform_id: "tui".to_owned(),
                        priority: 50,
                        interaction_kind: "example.interaction".to_owned(),
                        tui_surface_kind: None,
                    },
                ],
                env_keys: BTreeMap::from([("OPENROUTER_API_KEY".to_string(), true)]),
                process_env: BTreeMap::from([("AWS_REGION".to_string(), "us-east-1".to_string())]),
            }),
        };

        let encoded = encode(&request).expect("request should encode");
        let decoded: Request = decode(&encoded).expect("request should decode");

        assert_eq!(decoded, request);
    }

    #[test]
    fn representative_non_history_responses_round_trip_through_typed_stable_ipc() {
        let responses = vec![
            Response::Ok(ResponsePayload::Hello {
                protocol_version: ProtocolVersion::current(),
                client_id: ClientId::new(),
                daemon: DaemonStatus {
                    namespace: daemon_namespace(),
                    protocol_version: u32::from(ProtocolVersion::current().0),
                    build_fingerprint: BUILD_FINGERPRINT.to_string(),
                    executable_digest: Some("digest".to_string()),
                    ..DaemonStatus::default()
                },
            }),
            Response::Ok(ResponsePayload::ServerStatus {
                status: ServerStatus {
                    connected_client_count: 1,
                    sessions: vec![test_session_summary("status")],
                    session_catalog_loaded: true,
                    session_catalog_status: SessionCatalogStatus::Loaded,
                    session_catalog_sources: Vec::new(),
                    session_catalog_revision: 7,
                    selected_provider_plugin_id: Some("provider".to_string()),
                    selected_model_id: Some("model".to_string()),
                    plugin_runtime: Vec::new(),
                    daemon: DaemonStatus {
                        namespace: daemon_namespace(),
                        protocol_version: u32::from(ProtocolVersion::current().0),
                        artifact_id: Some(ArtifactId::current()),
                        build_fingerprint: "test-build".to_string(),
                        executable_digest: Some("digest".to_string()),
                        storage_writer_epoch: Some(2),
                        state_location_id: Some("state-location".to_string()),
                        session_event_schema_version: Some(38),
                        pid: Some(123),
                        instance_id: "instance".to_string(),
                        started_at_unix_ms: 456,
                    },
                    metrics: MetricsSnapshot::default(),
                    metrics_report: Box::default(),
                    workflow_unavailable_reason: None,
                    active_runtime_work: Vec::new(),
                    idle_shutdown_blocker: None,
                },
            }),
            Response::Ok(ResponsePayload::SessionList {
                sessions: vec![test_session_summary("listed")],
                catalog_status: SessionCatalogStatus::Loaded,
                catalog_sources: Vec::new(),
                catalog_revision: 7,
            }),
            Response::Ok(ResponsePayload::PluginServiceResult {
                response: PluginServiceResponse {
                    payload: b"payload".to_vec(),
                    error: None,
                },
            }),
            Response::Ok(ResponsePayload::WorktreeList(WorktreeListResponse {
                repo_root: "/tmp/repo".into(),
                current_worktree: "/tmp/repo".into(),
                worktrees: Vec::new(),
            })),
            Response::Ok(ResponsePayload::SessionCatalogRefreshed {
                sessions: vec![test_session_summary("refreshed")],
                catalog_status: SessionCatalogStatus::Loaded,
                catalog_sources: Vec::new(),
                catalog_revision: 8,
            }),
            Response::Ok(ResponsePayload::RuntimeWorkCancellationRequested { cancelled: true }),
            Response::Err(ErrorResponse::new("test_error", "something failed")),
        ];

        for response in responses {
            let encoded = encode_response(&response).expect("response should encode");
            let decoded = decode_response(&encoded).expect("response should decode");

            assert_eq!(decoded, response);
        }
    }

    fn chunked_inspection_definition() -> bcode_workflow::WorkflowDefinitionSnapshot {
        use bcode_workflow::{
            ValueSchema, WorkflowDefinition, WorkflowDefinitionRepresentation,
            WorkflowDefinitionSnapshot,
        };
        let schema = ValueSchema {
            type_name: "inspection-test".into(),
            schema: serde_json::json!({"type": "object"}),
        };
        let definition = WorkflowDefinition {
            schema_version: 1,
            name: "inspection-test".into(),
            input: schema.clone(),
            output: schema,
            nodes: BTreeMap::new(),
            entries: Vec::new(),
            exits: Vec::new(),
            edges: Vec::new(),
        };
        WorkflowDefinitionSnapshot {
            definition_id: "inspection-test".into(),
            version: 1,
            checksum_sha256: "a".repeat(64),
            definition: WorkflowDefinitionRepresentation::parse(
                serde_json::to_string(&definition).expect("encode definition"),
            )
            .expect("parse definition"),
        }
    }

    fn chunked_inspection_run(
        status: bcode_workflow::RunStatus,
    ) -> bcode_workflow::WorkflowRunSummary {
        use bcode_workflow::{RunStatus, WorkflowRunSummary};
        let completed = status == RunStatus::Completed;
        WorkflowRunSummary {
            run_id: "inspection-run".into(),
            definition_id: "inspection-test".into(),
            definition_version: 1,
            workspace_snapshot: "snapshot".into(),
            parent_session_id: None,
            parent_session_generation: None,
            binding: None,
            authored_provenance: None,
            terminal_output_id: completed.then(|| "output".into()),
            terminal_output_checksum_sha256: completed.then(|| "a".repeat(64)),
            authorization_profile: bcode_workflow::WorkflowAuthorizationProfileIdentity {
                version: 1,
                provider_id: "test-policy".into(),
                profile_id: "build".into(),
                policy_digest_sha256: "a".repeat(64),
            },
            authorization_ceiling: bcode_workflow::WorkflowToolCapability::Mutating,
            status,
            cancellation_requested_at_ms: (status == RunStatus::Cancelled).then_some(2),
            created_at_ms: 1,
            updated_at_ms: 2,
        }
    }

    fn chunked_workflow_inspection(status: bcode_workflow::RunStatus) -> Response {
        use bcode_workflow::{
            RunStatus, WaitingActivation, WorkflowActivationInputSummary,
            WorkflowActivationSummary, WorkflowOutputSummary, WorkflowRunInspection,
            WorkflowTerminalOutputInspection, WorkflowWaitKind,
        };
        // Physical framing chunks only after LZ4 compression. A repeated string
        // tests compression, not chunking; use bounded deterministic varied data.
        let mut varied = String::with_capacity(4 * MAX_FRAME_PAYLOAD_SIZE);
        for index in 0..(4 * MAX_FRAME_PAYLOAD_SIZE / 64) {
            for byte in Sha256::digest(index.to_le_bytes()) {
                const HEX: &[u8; 16] = b"0123456789abcdef";
                varied.push(char::from(HEX[usize::from(byte >> 4)]));
                varied.push(char::from(HEX[usize::from(byte & 15)]));
            }
        }
        let nested = serde_json::json!({"result": {"rows": [varied]}});
        let completed = status == RunStatus::Completed;
        let inspection = WorkflowRunInspection {
            canonical_eval_coverage: None,
            admission_observations: Vec::new(),
            admission_observations_complete: Some(true),
            unavailable_collections: Vec::new(),
            execution_allowance: None,
            continuation: None,
            replacement_readiness: None,
            run: chunked_inspection_run(status),
            graph: None,
            definition: chunked_inspection_definition(),
            terminal_output: completed.then(|| WorkflowTerminalOutputInspection {
                version: 1,
                output_id: "output".into(),
                node_id: "result".into(),
                activation_id: "activation".into(),
                schema_id: "inspection-test".into(),
                schema_version: 1,
                checksum_sha256: "a".repeat(64),
                value: nested.clone(),
                artifact_reference: None,
                created_at_ms: 2,
            }),
            activations: ["running", "waiting", "unresolved"]
                .into_iter()
                .map(|state| WorkflowActivationSummary {
                    run_id: "inspection-run".into(),
                    node_id: state.into(),
                    activation_id: state.into(),
                    dependency_generation: 0,
                    status: state.into(),
                    has_output: false,
                    input_summary: WorkflowActivationInputSummary::Inline {
                        value: nested.clone(),
                    },
                    created_at_ms: 1,
                })
                .collect(),
            waits: vec![WaitingActivation {
                run_id: "inspection-run".into(),
                node_id: "waiting".into(),
                activation_id: "waiting".into(),
                kind: WorkflowWaitKind::Input,
                input: Some(serde_json::json!({"prompt": "continue?"})),
                requested_at_ms: 1,
            }],
            mutation_approvals: Vec::new(),
            attempts: Vec::new(),
            events: Vec::new(),
            decisions: Vec::new(),
            grants: Vec::new(),
            resource_leases: Vec::new(),
            outputs: vec![WorkflowOutputSummary {
                output_id: "output".into(),
                run_id: "inspection-run".into(),
                node_id: "result".into(),
                activation_id: "activation".into(),
                schema_id: "inspection-test".into(),
                schema_version: 1,
                artifact_reference: None,
                checksum_sha256: "a".repeat(64),
                created_at_ms: 2,
            }],
            child_run_links: Vec::new(),
            descendant_runs: Vec::new(),
            repeat_outcomes: Vec::new(),
            child_sessions: Vec::new(),
            coordinator: None,
        };
        Response::Ok(ResponsePayload::WorkflowRunInspection {
            inspection: Box::new(inspection),
        })
    }

    #[tokio::test]
    async fn chunked_workflow_inspection_round_trips_complete_typed_states() {
        use bcode_workflow::RunStatus;
        for status in [
            RunStatus::Running,
            RunStatus::Completed,
            RunStatus::Failed,
            RunStatus::Cancelled,
        ] {
            let response = chunked_workflow_inspection(status);
            let envelope = response_envelope(42, &response).expect("encode inspection");
            assert!(envelope_frame_count(&envelope).expect("count frames") > 1);
            let received = round_trip_envelope(envelope.clone()).await;
            assert_eq!(received, envelope);
            assert_eq!(
                decode_response(&received.payload).expect("decode inspection"),
                response
            );
        }
    }

    #[tokio::test]
    async fn chunked_workflow_inspection_rejects_malformed_sequence_without_partial_response() {
        let response = chunked_workflow_inspection(bcode_workflow::RunStatus::Running);
        let envelope = response_envelope(42, &response).expect("encode inspection");
        let frames = encode_envelope_frames(&envelope).expect("encode frames");
        assert!(frames.len() > 1);
        // Duplicate the first chunk in place of its continuation. The receiver must
        // reject the whole logical response rather than expose the assembled prefix.
        let mut bytes = frames[0].clone();
        bytes.extend_from_slice(&frames[0]);
        let mut reader = bytes.as_slice();
        assert!(matches!(
            recv_envelope(&mut reader).await,
            Err(CodecError::InvalidChunk(_))
        ));

        let mut reader = frames[0].as_slice();
        assert!(matches!(
            recv_envelope(&mut reader).await,
            Err(CodecError::Io(_))
        ));
    }

    #[tokio::test]
    async fn oversized_response_envelope_round_trips_across_chunked_frames() {
        let payload = vec![b'x'; MAX_FRAME_PAYLOAD_SIZE + 100_000];
        let response = Response::Ok(ResponsePayload::PluginServiceResult {
            response: PluginServiceResponse {
                payload,
                error: None,
            },
        });
        let envelope = response_envelope(42, &response).expect("response should encode");
        assert!(encode(&envelope).expect("envelope should encode").len() > MAX_FRAME_PAYLOAD_SIZE);

        let received = round_trip_envelope(envelope.clone()).await;

        assert_eq!(received, envelope);
        let decoded = decode_response(&received.payload).expect("response should decode");
        assert_eq!(decoded, response);
    }

    #[tokio::test]
    async fn oversized_event_envelope_round_trips_across_chunked_frames() {
        let session_id = SessionId::new();
        let event = Event::Session(SessionEvent {
            schema_version: CURRENT_SESSION_EVENT_SCHEMA_VERSION,
            sequence: 7,
            timestamp_ms: 1,
            session_id,
            provenance: None,
            kind: SessionEventKind::ToolInvocationResultRecorded {
                record: bcode_session_models::ToolInvocationResultRecord {
                    invocation_id: "call-1".to_string(),
                    model_output: "z".repeat(MAX_FRAME_PAYLOAD_SIZE + 100_000),
                    is_error: false,
                    presentation: None,
                    result: None,
                    content: Vec::new(),
                },
            },
        });

        let envelope = event_envelope(&event).expect("event should encode");
        assert!(encode(&envelope).expect("envelope should encode").len() > MAX_FRAME_PAYLOAD_SIZE);

        let received = round_trip_envelope(envelope.clone()).await;

        assert_eq!(received, envelope);
        let decoded = decode_event(&received.payload).expect("event should decode");
        assert_eq!(decoded, event);
    }

    #[allow(clippy::too_many_lines)]
    fn sample_session_event_kinds(_session_id: SessionId) -> Vec<SessionEventKind> {
        vec![
            SessionEventKind::SessionCreated {
                name: Some("session".to_string()),
                working_directory: "/tmp/bcode".into(),
            },
            SessionEventKind::ClientAttached {
                client_id: ClientId::new(),
            },
            SessionEventKind::ClientDetached {
                client_id: ClientId::new(),
            },
            SessionEventKind::UserMessage {
                client_id: ClientId::new(),
                text: "hello".to_string(),
                admission: bcode_session_models::TurnAdmissionMetadata::default(),
            },
            SessionEventKind::AssistantDelta {
                text: "delta".to_string(),
            },
            SessionEventKind::AssistantMessage {
                text: "message".to_string(),
            },
            SessionEventKind::ToolCallRequested {
                tool_call_id: "call-1".to_string(),
                producer_plugin_id: None,
                tool_name: "shell.run".to_string(),
                arguments_json: "{}".to_string(),
                working_directory: None,
            },
            SessionEventKind::ToolInvocationResultRecorded {
                record: bcode_session_models::ToolInvocationResultRecord {
                    invocation_id: "call-1".to_string(),
                    model_output: "done".to_string(),
                    is_error: false,
                    presentation: None,
                    result: Some(ToolInvocationResult::Text {
                        text: "done".to_string(),
                    }),
                    content: Vec::new(),
                },
            },
            SessionEventKind::PermissionRequested {
                permission_id: "perm-1".to_string(),
                tool_call_id: "call-1".to_string(),
                producer_plugin_id: None,
                tool_name: "shell.run".to_string(),
                arguments_json: "{}".to_string(),
                batch: None,
                policy_source: None,
                policy_reason: None,
            },
            SessionEventKind::PermissionResolved {
                permission_id: "perm-1".to_string(),
                approved: true,
            },
            SessionEventKind::ModelChanged {
                provider: "provider".to_string(),
                model: "model".to_string(),
                selection_source: bcode_session_models::ModelSelectionSource::UserExplicit,
            },
            SessionEventKind::SystemMessage {
                text: "system".to_string(),
            },
            SessionEventKind::AgentChanged {
                agent_id: "agent".to_string(),
            },
            SessionEventKind::ModelTurnStarted {
                turn_id: "turn-1".to_string(),
            },
            SessionEventKind::ModelTurnFinished {
                turn_id: "turn-1".to_string(),
                outcome: ModelTurnOutcome::Completed,
                message: Some("ok".to_string()),
            },
            SessionEventKind::ModelUsage {
                turn_id: "turn-1".to_string(),
                usage: bcode_session_models::SessionTokenUsage {
                    input_tokens: Some(1),
                    output_tokens: Some(2),
                    total_tokens: Some(3),
                    cached_input_tokens: None,
                    cache_write_input_tokens: None,
                    reasoning_tokens: None,
                    ..bcode_session_models::SessionTokenUsage::default()
                },
            },
            SessionEventKind::ContextCompacted {
                summary: "summary".to_string(),
                compacted_through_sequence: 10,
            },
            SessionEventKind::SessionRenamed {
                name: Some("renamed".to_string()),
            },
            SessionEventKind::TraceEvent {
                trace: Box::new(SessionTraceEvent {
                    timestamp_ms: 1,
                    turn_id: Some("turn-1".to_string()),
                    phase: bcode_session_models::SessionTracePhase::ModelRequestBuilt,
                    payload: bcode_session_models::SessionTracePayload::ModelRequestBuilt {
                        provider: "provider".to_string(),
                        model: "model".to_string(),
                        agent_id: "agent".to_string(),
                        message_count: 1,
                        tool_count: 0,
                        system_prompt_chars: 10,
                        prompt_cache_mode: "off".to_string(),
                        conversation_reuse_mode: "none".to_string(),
                        uses_previous_provider_response: false,
                        metadata: BTreeMap::new(),
                        request: None,
                    },
                }),
            },
            SessionEventKind::SkillInvoked {
                skill_id: SkillId::new("skill"),
                arguments: "{}".to_string(),
                source: None,
                invoked_at_ms: 1,
            },
            SessionEventKind::SkillSuggested {
                skill_id: SkillId::new("skill"),
                reason: Some("reason".to_string()),
                suggested_at_ms: 1,
            },
            SessionEventKind::SkillActivated {
                skill_id: SkillId::new("skill"),
                source: None,
                mode: SkillActivationMode::Explicit,
                activated_at_ms: 1,
            },
            SessionEventKind::SkillDeactivated {
                skill_id: SkillId::new("skill"),
                deactivated_at_ms: 1,
            },
            SessionEventKind::SkillContextLoaded {
                skill_id: SkillId::new("skill"),
                bytes_loaded: 42,
                truncated: false,
                loaded_at_ms: 1,
                source: None,
                preview: None,
            },
            SessionEventKind::SkillInvocationFailed {
                skill_id: SkillId::new("skill"),
                error: "nope".to_string(),
                failed_at_ms: 1,
            },
            SessionEventKind::AssistantReasoningDelta {
                text: "thinking".to_string(),
            },
            SessionEventKind::AssistantReasoningMessage {
                text: "thought".to_string(),
            },
            SessionEventKind::RuntimeWorkStarted {
                work_id: WorkId::new("work-1"),
                kind: RuntimeWorkKind::Tool,
                label: "work".to_string(),
                tool_call_id: Some("call-1".to_string()),
                plugin_id: Some("plugin".to_string()),
                service_interface: Some("service".to_string()),
                operation: Some("op".to_string()),
                parent_work_id: None,
                started_at_ms: Some(1),
                cancellable: true,
            },
            SessionEventKind::RuntimeWorkCancelRequested {
                work_id: WorkId::new("work-1"),
                requested_at_ms: Some(2),
                client_id: Some(ClientId::new()),
            },
            SessionEventKind::RuntimeWorkFinished {
                work_id: WorkId::new("work-1"),
                status: RuntimeWorkStatus::Completed,
                finished_at_ms: Some(3),
                message: Some("done".to_string()),
            },
            SessionEventKind::RuntimeWorkProgress {
                work_id: WorkId::new("work-1"),
                message: "progress".to_string(),
                progress_at_ms: Some(2),
                completed_units: Some(1),
                total_units: Some(2),
            },
            SessionEventKind::ModelTurnCancelRequested {
                turn_id: "turn-1".to_string(),
                requested_at_ms: Some(2),
                client_id: Some(ClientId::new()),
            },
            SessionEventKind::ToolContribution {
                event: bcode_session_models::ToolContributionEvent {
                    invocation_id: "call-1".to_string(),
                    contribution_id: "surface".to_string(),
                    sequence: 4,
                    producer_id: "future.producer".to_string(),
                    schema: "future.unknown/schema".to_string(),
                    schema_version: 77,
                    operation: bcode_session_models::ToolContributionOperation::Append,
                    persistence: bcode_session_models::ToolContributionPersistence::Durable,
                    artifact: None,
                    payload: serde_json::json!({"opaque": [1, {"future": true}]}),
                },
            },
            SessionEventKind::ToolExchangeRequested {
                request: bcode_session_models::ToolExchangeRequest {
                    invocation_id: "call-1".to_string(),
                    exchange_id: "question".to_string(),
                    producer_id: "future.producer".to_string(),
                    schema: "future.question/schema".to_string(),
                    schema_version: 77,
                    payload: serde_json::json!({"opaque": "request"}),
                    response_policy: bcode_session_models::ToolExchangeResponsePolicy::Required,
                },
            },
            SessionEventKind::ToolExchangeResolved {
                event: bcode_session_models::ToolExchangeResolutionEvent {
                    invocation_id: "call-1".to_string(),
                    exchange_id: "question".to_string(),
                    resolution: bcode_session_models::ToolExchangeResolution::Responded {
                        payload: serde_json::json!({"opaque": "response"}),
                    },
                },
            },
            SessionEventKind::WorkingDirectoryChanged {
                old_working_directory: "/tmp/old".into(),
                new_working_directory: "/tmp/new".into(),
            },
            SessionEventKind::SessionImported {
                source_id: "source".to_string(),
                source_display_name: "Source".to_string(),
                external_session_id: "external".to_string(),
                imported_at_ms: 1,
            },
        ]
    }

    #[test]
    fn direct_domain_session_events_round_trip_through_typed_stable() {
        let session_id = SessionId::new();
        for (index, kind) in sample_session_event_kinds(session_id)
            .into_iter()
            .enumerate()
        {
            let event = Event::Session(SessionEvent {
                schema_version: CURRENT_SESSION_EVENT_SCHEMA_VERSION,
                sequence: u64::try_from(index).expect("index should fit u64"),
                timestamp_ms: 1,
                session_id,
                provenance: None,
                kind,
            });

            let encoded = encode_typed_stable(&event).expect("event should encode directly");
            let decoded: Event = decode_typed_stable(&encoded).unwrap_or_else(|error| {
                panic!("event {index} {event:?} should decode directly: {error}")
            });

            assert_eq!(decoded, event);
        }
    }

    #[tokio::test]
    async fn all_session_event_kinds_round_trip_across_ipc_frames() {
        let session_id = SessionId::new();
        for (index, kind) in sample_session_event_kinds(session_id)
            .into_iter()
            .enumerate()
        {
            let event = Event::Session(SessionEvent {
                schema_version: CURRENT_SESSION_EVENT_SCHEMA_VERSION,
                sequence: u64::try_from(index).expect("index should fit u64"),
                timestamp_ms: 1,
                session_id,
                provenance: None,
                kind,
            });
            let envelope = event_envelope(&event).expect("event should encode");

            let received = round_trip_envelope(envelope).await;

            let decoded = decode_event(&received.payload).expect("event should decode");
            assert_eq!(decoded, event);
        }
    }

    fn semantic_tool_result_event(
        session_id: SessionId,
        semantic_result: ToolInvocationResult,
    ) -> SessionEvent {
        SessionEvent {
            schema_version: CURRENT_SESSION_EVENT_SCHEMA_VERSION,
            sequence: 77,
            timestamp_ms: 1,
            session_id,
            provenance: None,
            kind: SessionEventKind::ToolInvocationResultRecorded {
                record: bcode_session_models::ToolInvocationResultRecord {
                    invocation_id: "call-1".to_string(),
                    model_output: "tool result".to_string(),
                    is_error: false,
                    presentation: None,
                    result: Some(semantic_result),
                    content: Vec::new(),
                },
            },
        }
    }

    fn semantic_tool_results() -> Vec<ToolInvocationResult> {
        vec![
            ToolInvocationResult::Artifact {
                artifact: Box::new(bcode_session_models::ToolArtifact {
                    artifact_id: "artifact-1".to_string(),
                    producer_plugin_id: "bcode.test".to_string(),
                    schema: "bcode.test.artifact".to_string(),
                    schema_version: 1,
                    tool_call_id: Some("call-1".to_string()),
                    title: Some("Test artifact".to_string()),
                    metadata: serde_json::json!({"ok": true}),
                    refs: vec![bcode_session_models::ToolArtifactRef {
                        key: "data".to_string(),
                        content_type: Some("application/json".to_string()),
                        storage_uri: None,
                        byte_len: Some(11),
                        metadata: None,
                    }],
                }),
            },
            ToolInvocationResult::Text {
                text: "plain text".to_string(),
            },
            ToolInvocationResult::Json {
                value: r#"{"ok":true}"#.to_string(),
            },
        ]
    }

    #[tokio::test]
    async fn session_open_preparation_requests_and_response_round_trip() {
        let session_id = SessionId::new();
        let operation_id = bcode_session_models::SessionOpenOperationId::new();
        let snapshot = bcode_session_models::SessionOpenOperationSnapshot {
            operation_id,
            revision: 7,
            session_id,
            source_writer_epoch: Some(3),
            target_writer_epoch: 4,
            progress: bcode_session_models::SessionMigrationProgress {
                stage: bcode_session_models::SessionMigrationStage::RebuildingProjections,
                completed_units: Some(100),
                total_units: Some(500),
                unit: Some(bcode_session_models::SessionMigrationProgressUnit::Events),
                message: "Rebuilding session indexes".to_owned(),
            },
            outcome: None,
            backup_path: Some(PathBuf::from("/tmp/backup")),
        };
        for request in [
            Request::PrepareSessionOpen { session_id },
            Request::WaitSessionOpenProgress {
                session_id,
                operation_id,
                after_revision: 6,
                timeout_ms: 1_000,
            },
        ] {
            let envelope = request_envelope(41, &request).expect("request encode");
            let received = round_trip_envelope(envelope).await;
            assert_eq!(
                decode_request(&received.payload).expect("request decode"),
                request
            );
        }
        let response = Response::Ok(ResponsePayload::SessionOpenPrepared { snapshot });
        let envelope = response_envelope(42, &response).expect("response encode");
        let received = round_trip_envelope(envelope).await;
        assert_eq!(
            decode_response(&received.payload).expect("response decode"),
            response
        );
    }

    #[tokio::test]
    async fn every_session_open_stage_unit_and_terminal_outcome_round_trips() {
        use bcode_session_models::{
            SessionMigrationProgress, SessionMigrationProgressUnit, SessionMigrationStage,
            SessionOpenFailureKind, SessionOpenOperationId, SessionOpenOperationSnapshot,
            SessionOpenTerminalOutcome,
        };

        let session_id = SessionId::new();
        let operation_id = SessionOpenOperationId::new();
        let stages = [
            SessionMigrationStage::WaitingForOwnership,
            SessionMigrationStage::InspectingStorage,
            SessionMigrationStage::PlanningBackup,
            SessionMigrationStage::CopyingBackup,
            SessionMigrationStage::VerifyingBackup,
            SessionMigrationStage::PreparingSchema,
            SessionMigrationStage::ReadingCanonicalHistory,
            SessionMigrationStage::RebuildingProjections,
            SessionMigrationStage::ValidatingProjections,
            SessionMigrationStage::Committing,
            SessionMigrationStage::ValidatingWriteReadiness,
            SessionMigrationStage::OpeningSession,
            SessionMigrationStage::Complete,
            SessionMigrationStage::Failed,
        ];
        let units = [
            SessionMigrationProgressUnit::Files,
            SessionMigrationProgressUnit::Bytes,
            SessionMigrationProgressUnit::Events,
        ];
        let failure_kinds = [
            SessionOpenFailureKind::OwnedByOtherDaemon,
            SessionOpenFailureKind::WriterIncompatible,
            SessionOpenFailureKind::ProjectionStale,
            SessionOpenFailureKind::RepairRequired,
            SessionOpenFailureKind::BackupFailed,
            SessionOpenFailureKind::MigrationFailed,
            SessionOpenFailureKind::NotFound,
        ];
        let mut outcomes = vec![
            SessionOpenTerminalOutcome::Ready,
            SessionOpenTerminalOutcome::DegradedReadOnly { issue_count: 2 },
            SessionOpenTerminalOutcome::WriterIncompatible {
                actual: Some(5),
                expected: 4,
            },
            SessionOpenTerminalOutcome::WriterIncompatible {
                actual: None,
                expected: 4,
            },
            SessionOpenTerminalOutcome::RepairRequired {
                reason: "damaged tail".to_owned(),
            },
        ];
        outcomes.extend(
            failure_kinds
                .into_iter()
                .map(|kind| SessionOpenTerminalOutcome::Failed {
                    kind,
                    message: "classified failure".to_owned(),
                    backup_path: Some(PathBuf::from("/tmp/retained-backup")),
                }),
        );

        for (index, stage) in stages.into_iter().enumerate() {
            let unit = units[index % units.len()];
            let snapshot = SessionOpenOperationSnapshot {
                operation_id,
                revision: u64::try_from(index).expect("stage index"),
                session_id,
                source_writer_epoch: Some(3),
                target_writer_epoch: 4,
                progress: SessionMigrationProgress {
                    stage,
                    completed_units: Some(2),
                    total_units: Some(3),
                    unit: Some(unit),
                    message: format!("stage {index}"),
                },
                outcome: None,
                backup_path: Some(PathBuf::from("/tmp/backup")),
            };
            assert_session_open_snapshot_frame_round_trip(snapshot).await;
        }
        for (index, outcome) in outcomes.into_iter().enumerate() {
            let snapshot = SessionOpenOperationSnapshot {
                operation_id,
                revision: u64::try_from(index).expect("outcome index"),
                session_id,
                source_writer_epoch: Some(3),
                target_writer_epoch: 4,
                progress: SessionMigrationProgress {
                    stage: SessionMigrationStage::Complete,
                    completed_units: None,
                    total_units: None,
                    unit: None,
                    message: "terminal".to_owned(),
                },
                outcome: Some(outcome),
                backup_path: Some(PathBuf::from("/tmp/backup")),
            };
            assert_session_open_snapshot_frame_round_trip(snapshot).await;
        }
    }

    async fn assert_session_open_snapshot_frame_round_trip(
        snapshot: bcode_session_models::SessionOpenOperationSnapshot,
    ) {
        let response = Response::Ok(ResponsePayload::SessionOpenPrepared { snapshot });
        let envelope = response_envelope(43, &response).expect("response encode");
        let received = round_trip_envelope(envelope).await;
        assert_eq!(
            decode_response(&received.payload).expect("response decode"),
            response
        );
    }

    #[tokio::test]
    async fn semantic_tool_result_events_round_trip_across_ipc_frames() {
        let session_id = SessionId::new();
        for semantic_result in semantic_tool_results() {
            let event = Event::Session(semantic_tool_result_event(session_id, semantic_result));
            let envelope = event_envelope(&event).expect("event should encode");

            let received = round_trip_envelope(envelope).await;

            let decoded = decode_event(&received.payload).expect("event should decode");
            assert_eq!(decoded, event);
        }
    }

    #[tokio::test]
    async fn semantic_tool_result_response_histories_round_trip_across_ipc_frames() {
        let session_id = SessionId::new();
        let session = test_session_summary("semantic history");

        for semantic_result in semantic_tool_results() {
            let event = semantic_tool_result_event(session_id, semantic_result);
            for response in [
                Response::Ok(ResponsePayload::Attached {
                    session_id,
                    session: session.clone(),
                    history: vec![event.clone()],
                    input_history: Vec::new(),
            usage_summary: None,
                    import_warnings: Vec::new(),
                    draft: Some("draft text".to_owned()),
                    runtime_selection: SessionRuntimeSelection::default(),
                    projection_window: None,
                }),
                Response::Ok(ResponsePayload::SessionHistory {
                    session_id,
                    history: vec![event.clone()],
                }),
                Response::Ok(ResponsePayload::SessionHistoryPage {
                    page: bcode_session_models::SessionHistoryPage {
                        session_id,
                        events: vec![event.clone()],
                        compatibility_issues: vec![
                            bcode_session_models::SessionEventCompatibilityIssue {
                                sequence: event.sequence,
                                event_kind: "future_event_kind".to_owned(),
                                schema_version: event.schema_version,
                                compatibility: bcode_session_models::SessionEventCompatibilityKind::UnknownEventKind,
                                remediation: "upgrade Bcode".to_owned(),
                            },
                        ],
                        next_cursor: None,
                        has_more: false,
                    },
                }),
                Response::Ok(ResponsePayload::RuntimeWorkHistory {
                    events: vec![event.clone()],
                }),
            ] {
                let envelope = response_envelope(42, &response).expect("response should encode");

                let received = round_trip_envelope(envelope).await;

                let decoded = decode_response(&received.payload).expect("response should decode");
                assert_eq!(decoded, response);
            }
        }
    }

    #[tokio::test]
    async fn graph_inspection_preserves_selected_edge_revision_across_frames() {
        let graph = WorkflowRunGraphInspection {
            revision: 3,
            next_edge_id: Some(8),
            nodes: Vec::new(),
            edges: vec![WorkflowRunGraphEdgeInspection {
                revision: 2,
                edge_id: 7,
                edge: bcode_workflow::EdgeDefinition {
                    from: "source".to_string(),
                    to: "target".to_string(),
                    kind: bcode_workflow::EdgeKind::Direct,
                    transform: None,
                },
            }],
            nodes_complete: false,
            edges_complete: true,
        };
        let response = Response::Ok(ResponsePayload::WorkflowRunGraphInspection {
            graph: graph.clone(),
        });
        let envelope = response_envelope(1, &response).expect("encode graph");
        let received = round_trip_envelope(envelope).await;
        let Response::Ok(ResponsePayload::WorkflowRunGraphInspection { graph: decoded }) =
            decode_response(&received.payload).expect("decode graph")
        else {
            panic!("unexpected response");
        };
        assert_eq!(decoded, graph);
        assert_ne!(decoded.edges[0].revision, decoded.revision);
        let mut missing_revision = serde_json::to_value(&graph.edges[0]).expect("edge JSON");
        missing_revision
            .as_object_mut()
            .expect("object")
            .remove("revision");
        assert!(
            serde_json::from_value::<WorkflowRunGraphEdgeInspection>(missing_revision).is_err()
        );
    }

    #[test]
    fn ralph_runner_requests_round_trip() {
        let requests = [
            Request::RunRalphLoop(RalphRunRequest {
                repo_root: PathBuf::from("/repo"),
                loop_state_dir: Some(PathBuf::from("/repo/.bcode/ralph/state")),
                max_iterations: Some(5),
                no_progress_limit: Some(2),
                require_approval: true,
            }),
            Request::ApproveRalphRun(RalphApproveRequest {
                repo_root: PathBuf::from("/repo"),
                loop_state_dir: None,
                run_id: Some("run-1".to_owned()),
            }),
            Request::CancelRalphLoop(RalphCancelRequest {
                repo_root: PathBuf::from("/repo"),
                run_id: Some("run-1".to_owned()),
                loop_state_dir: None,
            }),
            Request::RalphRunStatus(RalphRunStatusRequest {
                repo_root: PathBuf::from("/repo"),
                loop_state_dir: None,
            }),
            Request::ListRalphRuns(Box::new(RalphListRunsRequest {
                repo_root: PathBuf::from("/repo"),
                loop_state_dir: None,
            })),
            Request::ListRalphIterations(Box::new(RalphListIterationsRequest {
                repo_root: PathBuf::from("/repo"),
                loop_state_dir: None,
                run_id: Some("run-1".to_owned()),
            })),
            Request::ResumeRalphRun(RalphResumeRequest {
                repo_root: PathBuf::from("/repo"),
                loop_state_dir: None,
                interrupted_run_id: Some("run-1".to_owned()),
            }),
        ];
        for request in requests {
            let encoded = encode(&request).expect("request should encode");
            let decoded: Request = decode(&encoded).expect("request should decode");
            assert_eq!(decoded, request);
        }
    }

    async fn round_trip_envelope(envelope: Envelope) -> Envelope {
        let (mut sender, mut receiver) = tokio::io::duplex(64 * 1024);
        let send = send_envelope(&mut sender, &envelope);
        let receive = recv_envelope(&mut receiver);
        let (send_result, receive_result) = tokio::join!(send, receive);
        send_result.expect("send should succeed");
        receive_result.expect("receive should succeed")
    }
}