alma 0.1.1

//! Plugin runtime ECS adapter.

#[cfg(feature = "plugin-runtime")]
use crate::plugin::{PluginWorkspaceObserveTaskEnqueueReport, WasmtimePluginScheduledUpdate};
use crate::{
    config::AppConfig,
    ecs::{
        events::{
            edit::{
                PluginBufferEditProposalApplyRequested, PluginBufferEditProposalDecision,
                PluginBufferEditProposalDecisionRejected, PluginBufferEditProposalReceipt,
                PluginBufferEditProposalRejectRequested, PluginProposalDecisionRejectionReason,
                RevisionGuardedBufferEditRequested,
            },
            plugin::PluginOperationalEventReported,
            status::StatusMessageRequested,
        },
        plugin_proposal::{PluginProposalLane, PluginProposalLanePolicy, PluginProposalReceiptLog},
        schedules::EditorSet,
    },
    plugin::{
        PluginEffectDiscardReport, PluginHostOwnerBatches, PluginHostOwnerDiscardReport,
        PluginIdentity, PluginOperationalEvent, PluginOperationalQueue, PluginProposalReviewMode,
        PluginSchedulerPolicy, PluginWorkspaceIoWorkerQueue, PluginWorkspaceIoWorkerQueueError,
        SealedPluginEffectBatch,
    },
};
use bevy::prelude::{App, IntoScheduleConfigs, MessageWriter, Plugin, ResMut, Resource, Update};
use std::{
    collections::{VecDeque, vec_deque},
    fmt::{Debug, Display, Formatter},
    num::NonZeroUsize,
};

/// Fallback sealed-batch cap.
pub const DEFAULT_PLUGIN_INTENT_QUEUE_LIMIT: usize = 1024;

/// ECS publication cap for successful guest updates.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PluginIntentQueueLimit {
    /// Sealed batches accepted before owner publication.
    max_batches: NonZeroUsize,
}

impl PluginIntentQueueLimit {
    /// Creates a queue limit.
    ///
    /// # Errors
    ///
    /// Returns [`PluginIntentQueueError::ZeroLimit`] when `max_batches` is zero.
    pub const fn try_new(max_batches: usize) -> Result<Self, PluginIntentQueueError> {
        match NonZeroUsize::new(max_batches) {
            Some(max_batches) => Ok(Self { max_batches }),
            None => Err(PluginIntentQueueError::ZeroLimit),
        }
    }

    /// Converts registry scheduler policy for the ECS adapter.
    #[must_use]
    pub const fn from_scheduler_policy(policy: PluginSchedulerPolicy) -> Self {
        Self {
            max_batches: policy.max_intent_batches_limit(),
        }
    }

    /// Returns the sealed-batch cap.
    #[must_use]
    pub const fn max_batches(self) -> usize {
        self.max_batches.get()
    }

    /// Returns the sealed-batch cap as a non-zero proof for diagnostics.
    #[must_use]
    const fn max_batches_limit(self) -> NonZeroUsize {
        self.max_batches
    }
}

impl Default for PluginIntentQueueLimit {
    fn default() -> Self {
        Self::try_new(DEFAULT_PLUGIN_INTENT_QUEUE_LIMIT)
            .expect("default intent queue limit should be non-zero")
    }
}

/// Successful guest updates awaiting ECS publication.
#[derive(Resource)]
pub struct PluginIntentQueue {
    /// Queue cap.
    limit: PluginIntentQueueLimit,
    /// Sealed updates awaiting publication.
    sealed_batches: VecDeque<SealedPluginEffectBatch>,
}

impl PluginIntentQueue {
    /// Creates an empty bounded queue.
    #[must_use]
    pub const fn with_limit(limit: PluginIntentQueueLimit) -> Self {
        Self {
            limit,
            sealed_batches: VecDeque::new(),
        }
    }

    /// Enqueues a successful plugin update.
    ///
    /// Saturation keeps the sealed batch paired with the rejection so scheduler code can retry
    /// after capacity frees or explicitly discard the work with redacted cleanup evidence.
    ///
    /// # Errors
    ///
    /// Returns [`PluginIntentQueueEnqueueError`] when the queue is full.
    pub fn push_sealed(
        &mut self,
        batch: SealedPluginEffectBatch,
    ) -> Result<(), PluginIntentQueueEnqueueError> {
        self.admit(batch)?.publish();
        Ok(())
    }

    /// Validates queue admission and retains the admitted batch with the target queue.
    fn admit(
        &mut self,
        batch: SealedPluginEffectBatch,
    ) -> Result<PluginIntentQueueAdmission<'_>, PluginIntentQueueEnqueueError> {
        if batch.is_empty() {
            return Ok(PluginIntentQueueAdmission::new(
                batch,
                &mut self.sealed_batches,
            ));
        }
        if self.sealed_batches.len() >= self.limit.max_batches() {
            let source = PluginIntentQueueError::TooManyBatches {
                identity: batch.identity_proof().clone(),
                limit: self.limit.max_batches_limit(),
            };
            return Err(PluginIntentQueueEnqueueError::new(source, batch));
        }
        Ok(PluginIntentQueueAdmission::new(
            batch,
            &mut self.sealed_batches,
        ))
    }

    /// Drains successful updates in FIFO order.
    fn drain(&mut self) -> vec_deque::Drain<'_, SealedPluginEffectBatch> {
        self.sealed_batches.drain(..)
    }

    /// Returns the sealed-batch cap.
    #[must_use]
    pub const fn limit(&self) -> PluginIntentQueueLimit {
        self.limit
    }

    /// Returns the queued batch count.
    #[must_use]
    #[cfg(test)]
    pub fn len(&self) -> usize {
        self.sealed_batches.len()
    }

    /// Returns true when the queue is empty.
    #[must_use]
    #[cfg(test)]
    pub fn is_empty(&self) -> bool {
        self.sealed_batches.is_empty()
    }
}

impl Default for PluginIntentQueue {
    fn default() -> Self {
        Self::with_limit(PluginIntentQueueLimit::default())
    }
}

impl Debug for PluginIntentQueue {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginIntentQueue")
            .field("limit", &self.limit)
            .field("sealed_batch_count", &self.sealed_batches.len())
            .finish()
    }
}

/// Sealed effect batch paired with the intent owner queue that admitted it.
#[must_use]
struct PluginIntentQueueAdmission<'queue> {
    /// Admitted batch.
    batch: SealedPluginEffectBatch,
    /// Queue storage that produced the admission proof.
    sealed_batches: &'queue mut VecDeque<SealedPluginEffectBatch>,
}

impl<'queue> PluginIntentQueueAdmission<'queue> {
    /// Pairs an admitted batch with its target queue storage.
    const fn new(
        batch: SealedPluginEffectBatch,
        sealed_batches: &'queue mut VecDeque<SealedPluginEffectBatch>,
    ) -> Self {
        Self {
            batch,
            sealed_batches,
        }
    }

    /// Recovers the admitted batch without publishing it.
    fn into_batch(self) -> SealedPluginEffectBatch {
        self.batch
    }

    /// Publishes the admitted batch.
    fn publish(self) {
        let Self {
            batch,
            sealed_batches,
        } = self;
        if !batch.is_empty() {
            sealed_batches.push_back(batch);
        }
    }
}

/// Queue rejection before ECS publication.
#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
pub enum PluginIntentQueueError {
    /// Queue limit was zero.
    ZeroLimit,
    /// Queue already holds the maximum accepted sealed batches.
    TooManyBatches {
        /// Rejected plugin identity.
        identity: PluginIdentity,
        /// Maximum queued sealed batches.
        limit: NonZeroUsize,
    },
}

/// Plugin intent queue rejection class.
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum PluginIntentQueueErrorKind {
    /// Queue limit was zero.
    ZeroLimit,
    /// Sealed-batch capacity was exhausted.
    TooManyBatches,
}

/// Failed admission of a sealed plugin effect batch.
#[derive(Debug, Eq, PartialEq, thiserror::Error)]
#[error("{source}")]
pub struct PluginIntentQueueEnqueueError {
    /// Admission rejection.
    source: PluginIntentQueueError,
    /// Still-sealed effect batch retained for retry or explicit drop.
    batch: SealedPluginEffectBatch,
}

/// Retryable intent-queue enqueue failure split by queue boundary.
pub struct PluginIntentQueueEnqueueErrorParts {
    /// Closed queue rejection.
    pub queue_error: PluginIntentQueueError,
    /// Still-sealed effect batch retained for retry.
    pub batch: SealedPluginEffectBatch,
}

impl PluginIntentQueueEnqueueError {
    /// Builds a retry-safe enqueue error.
    const fn new(source: PluginIntentQueueError, batch: SealedPluginEffectBatch) -> Self {
        Self { source, batch }
    }

    /// Returns the closed queue rejection.
    #[must_use]
    pub const fn queue_error(&self) -> &PluginIntentQueueError {
        &self.source
    }

    /// Returns the sealed batch retained for retry.
    #[must_use]
    pub const fn batch(&self) -> &SealedPluginEffectBatch {
        &self.batch
    }

    /// Returns named queue rejection and retryable sealed-batch parts.
    #[must_use]
    pub fn into_parts(self) -> PluginIntentQueueEnqueueErrorParts {
        PluginIntentQueueEnqueueErrorParts {
            queue_error: self.source,
            batch: self.batch,
        }
    }

    /// Consumes the error and returns the retryable sealed batch.
    #[must_use]
    pub fn into_batch(self) -> SealedPluginEffectBatch {
        self.batch
    }

    /// Discards the retained sealed batch instead of retrying admission.
    #[must_use]
    pub fn discard_batch(self) -> PluginEffectDiscardReport {
        self.batch.discard()
    }
}

impl PluginIntentQueueErrorKind {
    /// Stable diagnostic spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::ZeroLimit => "zero-limit",
            Self::TooManyBatches => "too-many-batches",
        }
    }
}

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

impl Debug for PluginIntentQueueErrorKind {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl PluginIntentQueueError {
    /// Closed rejection class for caller decisions and diagnostics.
    #[must_use]
    pub const fn kind(&self) -> PluginIntentQueueErrorKind {
        match self {
            Self::ZeroLimit => PluginIntentQueueErrorKind::ZeroLimit,
            Self::TooManyBatches { .. } => PluginIntentQueueErrorKind::TooManyBatches,
        }
    }

    /// Returns the redacted operational event for this queue failure.
    #[must_use]
    pub fn operational_event(&self) -> Option<PluginOperationalEvent> {
        match self {
            Self::TooManyBatches { identity, limit } => {
                Some(PluginOperationalEvent::queue_saturated_for(
                    identity,
                    PluginOperationalQueue::Intent,
                    *limit,
                ))
            }
            Self::ZeroLimit => None,
        }
    }
}

impl Display for PluginIntentQueueError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ZeroLimit => formatter.write_str("plugin intent queue limit must be non-zero"),
            Self::TooManyBatches { identity, limit } => {
                let identity = identity.as_str();
                let limit = limit.get();
                write!(
                    formatter,
                    "plugin {identity:?} exceeded sealed-batch queue limit of {limit}"
                )
            }
        }
    }
}

/// Paired owner queues for successful plugin updates.
pub struct PluginOwnerQueues<'queue> {
    /// Editor effect publication.
    intent: &'queue mut PluginIntentQueue,
    /// Workspace I/O publication.
    workspace_io: &'queue mut PluginWorkspaceIoWorkerQueue,
}

impl<'queue> PluginOwnerQueues<'queue> {
    /// Borrows the owner queues as one publication boundary.
    pub const fn new(
        intent: &'queue mut PluginIntentQueue,
        workspace_io: &'queue mut PluginWorkspaceIoWorkerQueue,
    ) -> Self {
        Self {
            intent,
            workspace_io,
        }
    }

    /// Enqueues paired owner work after validating every target queue.
    ///
    /// If either queue rejects the batch, neither queue is mutated and the paired owner-work proof
    /// is returned for retry or explicit discard.
    ///
    /// # Errors
    ///
    /// Returns [`PluginOwnerQueueEnqueueError`] when either owner queue rejects admission.
    pub fn push_owner_batches(
        &mut self,
        batches: PluginHostOwnerBatches,
    ) -> Result<PluginOwnerPublicationReport, PluginOwnerQueueEnqueueError> {
        let admission = PluginOwnerQueueAdmission::validate(
            batches,
            &mut *self.intent,
            &mut *self.workspace_io,
        )?;
        Ok(admission.publish())
    }

    /// Publishes owner work from an update whose observe tasks are already runtime-scheduled.
    ///
    /// If owner queue admission fails, the error keeps the retryable owner batches and the
    /// scheduled observe-task report together. The observe tasks have already entered the runtime
    /// queue; only editor effects and workspace I/O remain unpublished.
    ///
    /// # Errors
    ///
    /// Returns [`PluginScheduledUpdateOwnerQueueError`] when either owner queue rejects admission.
    #[cfg(feature = "plugin-runtime")]
    pub fn push_scheduled_update(
        &mut self,
        update: WasmtimePluginScheduledUpdate,
    ) -> Result<PluginScheduledUpdateOwnerPublicationReport, PluginScheduledUpdateOwnerQueueError>
    {
        let update = update.into_parts();
        match self.push_owner_batches(update.owner_batches) {
            Ok(owner_publication) => Ok(PluginScheduledUpdateOwnerPublicationReport::new(
                owner_publication,
                update.scheduled_workspace_observe_tasks,
            )),
            Err(source) => Err(PluginScheduledUpdateOwnerQueueError::new(
                source,
                update.scheduled_workspace_observe_tasks,
            )),
        }
    }

    /// Retries owner publication after observe tasks were already runtime-scheduled.
    ///
    /// A failed retry returns the same paired error shape, so the scheduled task report cannot be
    /// dropped while the owner work remains retryable.
    ///
    /// # Errors
    ///
    /// Returns [`PluginScheduledUpdateOwnerQueueError`] when either owner queue still rejects
    /// admission.
    #[cfg(feature = "plugin-runtime")]
    pub fn retry_scheduled_update_owner_publication(
        &mut self,
        error: PluginScheduledUpdateOwnerQueueError,
    ) -> Result<PluginScheduledUpdateOwnerPublicationReport, PluginScheduledUpdateOwnerQueueError>
    {
        let error = error.into_parts();
        let owner_queue_error = error.owner_queue_error.into_parts();
        match self.push_owner_batches(owner_queue_error.owner_batches) {
            Ok(owner_publication) => Ok(PluginScheduledUpdateOwnerPublicationReport::new(
                owner_publication,
                error.scheduled_workspace_observe_tasks,
            )),
            Err(source) => Err(PluginScheduledUpdateOwnerQueueError::new(
                source,
                error.scheduled_workspace_observe_tasks,
            )),
        }
    }
}

/// Paired owner work proven to fit the borrowed target queues.
struct PluginOwnerQueueAdmission<'queue> {
    /// Redacted publication report captured before queue mutation.
    report: PluginOwnerPublicationReport,
    /// Editor-effect owner queue admission.
    intent_admission: PluginIntentQueueAdmission<'queue>,
    /// Workspace I/O owner queue admission.
    workspace_io_admission: crate::plugin::PluginWorkspaceIoWorkerQueueAdmission<'queue>,
}

impl<'queue> PluginOwnerQueueAdmission<'queue> {
    /// Validates all owner queues before mutation starts.
    fn validate(
        batches: PluginHostOwnerBatches,
        intent: &'queue mut PluginIntentQueue,
        workspace_io: &'queue mut PluginWorkspaceIoWorkerQueue,
    ) -> Result<Self, PluginOwnerQueueEnqueueError> {
        let report = PluginOwnerPublicationReport::from_batches(&batches);
        let parts = batches.into_parts();
        let intent_admission = match intent.admit(parts.effects) {
            Ok(admission) => admission,
            Err(source) => {
                let source = source.into_parts();
                return Err(PluginOwnerQueueEnqueueError::new(
                    PluginOwnerQueueError::Intent(source.queue_error),
                    PluginHostOwnerBatches::from_parts(source.batch, parts.workspace_io),
                ));
            }
        };
        let workspace_io_admission = match workspace_io.admit(parts.workspace_io) {
            Ok(admission) => admission,
            Err(source) => {
                let source = source.into_parts();
                let effects = intent_admission.into_batch();
                return Err(PluginOwnerQueueEnqueueError::new(
                    PluginOwnerQueueError::WorkspaceIo(source.queue_error),
                    PluginHostOwnerBatches::from_parts(effects, source.batch),
                ));
            }
        };
        Ok(Self {
            report,
            intent_admission,
            workspace_io_admission,
        })
    }

    /// Publishes already-admitted owner work.
    fn publish(self) -> PluginOwnerPublicationReport {
        let Self {
            report,
            intent_admission,
            workspace_io_admission,
        } = self;
        intent_admission.publish();
        workspace_io_admission.publish();
        report
    }
}

/// Successful publication of paired owner work.
#[derive(Clone, Eq, PartialEq)]
pub struct PluginOwnerPublicationReport {
    /// Plugin identity whose owner work was published.
    identity: PluginIdentity,
    /// Editor effects admitted to the intent owner queue.
    effect_count: usize,
    /// Workspace I/O requests admitted to the filesystem-policy worker queue.
    workspace_io_request_count: usize,
}

impl PluginOwnerPublicationReport {
    /// Captures the redacted shape of owner work admitted to both owner queues.
    fn from_batches(batches: &PluginHostOwnerBatches) -> Self {
        Self {
            identity: batches.identity_proof().clone(),
            effect_count: batches.effects().effect_count(),
            workspace_io_request_count: batches.workspace_io().request_count(),
        }
    }

    /// Returns the plugin identity whose owner work was published.
    #[must_use]
    pub fn identity(&self) -> &str {
        self.identity_proof().as_str()
    }

    /// Returns the validated identity whose owner work was published.
    #[must_use]
    pub const fn identity_proof(&self) -> &PluginIdentity {
        &self.identity
    }

    /// Returns the number of editor effects admitted to the intent owner queue.
    #[must_use]
    pub const fn effect_count(&self) -> usize {
        self.effect_count
    }

    /// Returns the number of workspace I/O requests admitted to the worker queue.
    #[must_use]
    pub const fn workspace_io_request_count(&self) -> usize {
        self.workspace_io_request_count
    }
}

impl Debug for PluginOwnerPublicationReport {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginOwnerPublicationReport")
            .field("identity", self.identity_proof())
            .field("effect_count", &self.effect_count)
            .field(
                "workspace_io_request_count",
                &self.workspace_io_request_count,
            )
            .finish()
    }
}

/// Owner queue rejection class.
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum PluginOwnerQueueErrorKind {
    /// Editor effect queue rejected admission.
    Intent,
    /// Workspace I/O queue rejected admission.
    WorkspaceIo,
}

impl PluginOwnerQueueErrorKind {
    /// Stable diagnostic spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Intent => "intent",
            Self::WorkspaceIo => "workspace-io",
        }
    }
}

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

impl Debug for PluginOwnerQueueErrorKind {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Failed owner queue admission source.
#[derive(Debug, Eq, PartialEq, thiserror::Error)]
pub enum PluginOwnerQueueError {
    /// Editor effect queue rejected admission.
    #[error("plugin owner effect queue rejected publication: {0}")]
    Intent(#[source] PluginIntentQueueError),
    /// Workspace I/O queue rejected admission.
    #[error("plugin owner workspace I/O queue rejected publication: {0}")]
    WorkspaceIo(#[source] PluginWorkspaceIoWorkerQueueError),
}

impl PluginOwnerQueueError {
    /// Closed rejection class for caller decisions and diagnostics.
    #[must_use]
    pub const fn kind(&self) -> PluginOwnerQueueErrorKind {
        match self {
            Self::Intent(_) => PluginOwnerQueueErrorKind::Intent,
            Self::WorkspaceIo(_) => PluginOwnerQueueErrorKind::WorkspaceIo,
        }
    }

    /// Returns the redacted operational event for this queue failure.
    #[must_use]
    pub fn operational_event(&self) -> Option<PluginOperationalEvent> {
        match self {
            Self::Intent(source) => source.operational_event(),
            Self::WorkspaceIo(source) => source.operational_event(),
        }
    }
}

/// Failed admission of paired owner work.
#[derive(Eq, PartialEq, thiserror::Error)]
#[error("{source}")]
pub struct PluginOwnerQueueEnqueueError {
    /// Queue rejection.
    source: PluginOwnerQueueError,
    /// Still-sealed owner work retained for retry or explicit drop.
    batches: Box<PluginHostOwnerBatches>,
}

/// Retryable owner-queue failure split by owner boundary.
pub struct PluginOwnerQueueEnqueueErrorParts {
    /// Closed owner queue rejection.
    pub queue_error: PluginOwnerQueueError,
    /// Paired owner work retained for retry.
    pub owner_batches: PluginHostOwnerBatches,
}

impl PluginOwnerQueueEnqueueError {
    /// Builds a retry-safe owner queue error.
    fn new(source: PluginOwnerQueueError, batches: PluginHostOwnerBatches) -> Self {
        Self {
            source,
            batches: Box::new(batches),
        }
    }

    /// Returns the closed owner queue rejection.
    #[must_use]
    pub const fn queue_error(&self) -> &PluginOwnerQueueError {
        &self.source
    }

    /// Returns the paired owner work retained for retry.
    #[must_use]
    pub fn owner_batches(&self) -> &PluginHostOwnerBatches {
        &self.batches
    }

    /// Returns the redacted operational event for this queue failure.
    #[must_use]
    pub fn operational_event(&self) -> Option<PluginOperationalEvent> {
        self.source.operational_event()
    }

    /// Returns named queue rejection and retryable owner-work parts.
    #[must_use]
    pub fn into_parts(self) -> PluginOwnerQueueEnqueueErrorParts {
        PluginOwnerQueueEnqueueErrorParts {
            queue_error: self.source,
            owner_batches: *self.batches,
        }
    }

    /// Consumes the error and returns the retryable owner work.
    #[must_use]
    pub fn into_owner_batches(self) -> PluginHostOwnerBatches {
        self.into_parts().owner_batches
    }

    /// Discards the retained owner work instead of retrying admission.
    #[must_use]
    pub fn discard_owner_batches(self) -> PluginHostOwnerDiscardReport {
        self.into_parts().owner_batches.discard()
    }
}

impl Debug for PluginOwnerQueueEnqueueError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginOwnerQueueEnqueueError")
            .field("queue", &self.source.kind())
            .field("identity", self.batches.identity_proof())
            .field("effect_count", &self.batches.effects().effect_count())
            .field(
                "workspace_io_request_count",
                &self.batches.workspace_io().request_count(),
            )
            .finish()
    }
}

/// Successful owner publication for a scheduled Wasmtime update.
#[cfg(feature = "plugin-runtime")]
#[derive(Clone, Eq, PartialEq)]
pub struct PluginScheduledUpdateOwnerPublicationReport {
    /// Owner work admitted after runtime task scheduling.
    owner_publication: PluginOwnerPublicationReport,
    /// Runtime task scheduling report admitted before owner publication.
    scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
}

/// Scheduled-update publication evidence split by owner boundary.
#[cfg(feature = "plugin-runtime")]
pub struct PluginScheduledUpdateOwnerPublicationReportParts {
    /// Owner work admitted after runtime task scheduling.
    pub owner_publication: PluginOwnerPublicationReport,
    /// Runtime task scheduling report admitted before owner publication.
    pub scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
}

#[cfg(feature = "plugin-runtime")]
impl PluginScheduledUpdateOwnerPublicationReport {
    /// Builds a success report after paired owner queues accept the remaining update work.
    fn new(
        owner_publication: PluginOwnerPublicationReport,
        scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
    ) -> Self {
        assert_eq!(
            owner_publication.identity_proof(),
            scheduled_workspace_observe_tasks.identity_proof(),
            "scheduled update owner publication report must share one plugin identity",
        );
        Self {
            owner_publication,
            scheduled_workspace_observe_tasks,
        }
    }

    /// Returns the plugin identity whose scheduled update was published.
    #[must_use]
    pub fn identity(&self) -> &str {
        self.identity_proof().as_str()
    }

    /// Returns the validated identity for the scheduled update.
    #[must_use]
    pub const fn identity_proof(&self) -> &PluginIdentity {
        self.owner_publication.identity_proof()
    }

    /// Returns the paired owner publication report.
    #[must_use]
    pub const fn owner_publication(&self) -> &PluginOwnerPublicationReport {
        &self.owner_publication
    }

    /// Returns the number of editor effects admitted to the intent owner queue.
    #[must_use]
    pub const fn effect_count(&self) -> usize {
        self.owner_publication.effect_count()
    }

    /// Returns the number of workspace I/O requests admitted to the worker queue.
    #[must_use]
    pub const fn workspace_io_request_count(&self) -> usize {
        self.owner_publication.workspace_io_request_count()
    }

    /// Returns the observe-task scheduling report already admitted to the runtime queue.
    #[must_use]
    pub const fn scheduled_workspace_observe_tasks(
        &self,
    ) -> &PluginWorkspaceObserveTaskEnqueueReport {
        &self.scheduled_workspace_observe_tasks
    }

    /// Splits owner-publication evidence from the scheduled task report.
    #[must_use]
    pub fn into_parts(self) -> PluginScheduledUpdateOwnerPublicationReportParts {
        PluginScheduledUpdateOwnerPublicationReportParts {
            owner_publication: self.owner_publication,
            scheduled_workspace_observe_tasks: self.scheduled_workspace_observe_tasks,
        }
    }
}

#[cfg(feature = "plugin-runtime")]
impl Debug for PluginScheduledUpdateOwnerPublicationReport {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginScheduledUpdateOwnerPublicationReport")
            .field("identity", self.identity_proof())
            .field("effect_count", &self.owner_publication.effect_count())
            .field(
                "workspace_io_request_count",
                &self.owner_publication.workspace_io_request_count(),
            )
            .field(
                "scheduled_workspace_observe_task_count",
                &self.scheduled_workspace_observe_tasks.task_count(),
            )
            .finish()
    }
}

/// Discard report for unpublished owner work after observe tasks were scheduled.
#[cfg(feature = "plugin-runtime")]
#[derive(Eq, PartialEq)]
pub struct PluginScheduledUpdateOwnerDiscardReport {
    /// Owner work discarded instead of retried.
    owner_work: PluginHostOwnerDiscardReport,
    /// Runtime task scheduling report admitted before owner publication.
    scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
}

/// Scheduled-update owner discard report split by owner boundary.
#[cfg(feature = "plugin-runtime")]
pub struct PluginScheduledUpdateOwnerDiscardReportParts {
    /// Owner work discarded instead of retried.
    pub owner_work: PluginHostOwnerDiscardReport,
    /// Runtime task scheduling report admitted before owner publication.
    pub scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
}

#[cfg(feature = "plugin-runtime")]
impl PluginScheduledUpdateOwnerDiscardReport {
    /// Builds the paired discard report for a scheduled update.
    fn new(
        owner_work: PluginHostOwnerDiscardReport,
        scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
    ) -> Self {
        assert_eq!(
            owner_work.identity_proof(),
            scheduled_workspace_observe_tasks.identity_proof(),
            "scheduled update owner discard report must share one plugin identity",
        );
        Self {
            owner_work,
            scheduled_workspace_observe_tasks,
        }
    }

    /// Returns the plugin identity whose owner work was discarded.
    #[must_use]
    pub fn identity(&self) -> &str {
        self.identity_proof().as_str()
    }

    /// Returns the validated identity shared by both report halves.
    #[must_use]
    pub const fn identity_proof(&self) -> &PluginIdentity {
        self.owner_work.identity_proof()
    }

    /// Returns the discarded owner work report.
    #[must_use]
    pub const fn owner_work(&self) -> &PluginHostOwnerDiscardReport {
        &self.owner_work
    }

    /// Returns the observe-task scheduling report already admitted to the runtime queue.
    #[must_use]
    pub const fn scheduled_workspace_observe_tasks(
        &self,
    ) -> &PluginWorkspaceObserveTaskEnqueueReport {
        &self.scheduled_workspace_observe_tasks
    }

    /// Splits the paired report by owner boundary.
    #[must_use]
    pub fn into_parts(self) -> PluginScheduledUpdateOwnerDiscardReportParts {
        PluginScheduledUpdateOwnerDiscardReportParts {
            owner_work: self.owner_work,
            scheduled_workspace_observe_tasks: self.scheduled_workspace_observe_tasks,
        }
    }
}

#[cfg(feature = "plugin-runtime")]
impl Debug for PluginScheduledUpdateOwnerDiscardReport {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginScheduledUpdateOwnerDiscardReport")
            .field("identity", self.identity_proof())
            .field(
                "discarded_effect_count",
                &self.owner_work.effects().discarded_effects(),
            )
            .field(
                "discarded_workspace_io_request_count",
                &self.owner_work.workspace_io().discarded_requests(),
            )
            .field(
                "scheduled_workspace_observe_task_count",
                &self.scheduled_workspace_observe_tasks.task_count(),
            )
            .finish()
    }
}

/// Owner queue failure after a Wasmtime update scheduled observe tasks.
#[cfg(feature = "plugin-runtime")]
#[derive(Eq, PartialEq, thiserror::Error)]
#[error("{source}")]
pub struct PluginScheduledUpdateOwnerQueueError {
    /// Retryable owner queue failure.
    source: PluginOwnerQueueEnqueueError,
    /// Runtime task scheduling report already admitted before owner publication.
    scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
}

/// Retryable scheduled-update owner failure split by retry owner.
#[cfg(feature = "plugin-runtime")]
pub struct PluginScheduledUpdateOwnerQueueErrorParts {
    /// Retryable owner queue failure.
    pub owner_queue_error: PluginOwnerQueueEnqueueError,
    /// Runtime task scheduling report already admitted before owner publication.
    pub scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
}

#[cfg(feature = "plugin-runtime")]
impl PluginScheduledUpdateOwnerQueueError {
    /// Builds a retry-safe owner publication failure.
    fn new(
        source: PluginOwnerQueueEnqueueError,
        scheduled_workspace_observe_tasks: PluginWorkspaceObserveTaskEnqueueReport,
    ) -> Self {
        assert_eq!(
            source.owner_batches().identity_proof(),
            scheduled_workspace_observe_tasks.identity_proof(),
            "scheduled update owner queue error must share one plugin identity",
        );
        Self {
            source,
            scheduled_workspace_observe_tasks,
        }
    }

    /// Returns the retry-safe owner queue failure.
    #[must_use]
    pub const fn owner_queue_error(&self) -> &PluginOwnerQueueEnqueueError {
        &self.source
    }

    /// Returns the closed owner queue rejection.
    #[must_use]
    pub const fn queue_error(&self) -> &PluginOwnerQueueError {
        self.source.queue_error()
    }

    /// Returns the paired owner work retained for retry.
    #[must_use]
    pub fn owner_batches(&self) -> &PluginHostOwnerBatches {
        self.source.owner_batches()
    }

    /// Returns the observe-task scheduling report already admitted to the runtime queue.
    #[must_use]
    pub const fn scheduled_workspace_observe_tasks(
        &self,
    ) -> &PluginWorkspaceObserveTaskEnqueueReport {
        &self.scheduled_workspace_observe_tasks
    }

    /// Returns the redacted operational event for the owner queue failure.
    #[must_use]
    pub fn operational_event(&self) -> Option<PluginOperationalEvent> {
        self.source.operational_event()
    }

    /// Splits the retryable owner failure from the already-scheduled task report.
    #[must_use]
    pub fn into_parts(self) -> PluginScheduledUpdateOwnerQueueErrorParts {
        PluginScheduledUpdateOwnerQueueErrorParts {
            owner_queue_error: self.source,
            scheduled_workspace_observe_tasks: self.scheduled_workspace_observe_tasks,
        }
    }

    /// Discards unpublished owner work while retaining the already-scheduled task report.
    #[must_use]
    pub fn discard_owner_batches(self) -> PluginScheduledUpdateOwnerDiscardReport {
        let PluginScheduledUpdateOwnerQueueErrorParts {
            owner_queue_error,
            scheduled_workspace_observe_tasks,
        } = self.into_parts();
        PluginScheduledUpdateOwnerDiscardReport::new(
            owner_queue_error.discard_owner_batches(),
            scheduled_workspace_observe_tasks,
        )
    }
}

#[cfg(feature = "plugin-runtime")]
impl Debug for PluginScheduledUpdateOwnerQueueError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginScheduledUpdateOwnerQueueError")
            .field("queue", &self.source.queue_error().kind())
            .field("identity", self.source.owner_batches().identity_proof())
            .field(
                "effect_count",
                &self.source.owner_batches().effects().effect_count(),
            )
            .field(
                "workspace_io_request_count",
                &self.source.owner_batches().workspace_io().request_count(),
            )
            .field(
                "scheduled_workspace_observe_task_count",
                &self.scheduled_workspace_observe_tasks.task_count(),
            )
            .finish()
    }
}

/// Publishes plugin effects before mutation owners run.
#[derive(Clone, Copy, Debug, Default)]
pub struct PluginRuntimeEcsPlugin;

impl Plugin for PluginRuntimeEcsPlugin {
    fn build(&self, app: &mut App) {
        if app.world().get_resource::<PluginIntentQueue>().is_none() {
            let limit = app.world().get_resource::<AppConfig>().map_or_else(
                PluginIntentQueueLimit::default,
                |config| {
                    PluginIntentQueueLimit::from_scheduler_policy(
                        PluginSchedulerPolicy::from_validated_registry(config.plugins()),
                    )
                },
            );
            let _app = app.insert_resource(PluginIntentQueue::with_limit(limit));
        }

        if app.world().get_resource::<PluginProposalLane>().is_none() {
            let _app = app.insert_resource(PluginProposalLane::default());
        }
        if app
            .world()
            .get_resource::<PluginProposalLanePolicy>()
            .is_none()
        {
            let policy = app
                .world()
                .get_resource::<AppConfig>()
                .map_or_else(PluginProposalLanePolicy::default, |config| {
                    PluginProposalLanePolicy::from_validated_registry(config.plugins())
                });
            let _app = app.insert_resource(policy);
        }
        if app
            .world()
            .get_resource::<PluginProposalReceiptLog>()
            .is_none()
        {
            let _app = app.insert_resource(PluginProposalReceiptLog::default());
        }

        let _app = app.add_systems(
            Update,
            (
                drain_plugin_effects,
                apply_plugin_buffer_edit_proposals,
                reject_plugin_buffer_edit_proposals,
            )
                .chain()
                .in_set(EditorSet::PluginIntent),
        );
        let _app = app.add_systems(
            Update,
            record_plugin_buffer_edit_proposal_receipts.in_set(EditorSet::Layout),
        );
    }
}

/// Publishes sealed effects as ECS requests.
#[allow(clippy::needless_pass_by_value)]
fn drain_plugin_effects(
    mut queue: ResMut<PluginIntentQueue>,
    mut proposals: ResMut<PluginProposalLane>,
    policy: bevy::prelude::Res<PluginProposalLanePolicy>,
    mut buffer_edits: MessageWriter<RevisionGuardedBufferEditRequested>,
    mut status_messages: MessageWriter<StatusMessageRequested>,
    mut operational_events: MessageWriter<PluginOperationalEventReported>,
) {
    for batch in queue.drain() {
        let drained_requests = batch.drain().into_requests();
        for request in drained_requests.buffer_edits {
            publish_or_store_buffer_edit_proposal(
                request,
                &mut proposals,
                &policy,
                &mut buffer_edits,
                &mut operational_events,
            );
        }
        for request in drained_requests.status_messages {
            let _sent = status_messages.write(request);
        }
    }
}

/// Routes one plugin proposal according to the configured review policy.
fn publish_or_store_buffer_edit_proposal(
    request: RevisionGuardedBufferEditRequested,
    proposals: &mut PluginProposalLane,
    policy: &PluginProposalLanePolicy,
    buffer_edits: &mut MessageWriter<RevisionGuardedBufferEditRequested>,
    operational_events: &mut MessageWriter<PluginOperationalEventReported>,
) {
    let review_mode = policy.buffer_edit_mode_for(request.provenance.source_identity_proof());
    let id = match proposals.push_buffer_edit(request) {
        Ok(id) => id,
        Err(error) => {
            if let Some(event) = error.operational_event() {
                let _sent = operational_events.write(PluginOperationalEventReported { event });
            }
            return;
        }
    };
    if review_mode == PluginProposalReviewMode::AutoApply {
        if let Ok(request) = proposals.take_buffer_edit_for_apply(id) {
            let _sent = buffer_edits.write(request);
        }
    }
}

/// Applies explicitly accepted plugin buffer edit proposals through the buffer owner.
fn apply_plugin_buffer_edit_proposals(
    mut requests: bevy::prelude::MessageReader<PluginBufferEditProposalApplyRequested>,
    mut proposals: ResMut<PluginProposalLane>,
    mut buffer_edits: MessageWriter<RevisionGuardedBufferEditRequested>,
    mut rejected: MessageWriter<PluginBufferEditProposalDecisionRejected>,
) {
    for request in requests.read() {
        match proposals.take_buffer_edit_for_apply(request.proposal) {
            Ok(edit) => {
                let _sent = buffer_edits.write(edit);
            }
            Err(error) => {
                let _sent = rejected.write(proposal_decision_rejection(
                    request.proposal,
                    PluginBufferEditProposalDecision::Apply,
                    &error,
                ));
            }
        }
    }
}

/// Drops explicitly rejected plugin buffer edit proposals before mutation.
fn reject_plugin_buffer_edit_proposals(
    mut requests: bevy::prelude::MessageReader<PluginBufferEditProposalRejectRequested>,
    mut proposals: ResMut<PluginProposalLane>,
    mut rejected: MessageWriter<PluginBufferEditProposalDecisionRejected>,
    mut receipts: MessageWriter<PluginBufferEditProposalReceipt>,
) {
    for request in requests.read() {
        match proposals.reject_buffer_edit(request.proposal) {
            Ok(proposal) => {
                let _sent = receipts.write(PluginBufferEditProposalReceipt::rejected_before_owner(
                    proposal.id(),
                    proposal.provenance(),
                    proposal.target(),
                    proposal.rejected().clone(),
                ));
            }
            Err(error) => {
                let _sent = rejected.write(proposal_decision_rejection(
                    request.proposal,
                    PluginBufferEditProposalDecision::Reject,
                    &error,
                ));
            }
        }
    }
}

/// Converts proposal-lane errors into the ECS rejection vocabulary.
const fn proposal_decision_rejection(
    proposal: crate::ecs::events::edit::PluginProposalId,
    decision: PluginBufferEditProposalDecision,
    error: &crate::ecs::plugin_proposal::PluginProposalDecisionError,
) -> PluginBufferEditProposalDecisionRejected {
    let reason = match error {
        crate::ecs::plugin_proposal::PluginProposalDecisionError::UnknownProposal { id } => {
            PluginProposalDecisionRejectionReason::UnknownProposal { id: *id }
        }
    };
    PluginBufferEditProposalDecisionRejected {
        proposal,
        decision,
        reason,
    }
}

/// Retains proposal receipts after owner systems have made their decisions.
fn record_plugin_buffer_edit_proposal_receipts(
    mut receipts: bevy::prelude::MessageReader<PluginBufferEditProposalReceipt>,
    mut log: ResMut<PluginProposalReceiptLog>,
) {
    for receipt in receipts.read() {
        log.push(receipt.clone());
    }
}

#[cfg(test)]
mod tests {
    use super::{
        PluginIntentQueue, PluginIntentQueueEnqueueError, PluginIntentQueueEnqueueErrorParts,
        PluginIntentQueueError, PluginIntentQueueErrorKind, PluginIntentQueueLimit,
        PluginOwnerPublicationReport, PluginOwnerQueueEnqueueError,
        PluginOwnerQueueEnqueueErrorParts, PluginOwnerQueueError, PluginOwnerQueueErrorKind,
        PluginOwnerQueues,
    };
    use crate::ecs::plugin_proposal::{
        PluginProposalLane, PluginProposalLanePolicy, PluginProposalReceiptLog,
    };
    use crate::{
        buffer::BufferEdit,
        config::AppConfig,
        ecs::{
            BufferPlugin, EditorCorePlugin, InitialEditorBuffer,
            components::{
                buffer::{
                    BufferEntity, BufferRevision, BufferText, EditorBuffer, EditorView, ViewEntity,
                },
                vim::VimModalState,
            },
            events::{
                edit::{
                    BufferEditRejected, BufferEditRejectionReason, BufferEditRequested,
                    BufferEdited, PluginBufferEditProposalApplyRequested,
                    PluginBufferEditProposalDecision, PluginBufferEditProposalDecisionRejected,
                    PluginBufferEditProposalReceipt, PluginBufferEditProposalReceiptOutcome,
                    PluginBufferEditProposalReceiptOwner,
                    PluginBufferEditProposalReceiptRejectionReason,
                    PluginBufferEditProposalRejectRequested, PluginProposalDecisionRejectionReason,
                    PluginProposalId,
                },
                plugin::PluginOperationalEventReported,
                status::StatusMessageRejected,
            },
            schedules::EditorSet,
        },
        plugin::{
            PendingPluginUpdateBatch, PluginAuthorizationError, PluginBufferEditProposal,
            PluginCapabilities, PluginCapabilityShape, PluginEffectBatchLimit, PluginHandleError,
            PluginHandleStore, PluginHostContext, PluginHostImportError, PluginHostOwnerBatches,
            PluginIdentity, PluginOperationalEventKind, PluginOperationalQueue,
            PluginProposalReviewMode, PluginWorkspaceIoWorkerQueue,
            PluginWorkspaceIoWorkerQueueLimit, SealedPluginEffectBatch, WorkspacePathGrant,
            WorkspacePathRef,
        },
        text_stream::{TextByteStream, TextRevision},
        vim::VimStatusMessage,
    };
    use bevy::prelude::{App, Entity, IntoScheduleConfigs, MessageReader, Resource, Update, With};
    use std::num::{NonZeroU64, NonZeroUsize};
    use std::path::Path;

    #[derive(Default, Resource)]
    struct ObservedPluginOwners {
        edited: Vec<BufferEdited>,
        edit_rejected: Vec<BufferEditRejected>,
        proposal_decision_rejected: Vec<PluginBufferEditProposalDecisionRejected>,
        proposal_receipts: Vec<PluginBufferEditProposalReceipt>,
        status_rejected: Vec<StatusMessageRejected>,
        operational_events: Vec<PluginOperationalEventReported>,
    }

    fn assert_send_sync<T: Send + Sync>() {}

    #[test]
    fn plugin_intent_queue_retry_error_is_send_sync() {
        assert_send_sync::<PluginIntentQueueEnqueueError>();
        assert_send_sync::<PluginIntentQueueEnqueueErrorParts>();
        assert_send_sync::<PluginOwnerQueueEnqueueError>();
        assert_send_sync::<PluginOwnerQueueEnqueueErrorParts>();
        assert_send_sync::<PluginOwnerPublicationReport>();
        #[cfg(feature = "plugin-runtime")]
        assert_send_sync::<super::PluginScheduledUpdateOwnerPublicationReport>();
        #[cfg(feature = "plugin-runtime")]
        assert_send_sync::<super::PluginScheduledUpdateOwnerDiscardReport>();
        #[cfg(feature = "plugin-runtime")]
        assert_send_sync::<super::PluginScheduledUpdateOwnerQueueError>();
    }

    #[test]
    fn plugin_intent_queue_error_kinds_are_closed() {
        let identity = PluginIdentity::try_new("intent-test").expect("identity");
        let cases = [
            (
                PluginIntentQueueError::ZeroLimit,
                PluginIntentQueueErrorKind::ZeroLimit,
                "zero-limit",
            ),
            (
                PluginIntentQueueError::TooManyBatches {
                    identity,
                    limit: count_limit(1),
                },
                PluginIntentQueueErrorKind::TooManyBatches,
                "too-many-batches",
            ),
        ];

        for (error, kind, expected) in cases {
            assert_eq!(error.kind(), kind);
            assert_eq!(kind.as_str(), expected);
            assert_eq!(kind.to_string(), expected);
            assert_eq!(format!("{kind:?}"), expected);
        }
    }

    #[test]
    fn plugin_owner_queue_error_kinds_are_closed() {
        let identity = PluginIdentity::try_new("owner-test").expect("identity");
        let cases = [
            (
                PluginOwnerQueueError::Intent(PluginIntentQueueError::TooManyBatches {
                    identity: identity.clone(),
                    limit: count_limit(1),
                }),
                PluginOwnerQueueErrorKind::Intent,
                "intent",
            ),
            (
                PluginOwnerQueueError::WorkspaceIo(
                    crate::plugin::PluginWorkspaceIoWorkerQueueError::TooManyBatches {
                        identity,
                        limit: count_limit(1),
                    },
                ),
                PluginOwnerQueueErrorKind::WorkspaceIo,
                "workspace-io",
            ),
        ];

        for (error, kind, expected) in cases {
            assert_eq!(error.kind(), kind);
            assert_eq!(kind.as_str(), expected);
            assert_eq!(kind.to_string(), expected);
            assert_eq!(format!("{kind:?}"), expected);
        }
    }

    #[test]
    fn plugin_intent_drain_runs_before_edit_owner_mutation() {
        let mut app = plugin_test_app();
        let buffer = editor_buffer_entity(&mut app);
        let view = editor_view_entity(&mut app);
        enqueue_plugin_effects(&mut app, BufferEntity(buffer), ViewEntity(view));

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer)
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "pluginalma");

        let modal_state = app
            .world()
            .get::<VimModalState>(view)
            .expect("editor view should have modal state");
        assert_eq!(
            modal_state.editor.status.message(),
            Some(&VimStatusMessage::Info(
                crate::StatusInfoText::escaped_display("plugin ok")
            ))
        );

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert_eq!(observed.edited.len(), 1);
        assert!(observed.edit_rejected.is_empty());
        assert!(observed.status_rejected.is_empty());
        assert!(app.world().resource::<PluginIntentQueue>().is_empty());
    }

    #[test]
    fn plugin_intent_drain_preserves_owner_rejection_boundary() {
        let mut app = plugin_test_app();
        let view = editor_view_entity(&mut app);
        let missing_buffer = BufferEntity(Entity::from_raw_u32(99).expect("test entity"));
        enqueue_batch(&mut app, missing_buffer, ViewEntity(view));

        app.update();
        app.update();

        let buffer = editor_buffer_entity(&mut app);
        let buffer_text = app
            .world()
            .get::<BufferText>(buffer)
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma");

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert_eq!(observed.edit_rejected.len(), 1);
        assert!(observed.status_rejected.is_empty());
    }

    #[test]
    fn plugin_intent_drains_batches_in_fifo_order() {
        let mut app = plugin_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));

        enqueue_text_insert(&mut app, buffer, view, 0, "a", TextRevision::from(0));
        enqueue_text_insert(&mut app, buffer, view, 1, "b", TextRevision::from(1));

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "abalma");

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert_eq!(observed.edited.len(), 2);
        assert!(observed.edit_rejected.is_empty());
        assert!(app.world().resource::<PluginIntentQueue>().is_empty());
    }

    #[test]
    fn plugin_intent_drain_keeps_stale_revision_rejection_at_edit_owner() {
        let mut app = plugin_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));

        let _message = app.world_mut().write_message(BufferEditRequested {
            target: buffer,
            edit: BufferEdit::insert(4, "!"),
        });
        app.update();
        app.update();
        app.world_mut()
            .resource_mut::<ObservedPluginOwners>()
            .clear();

        enqueue_text_insert(&mut app, buffer, view, 0, "plugin ", TextRevision::from(0));
        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma!");

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert_eq!(observed.edit_rejected.len(), 1);
        assert_eq!(
            observed.edit_rejected[0].reason,
            BufferEditRejectionReason::StaleRevision {
                expected: TextRevision::from(0),
                actual: TextRevision::from(1),
            }
        );
        assert_eq!(observed.proposal_receipts.len(), 1);
        let receipt = &observed.proposal_receipts[0];
        assert_eq!(receipt.source_identity(), "integration");
        assert_eq!(receipt.target(), buffer);
        assert_eq!(receipt.base_revision(), TextRevision::from(0));
        assert_eq!(
            receipt.outcome(),
            &PluginBufferEditProposalReceiptOutcome::Rejected {
                owner: PluginBufferEditProposalReceiptOwner::BufferOwner,
                reason: PluginBufferEditProposalReceiptRejectionReason::StaleRevision {
                    expected: TextRevision::from(0),
                    actual: TextRevision::from(1),
                },
            }
        );
        let log = app.world().resource::<PluginProposalReceiptLog>();
        assert_eq!(log.len(), 1);
        assert_eq!(
            log.receipts().next().expect("retained receipt").outcome(),
            receipt.outcome()
        );
    }

    #[test]
    fn manual_proposal_policy_stores_plugin_edits_without_owner_mutation() {
        let mut app = manual_proposal_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));

        enqueue_text_insert(&mut app, buffer, view, 0, "plugin ", TextRevision::from(0));
        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma");

        let proposals = app.world().resource::<PluginProposalLane>();
        assert_eq!(proposals.len(), 1);
        let id = proposals.first_pending_id().expect("proposal id");
        let proposal = proposals
            .pending_buffer_edit(id)
            .expect("proposal should remain pending");
        assert_eq!(proposal.id(), id);
        assert_eq!(proposal.target(), buffer);
        assert_eq!(proposal.provenance().source_identity(), "integration");
        assert_eq!(proposal.provenance().base_revision(), TextRevision::from(0));
        assert_eq!(proposal.provenance().proposal_id(), Some(id));
        assert_eq!(
            proposal.edit_shape().kind(),
            crate::buffer::BufferEditKind::Insert
        );

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
    }

    #[test]
    fn applying_manual_plugin_proposal_routes_through_buffer_owner() {
        let mut app = manual_proposal_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));

        enqueue_text_insert(&mut app, buffer, view, 0, "plugin ", TextRevision::from(0));
        app.update();
        let id = app
            .world()
            .resource::<PluginProposalLane>()
            .first_pending_id()
            .expect("proposal id");
        let _message = app
            .world_mut()
            .write_message(PluginBufferEditProposalApplyRequested { proposal: id });

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "plugin alma");
        assert!(app.world().resource::<PluginProposalLane>().is_empty());

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert_eq!(observed.edited.len(), 1);
        assert!(observed.edit_rejected.is_empty());
        assert_eq!(observed.proposal_receipts.len(), 1);
        let receipt = &observed.proposal_receipts[0];
        assert_eq!(receipt.proposal(), id);
        assert_eq!(receipt.source_identity(), "integration");
        assert_eq!(receipt.target(), buffer);
        assert_eq!(receipt.base_revision(), TextRevision::from(0));
        assert_eq!(
            receipt.outcome(),
            &PluginBufferEditProposalReceiptOutcome::Applied {
                previous_revision: BufferRevision {
                    revision: TextRevision::from(0),
                },
                current_revision: BufferRevision {
                    revision: TextRevision::from(1),
                },
            }
        );
        let log = app.world().resource::<PluginProposalReceiptLog>();
        assert_eq!(log.len(), 1);
        assert_eq!(
            log.receipts().next().expect("retained receipt").proposal(),
            id
        );
    }

    #[test]
    fn rejecting_manual_plugin_proposal_drops_it_without_mutation() {
        let mut app = manual_proposal_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));

        enqueue_text_insert(&mut app, buffer, view, 0, "plugin ", TextRevision::from(0));
        app.update();
        let id = app
            .world()
            .resource::<PluginProposalLane>()
            .first_pending_id()
            .expect("proposal id");
        let _message = app
            .world_mut()
            .write_message(PluginBufferEditProposalRejectRequested { proposal: id });

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma");
        assert!(app.world().resource::<PluginProposalLane>().is_empty());

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
        assert_eq!(observed.proposal_receipts.len(), 1);
        let receipt = &observed.proposal_receipts[0];
        assert_eq!(receipt.proposal(), id);
        assert_eq!(receipt.source_identity(), "integration");
        assert_eq!(receipt.target(), buffer);
        assert_eq!(receipt.base_revision(), TextRevision::from(0));
        assert_eq!(
            receipt.outcome(),
            &PluginBufferEditProposalReceiptOutcome::Rejected {
                owner: PluginBufferEditProposalReceiptOwner::ProposalLane,
                reason: PluginBufferEditProposalReceiptRejectionReason::RejectedBeforeOwner,
            }
        );
        let log = app.world().resource::<PluginProposalReceiptLog>();
        assert_eq!(log.len(), 1);
        assert_eq!(
            log.receipts().next().expect("retained receipt").proposal(),
            id
        );
    }

    #[test]
    fn stale_manual_plugin_proposal_decisions_emit_typed_rejections() {
        let mut app = manual_proposal_test_app();
        let missing = PluginProposalId::from_nonzero(NonZeroU64::MIN);
        let _message = app
            .world_mut()
            .write_message(PluginBufferEditProposalApplyRequested { proposal: missing });
        let _message = app
            .world_mut()
            .write_message(PluginBufferEditProposalRejectRequested { proposal: missing });

        app.update();

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert_eq!(
            observed.proposal_decision_rejected,
            vec![
                PluginBufferEditProposalDecisionRejected {
                    proposal: missing,
                    decision: PluginBufferEditProposalDecision::Apply,
                    reason: PluginProposalDecisionRejectionReason::UnknownProposal { id: missing },
                },
                PluginBufferEditProposalDecisionRejected {
                    proposal: missing,
                    decision: PluginBufferEditProposalDecision::Reject,
                    reason: PluginProposalDecisionRejectionReason::UnknownProposal { id: missing },
                },
            ]
        );
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
        assert!(observed.proposal_receipts.is_empty());
        assert!(
            app.world()
                .resource::<PluginProposalReceiptLog>()
                .is_empty()
        );
    }

    #[test]
    fn manual_proposal_lane_saturation_emits_redacted_operational_event() {
        let mut app = manual_proposal_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));
        app.world_mut()
            .insert_resource(PluginProposalLane::with_limit(
                crate::ecs::plugin_proposal::PluginProposalLaneLimit::try_new(1)
                    .expect("limit should be valid"),
            ));

        enqueue_text_insert(
            &mut app,
            buffer,
            view,
            0,
            "first secret",
            TextRevision::from(0),
        );
        enqueue_text_insert(
            &mut app,
            buffer,
            view,
            0,
            "second secret",
            TextRevision::from(0),
        );
        app.update();
        app.update();

        let proposals = app.world().resource::<PluginProposalLane>();
        assert_eq!(proposals.len(), 1);

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
        assert_eq!(observed.operational_events.len(), 1);
        let event = &observed.operational_events[0].event;
        assert_eq!(event.identity(), Some("integration"));
        assert_eq!(
            event.kind(),
            PluginOperationalEventKind::QueueSaturated {
                queue: PluginOperationalQueue::ProposalLane,
                limit: count_limit(1),
            }
        );
        assert!(!format!("{event:?}").contains("secret"));
        assert!(!event.to_string().contains("secret"));
    }

    #[test]
    fn plugin_intent_queue_saturation_keeps_retryable_batch() {
        let mut app = plugin_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));
        app.world_mut()
            .insert_resource(PluginIntentQueue::with_limit(
                PluginIntentQueueLimit::try_new(1).expect("limit should be valid"),
            ));

        enqueue_text_insert(&mut app, buffer, view, 0, "a", TextRevision::from(0));
        let second =
            sealed_text_insert_batch(buffer, view, 1, "secret payload", TextRevision::from(1));
        let error = app
            .world_mut()
            .resource_mut::<PluginIntentQueue>()
            .push_sealed(second)
            .expect_err("second sealed batch should exceed the queue limit");

        assert_eq!(
            error.queue_error(),
            &PluginIntentQueueError::TooManyBatches {
                identity: PluginIdentity::try_new("integration").expect("identity"),
                limit: count_limit(1),
            }
        );
        assert_eq!(error.batch().identity(), "integration");
        assert_eq!(error.batch().effect_count(), 2);
        let event = error
            .queue_error()
            .operational_event()
            .expect("saturation event");
        assert_eq!(event.identity(), Some("integration"));
        assert_eq!(
            event.kind(),
            PluginOperationalEventKind::QueueSaturated {
                queue: PluginOperationalQueue::Intent,
                limit: count_limit(1),
            }
        );
        let debug = format!("{error:?}");
        assert!(debug.contains("PluginIntentQueueEnqueueError"));
        assert!(!debug.contains("secret payload"));
        assert!(!event.to_string().contains("secret"));
        let parts = error.into_parts();
        assert_eq!(
            parts.queue_error,
            PluginIntentQueueError::TooManyBatches {
                identity: PluginIdentity::try_new("integration").expect("identity"),
                limit: count_limit(1),
            }
        );
        let retry = parts.batch;

        assert!(
            app.world()
                .resource::<ObservedPluginOwners>()
                .edited
                .is_empty()
        );

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "aalma");
        assert!(app.world().resource::<PluginIntentQueue>().is_empty());

        app.world_mut()
            .resource_mut::<PluginIntentQueue>()
            .push_sealed(retry)
            .expect("retry batch should enqueue after capacity frees");

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "asecret payloadalma");

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert_eq!(observed.edited.len(), 2);
        assert!(observed.edit_rejected.is_empty());
        assert!(app.world().resource::<PluginIntentQueue>().is_empty());
    }

    #[test]
    fn plugin_intent_queue_enqueue_error_can_discard_retry_batch() {
        let buffer = BufferEntity(Entity::from_raw_u32(11).expect("test entity"));
        let view = ViewEntity(Entity::from_raw_u32(12).expect("test entity"));
        let mut queue = PluginIntentQueue::with_limit(
            PluginIntentQueueLimit::try_new(1).expect("limit should validate"),
        );
        queue
            .push_sealed(sealed_text_insert_batch(
                buffer,
                view,
                0,
                "kept",
                TextRevision::from(0),
            ))
            .expect("first batch should fit");

        let error = queue
            .push_sealed(sealed_text_insert_batch(
                buffer,
                view,
                0,
                "secret discarded intent",
                TextRevision::from(0),
            ))
            .expect_err("full queue should keep the batch retryable");
        let debug = format!("{error:?}");

        assert!(debug.contains("PluginIntentQueueEnqueueError"));
        assert!(!debug.contains("secret discarded intent"));
        assert_eq!(queue.len(), 1);

        let report = error.discard_batch();

        assert_eq!(report.identity(), "integration");
        assert_eq!(report.discarded_effects(), 2);
        assert_eq!(queue.len(), 1);
    }

    #[test]
    fn plugin_intent_queue_ignores_empty_sealed_batches() {
        let buffer = BufferEntity(Entity::from_raw_u32(13).expect("test entity"));
        let view = ViewEntity(Entity::from_raw_u32(14).expect("test entity"));
        let mut queue = PluginIntentQueue::with_limit(
            PluginIntentQueueLimit::try_new(1).expect("limit should validate"),
        );

        let empty = empty_effect_batch("integration");
        assert!(empty.is_empty());
        queue
            .push_sealed(empty)
            .expect("empty batch should be accepted as a no-op");
        assert_eq!(queue.len(), 0);

        queue
            .push_sealed(sealed_text_insert_batch(
                buffer,
                view,
                0,
                "kept",
                TextRevision::from(0),
            ))
            .expect("non-empty batch should fit");
        queue
            .push_sealed(empty_effect_batch("integration"))
            .expect("empty batch should not saturate a full queue");
        assert_eq!(queue.len(), 1);

        let error = queue
            .push_sealed(sealed_text_insert_batch(
                buffer,
                view,
                0,
                "secret overflow",
                TextRevision::from(0),
            ))
            .expect_err("non-empty batch should still observe queue capacity");
        assert_eq!(error.batch().effect_count(), 2);
        assert!(!format!("{error:?}").contains("secret overflow"));
    }

    #[test]
    fn plugin_owner_queues_publish_paired_owner_batches() {
        let view = ViewEntity(Entity::from_raw_u32(15).expect("test entity"));
        let mut intent_queue = PluginIntentQueue::with_limit(
            PluginIntentQueueLimit::try_new(1).expect("intent limit should validate"),
        );
        let mut workspace_queue = PluginWorkspaceIoWorkerQueue::with_limit(
            PluginWorkspaceIoWorkerQueueLimit::try_new(1)
                .expect("workspace queue limit should validate"),
        );

        let publication = PluginOwnerQueues::new(&mut intent_queue, &mut workspace_queue)
            .push_owner_batches(owner_batches_with_status_and_workspace_write(
                view,
                "owner status",
                "docs/owner-output.txt",
                b"owner bytes",
            ))
            .expect("paired owner queues should accept work");

        assert_eq!(publication.identity(), "integration");
        assert_eq!(publication.effect_count(), 1);
        assert_eq!(publication.workspace_io_request_count(), 1);
        let debug = format!("{publication:?}");
        assert!(debug.contains("effect_count: 1"));
        assert!(debug.contains("workspace_io_request_count: 1"));
        assert!(!debug.contains("owner status"));
        assert!(!debug.contains("owner-output"));
        assert!(!debug.contains("owner bytes"));
        assert_eq!(intent_queue.len(), 1);
        assert_eq!(workspace_queue.len(), 1);
    }

    #[test]
    fn plugin_owner_queue_intent_rejection_leaves_workspace_io_unpublished() {
        let view = ViewEntity(Entity::from_raw_u32(16).expect("test entity"));
        let mut intent_queue = PluginIntentQueue::with_limit(
            PluginIntentQueueLimit::try_new(1).expect("intent limit should validate"),
        );
        let mut workspace_queue = PluginWorkspaceIoWorkerQueue::with_limit(
            PluginWorkspaceIoWorkerQueueLimit::try_new(2)
                .expect("workspace queue limit should validate"),
        );
        intent_queue
            .push_sealed(sealed_status_batch(view, "queued status"))
            .expect("first intent batch should fit");

        let error = PluginOwnerQueues::new(&mut intent_queue, &mut workspace_queue)
            .push_owner_batches(owner_batches_with_status_and_workspace_write(
                view,
                "secret intent overflow",
                "docs/intent-overflow.txt",
                b"secret intent workspace bytes",
            ))
            .expect_err("full intent queue should reject paired owner work");

        assert_eq!(
            error.queue_error().kind(),
            PluginOwnerQueueErrorKind::Intent
        );
        assert_eq!(intent_queue.len(), 1);
        assert_eq!(workspace_queue.len(), 0);
        assert_eq!(error.owner_batches().identity(), "integration");
        assert!(!error.owner_batches().is_empty());
        let event = error.operational_event().expect("intent saturation event");
        assert_eq!(event.identity(), Some("integration"));
        assert_eq!(
            event.kind(),
            PluginOperationalEventKind::QueueSaturated {
                queue: PluginOperationalQueue::Intent,
                limit: count_limit(1),
            }
        );
        let debug = format!("{error:?}");
        let display = error.to_string();
        assert!(debug.contains("effect_count: 1"));
        assert!(debug.contains("workspace_io_request_count: 1"));
        assert!(!debug.contains("PluginHostOwnerBatches"));
        assert!(!debug.contains("secret intent overflow"));
        assert!(!debug.contains("intent-overflow"));
        assert!(!debug.contains("secret intent workspace bytes"));
        assert!(!display.contains("secret intent overflow"));
        assert!(!display.contains("intent-overflow"));
        assert!(!display.contains("secret intent workspace bytes"));

        let discard = error.discard_owner_batches();

        assert_eq!(discard.identity(), "integration");
        assert_eq!(discard.effects().discarded_effects(), 1);
        assert_eq!(discard.workspace_io().discarded_requests(), 1);
    }

    #[test]
    fn plugin_owner_queue_workspace_rejection_leaves_intent_unpublished() {
        let view = ViewEntity(Entity::from_raw_u32(17).expect("test entity"));
        let mut intent_queue = PluginIntentQueue::with_limit(
            PluginIntentQueueLimit::try_new(2).expect("intent limit should validate"),
        );
        let mut workspace_queue = PluginWorkspaceIoWorkerQueue::with_limit(
            PluginWorkspaceIoWorkerQueueLimit::try_new(1)
                .expect("workspace queue limit should validate"),
        );
        let _publication = PluginOwnerQueues::new(&mut intent_queue, &mut workspace_queue)
            .push_owner_batches(owner_batches_with_status_and_workspace_write(
                view,
                "queued status",
                "docs/queued-output.txt",
                b"queued bytes",
            ))
            .expect("first owner batch should fit");

        let error = PluginOwnerQueues::new(&mut intent_queue, &mut workspace_queue)
            .push_owner_batches(owner_batches_with_status_and_workspace_write(
                view,
                "secret workspace overflow",
                "docs/workspace-overflow.txt",
                b"secret workspace bytes",
            ))
            .expect_err("full workspace queue should reject paired owner work");

        assert_eq!(
            error.queue_error().kind(),
            PluginOwnerQueueErrorKind::WorkspaceIo
        );
        assert_eq!(intent_queue.len(), 1);
        assert_eq!(workspace_queue.len(), 1);
        assert_eq!(error.owner_batches().effects().effect_count(), 1);
        assert_eq!(error.owner_batches().workspace_io().request_count(), 1);
        let event = error
            .operational_event()
            .expect("workspace saturation event");
        assert_eq!(event.identity(), Some("integration"));
        assert_eq!(
            event.kind(),
            PluginOperationalEventKind::QueueSaturated {
                queue: PluginOperationalQueue::WorkspaceIo,
                limit: count_limit(1),
            }
        );
        let debug = format!("{error:?}");
        let display = error.to_string();
        assert!(debug.contains("effect_count: 1"));
        assert!(debug.contains("workspace_io_request_count: 1"));
        assert!(!debug.contains("PluginHostOwnerBatches"));
        assert!(!debug.contains("secret workspace overflow"));
        assert!(!debug.contains("workspace-overflow"));
        assert!(!debug.contains("secret workspace bytes"));
        assert!(!display.contains("secret workspace overflow"));
        assert!(!display.contains("workspace-overflow"));
        assert!(!display.contains("secret workspace bytes"));
        let parts = error.into_parts();

        assert_eq!(
            parts.queue_error.kind(),
            PluginOwnerQueueErrorKind::WorkspaceIo
        );
        assert_eq!(parts.owner_batches.identity(), "integration");
        assert_eq!(parts.owner_batches.effects().effect_count(), 1);
        assert_eq!(parts.owner_batches.workspace_io().request_count(), 1);
    }

    #[test]
    fn denied_guest_import_never_reaches_plugin_intent_queue() {
        let mut app = plugin_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let host = PluginHostContext::for_test("denied", PluginCapabilities::default());
        let mut handles =
            PluginHandleStore::new(PluginIdentity::try_new("denied").expect("identity"));
        let handle = handles
            .issue_buffer(buffer, Some(TextRevision::from(0)), None)
            .expect("buffer handle");
        let mut session = host_import_session(&host, &handles);

        let error = session
            .propose_buffer_edit(handle, BufferEdit::insert(0, "plugin"))
            .expect_err("capability denial should reject before publication");
        let discard = session.discard();

        assert_eq!(
            error,
            PluginHostImportError::Authorization(PluginAuthorizationError::Denied {
                identity: PluginIdentity::try_new("denied").expect("identity"),
                capability: PluginCapabilityShape::BufferProposeEdit,
            })
        );
        assert_eq!(discard.effects().discarded_effects(), 0);
        assert!(app.world().resource::<PluginIntentQueue>().is_empty());

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma");
        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
    }

    #[test]
    fn stale_handle_import_never_reaches_plugin_intent_queue() {
        let mut app = plugin_test_app();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let host = PluginHostContext::for_test(
            "stale-handle",
            PluginCapabilities {
                buffer_propose_edit: true,
                ..PluginCapabilities::default()
            },
        );
        let mut handles =
            PluginHandleStore::new(PluginIdentity::try_new("stale-handle").expect("identity"));
        let handle = handles
            .issue_buffer(buffer, Some(TextRevision::from(0)), None)
            .expect("buffer handle");

        handles.revoke_all();
        let current = handles.generation();
        let mut session = host_import_session(&host, &handles);

        let error = session
            .propose_buffer_edit(handle, BufferEdit::insert(0, "plugin"))
            .expect_err("stale handle should reject before effect creation");
        let discard = session.discard();

        assert_eq!(
            error,
            PluginHostImportError::Handle(PluginHandleError::StaleGeneration {
                handle: handle.shape(),
                current,
            })
        );
        assert_eq!(discard.effects().discarded_effects(), 0);
        assert!(app.world().resource::<PluginIntentQueue>().is_empty());

        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma");
        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
    }

    #[test]
    fn plugin_intent_queue_uses_validated_runtime_config() {
        let config = AppConfig::from_json_slice(
            br#"{
                "plugins": {
                    "plugins": [
                        {
                            "identity": "formatter",
                            "component_path": "plugins/formatter.wasm",
                            "enabled": true,
                            "runtime_limits": {
                                "max_intent_batches": 3
                            }
                        }
                    ]
                }
            }"#,
            Path::new("test-config.json"),
        )
        .expect("config should validate");
        let mut app = App::new();
        let _app = app.insert_resource(config).add_plugins(EditorCorePlugin);

        assert_eq!(
            app.world()
                .resource::<PluginIntentQueue>()
                .limit()
                .max_batches(),
            3
        );
    }

    #[test]
    fn plugin_intent_queue_uses_registry_scheduler_policy() {
        let config = AppConfig::from_json_slice(
            br#"{
                "plugins": {
                    "plugins": [
                        {
                            "identity": "formatter",
                            "component_path": "plugins/formatter.wasm",
                            "enabled": true,
                            "runtime_limits": {
                                "max_intent_batches": 2
                            }
                        },
                        {
                            "identity": "linter",
                            "component_path": "plugins/linter.wasm",
                            "enabled": true,
                            "runtime_limits": {
                                "max_intent_batches": 3
                            }
                        }
                    ]
                }
            }"#,
            Path::new("test-config.json"),
        )
        .expect("config should validate");
        let mut app = App::new();
        let _app = app.insert_resource(config).add_plugins(EditorCorePlugin);

        assert_eq!(
            app.world()
                .resource::<PluginIntentQueue>()
                .limit()
                .max_batches(),
            5
        );
    }

    #[test]
    fn plugin_proposal_policy_uses_validated_runtime_config() {
        let config = AppConfig::from_json_slice(
            br#"{
                "plugins": {
                    "plugins": [
                        {
                            "identity": "integration",
                            "component_path": "plugins/integration.wasm",
                            "enabled": true,
                            "proposal_policy": {
                                "buffer_edits": "manual_review"
                            }
                        }
                    ]
                }
            }"#,
            Path::new("test-config.json"),
        )
        .expect("config should validate");
        let mut app = App::new();
        let _app = app
            .insert_resource(config)
            .insert_resource(InitialEditorBuffer {
                stream: TextByteStream::new("alma"),
                file: crate::buffer::BufferFile::scratch(0),
            })
            .init_resource::<ObservedPluginOwners>()
            .add_plugins(EditorCorePlugin)
            .add_plugins(BufferPlugin)
            .add_systems(Update, capture_owner_results.in_set(EditorSet::Render));
        app.update();
        let buffer = BufferEntity(editor_buffer_entity(&mut app));
        let view = ViewEntity(editor_view_entity(&mut app));

        enqueue_text_insert(&mut app, buffer, view, 0, "plugin ", TextRevision::from(0));
        app.update();
        app.update();

        let buffer_text = app
            .world()
            .get::<BufferText>(buffer.get())
            .expect("editor buffer should exist");
        assert_eq!(buffer_text.stream.as_str(), "alma");
        assert_eq!(app.world().resource::<PluginProposalLane>().len(), 1);
        assert_eq!(
            app.world()
                .resource::<PluginProposalLanePolicy>()
                .buffer_edit_mode_for(&PluginIdentity::try_new("integration").expect("identity")),
            PluginProposalReviewMode::ManualReview
        );

        let observed = app.world().resource::<ObservedPluginOwners>();
        assert!(observed.edited.is_empty());
        assert!(observed.edit_rejected.is_empty());
    }

    fn plugin_test_app() -> App {
        plugin_test_app_with_policy(PluginProposalLanePolicy::default())
    }

    fn manual_proposal_test_app() -> App {
        plugin_test_app_with_policy(PluginProposalLanePolicy::new(
            PluginProposalReviewMode::ManualReview,
        ))
    }

    fn plugin_test_app_with_policy(policy: PluginProposalLanePolicy) -> App {
        let mut app = App::new();
        let _app = app
            .insert_resource(InitialEditorBuffer {
                stream: TextByteStream::new("alma"),
                file: crate::buffer::BufferFile::scratch(0),
            })
            .insert_resource(policy)
            .init_resource::<ObservedPluginOwners>()
            .add_plugins(EditorCorePlugin)
            .add_plugins(BufferPlugin)
            .add_systems(Update, capture_owner_results.in_set(EditorSet::Render));
        app.update();
        app
    }

    fn enqueue_plugin_effects(app: &mut App, buffer: BufferEntity, view: ViewEntity) {
        enqueue_batch(app, buffer, view);
        assert_eq!(app.world().resource::<PluginIntentQueue>().len(), 1);
    }

    fn enqueue_batch(app: &mut App, buffer: BufferEntity, view: ViewEntity) {
        enqueue_text_insert(app, buffer, view, 0, "plugin", TextRevision::from(0));
    }

    fn enqueue_text_insert(
        app: &mut App,
        buffer: BufferEntity,
        view: ViewEntity,
        byte_index: usize,
        text: &str,
        base_revision: TextRevision,
    ) {
        let batch = sealed_text_insert_batch(buffer, view, byte_index, text, base_revision);

        app.world_mut()
            .resource_mut::<PluginIntentQueue>()
            .push_sealed(batch)
            .expect("intent queue should accept batch");
    }

    fn sealed_text_insert_batch(
        buffer: BufferEntity,
        view: ViewEntity,
        byte_index: usize,
        text: &str,
        base_revision: TextRevision,
    ) -> SealedPluginEffectBatch {
        let mut pending = PendingPluginUpdateBatch::new(
            PluginIdentity::try_new("integration").expect("test identity should validate"),
            PluginEffectBatchLimit::default(),
        );
        pending
            .propose_buffer_edit(PluginBufferEditProposal::from_observed_revision(
                buffer,
                BufferEdit::insert(byte_index, text),
                base_revision,
            ))
            .expect("buffer edit should fit");
        pending
            .push_status_text(view, "plugin ok")
            .expect("status should fit");
        pending.seal()
    }

    fn sealed_status_batch(view: ViewEntity, text: &str) -> SealedPluginEffectBatch {
        let mut pending = PendingPluginUpdateBatch::new(
            PluginIdentity::try_new("integration").expect("test identity should validate"),
            PluginEffectBatchLimit::default(),
        );
        pending
            .push_status_text(view, text)
            .expect("status should fit");
        pending.seal()
    }

    fn owner_batches_with_status_and_workspace_write(
        view: ViewEntity,
        status: &str,
        path: &str,
        bytes: &[u8],
    ) -> PluginHostOwnerBatches {
        let host = PluginHostContext::for_test(
            "integration",
            PluginCapabilities {
                status_publish: true,
                workspace_artifact_write: vec![WorkspacePathGrant::new("docs")],
                ..PluginCapabilities::default()
            },
        );
        let mut handles =
            PluginHandleStore::new(PluginIdentity::try_new("integration").expect("identity"));
        let view_handle = handles.issue_view(view).expect("view handle");
        let mut session = host_import_session(&host, &handles);

        session
            .push_status_text(view_handle, status)
            .expect("status should queue");
        session
            .queue_workspace_artifact_write(
                WorkspacePathRef::try_from(path).expect("workspace path should validate"),
                bytes.to_vec(),
            )
            .expect("workspace write should queue");
        let scheduled = session.seal().into_owner_and_workspace_observe_tasks();

        assert!(scheduled.workspace_observe_tasks.is_empty());
        scheduled.owner_batches
    }

    fn count_limit(value: usize) -> NonZeroUsize {
        NonZeroUsize::new(value).expect("test count limit should be non-zero")
    }

    fn empty_effect_batch(identity: &str) -> SealedPluginEffectBatch {
        PendingPluginUpdateBatch::new(
            PluginIdentity::try_new(identity).expect("test identity should validate"),
            PluginEffectBatchLimit::default(),
        )
        .seal()
    }

    fn host_import_session(
        host: &PluginHostContext,
        handles: &PluginHandleStore,
    ) -> crate::plugin::PluginHostImportSession {
        crate::plugin::PluginHostImportSession::new(
            host,
            handles,
            PluginEffectBatchLimit::default(),
            crate::plugin::PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget"),
        )
        .expect("session")
    }

    fn editor_buffer_entity(app: &mut App) -> Entity {
        let mut query = app
            .world_mut()
            .query_filtered::<Entity, With<EditorBuffer>>();
        query
            .iter(app.world())
            .next()
            .expect("editor buffer should exist")
    }

    fn editor_view_entity(app: &mut App) -> Entity {
        let mut query = app.world_mut().query_filtered::<Entity, With<EditorView>>();
        query
            .iter(app.world())
            .next()
            .expect("editor view should exist")
    }

    fn capture_owner_results(
        mut edited: MessageReader<BufferEdited>,
        mut edit_rejected: MessageReader<BufferEditRejected>,
        mut proposal_decision_rejected: MessageReader<PluginBufferEditProposalDecisionRejected>,
        mut proposal_receipts: MessageReader<PluginBufferEditProposalReceipt>,
        mut status_rejected: MessageReader<StatusMessageRejected>,
        mut operational_events: MessageReader<PluginOperationalEventReported>,
        mut observed: bevy::prelude::ResMut<ObservedPluginOwners>,
    ) {
        observed.edited.extend(edited.read().copied());
        observed.edit_rejected.extend(edit_rejected.read().cloned());
        observed
            .proposal_decision_rejected
            .extend(proposal_decision_rejected.read().copied());
        observed
            .proposal_receipts
            .extend(proposal_receipts.read().cloned());
        observed
            .status_rejected
            .extend(status_rejected.read().cloned());
        observed
            .operational_events
            .extend(operational_events.read().cloned());
    }

    impl ObservedPluginOwners {
        fn clear(&mut self) {
            self.edited.clear();
            self.edit_rejected.clear();
            self.proposal_decision_rejected.clear();
            self.proposal_receipts.clear();
            self.status_rejected.clear();
            self.operational_events.clear();
        }
    }
}