alma 0.1.1

use super::{
    PendingPluginWorkspaceObserveTaskBatch, PluginWorkspaceReadByteLimit,
    PluginWorkspaceWriteByteLimit, PluginWorkspaceWritePayload,
};
use crate::fs_utils::FilesystemConfig;
use crate::plugin::{
    PendingPluginWorkspaceIoBatch, PluginAuthorizationError, PluginCapabilities,
    PluginCapabilityRef, PluginCapabilityShape, PluginHostContext, PluginIdentity,
    PluginOperationalImportRejection, PluginOperationalQueue, PluginWorkspaceIoBudget,
    PluginWorkspaceIoBudgetField, PluginWorkspaceIoCommitError, PluginWorkspaceIoCommitErrorKind,
    PluginWorkspaceIoExecutionModel, PluginWorkspaceIoIdentityMismatchSource,
    PluginWorkspaceIoWorkerEnqueueError, PluginWorkspaceIoWorkerEnqueueErrorParts,
    PluginWorkspaceIoWorkerQueue, PluginWorkspaceIoWorkerQueueError,
    PluginWorkspaceIoWorkerQueueErrorKind, PluginWorkspaceIoWorkerQueueLimit,
    PluginWorkspaceIoWorkerQueueLimitField, PluginWorkspaceObserveOutcome,
    PluginWorkspaceObserveOutcomeShape, PluginWorkspaceObserveProjectionError,
    PluginWorkspaceObserveTaskCancellationReport, PluginWorkspaceObserveTaskCompletionShape,
    PluginWorkspaceObserveTaskDiscardReport, PluginWorkspaceObserveTaskEnqueueError,
    PluginWorkspaceObserveTaskEnqueueErrorParts, PluginWorkspaceObserveTaskEnqueueReport,
    PluginWorkspaceObserveTaskHandle, PluginWorkspaceObserveTaskId, PluginWorkspaceObserveTaskPoll,
    PluginWorkspaceObserveTaskQueue, PluginWorkspaceObserveTaskQueueError,
    PluginWorkspaceObserveTaskQueueErrorKind, PluginWorkspaceObserveTaskQueueLimit,
    PluginWorkspaceObserveTaskQueueLimitField, PluginWorkspaceObserveTaskReservationMismatch,
    PluginWorkspaceObserveTaskReserveError, PluginWorkspaceObserveTaskResultShape,
    PluginWorkspaceObserveTaskState, PluginWorkspaceObserveTaskTake, PluginWorkspaceWriteError,
    SealedPluginWorkspaceIoBatch, SealedPluginWorkspaceObserveTaskBatch, WorkspaceAccessKind,
    WorkspacePathGrant, WorkspacePathRef,
};
use proptest::prelude::*;
use std::{
    fs,
    num::{NonZeroU64, NonZeroUsize},
    path::{Path, PathBuf},
    time::{SystemTime, UNIX_EPOCH},
};

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

fn task_handle(
    identity: &PluginIdentity,
    task_id: PluginWorkspaceObserveTaskId,
) -> PluginWorkspaceObserveTaskHandle {
    PluginWorkspaceObserveTaskHandle::new(identity.clone(), task_id)
}

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

fn assert_count_shaped_observe_task_queue_debug(debug: &str, root: &Path) {
    for expected in [
        "PluginWorkspaceObserveTaskQueue",
        "pending_len",
        "retained_len",
    ] {
        assert!(debug.contains(expected));
    }
    for redacted in [
        "pending_task_shapes",
        "retained_tasks",
        "task_id",
        "byte_len",
        "secret task bytes",
        "input.txt",
    ] {
        assert!(!debug.contains(redacted));
    }
    assert!(!debug.contains(root.to_string_lossy().as_ref()));
}

#[test]
fn workspace_observe_task_public_types_are_send_sync() {
    assert_send_sync::<PendingPluginWorkspaceObserveTaskBatch>();
    assert_send_sync::<PluginWorkspaceObserveTaskQueue>();
    assert_send_sync::<PluginWorkspaceObserveTaskHandle>();
    assert_send_sync::<SealedPluginWorkspaceObserveTaskBatch>();
    assert_send_sync::<PluginWorkspaceObserveTaskDiscardReport>();
    assert_send_sync::<PluginWorkspaceObserveTaskEnqueueError>();
    assert_send_sync::<PluginWorkspaceObserveTaskEnqueueErrorParts>();
    assert_send_sync::<PluginWorkspaceObserveTaskEnqueueReport>();
    assert_send_sync::<PluginWorkspaceObserveTaskCancellationReport>();
    assert_send_sync::<PluginWorkspaceObserveTaskCompletionShape>();
    assert_send_sync::<PluginWorkspaceObserveTaskPoll>();
    assert_send_sync::<PluginWorkspaceObserveTaskResultShape>();
    assert_send_sync::<PluginWorkspaceObserveTaskState>();
    assert_send_sync::<PluginWorkspaceObserveTaskTake>();
}

#[test]
fn workspace_worker_public_types_are_send_sync() {
    assert_send_sync::<PluginWorkspaceIoWorkerEnqueueError>();
    assert_send_sync::<PluginWorkspaceIoWorkerEnqueueErrorParts>();
}

#[test]
fn workspace_io_budget_uses_validated_message_byte_proof() {
    let max_message_bytes = NonZeroU64::new(96).expect("test message limit should be non-zero");

    let budget = PluginWorkspaceIoBudget::from_max_message_byte_limit(max_message_bytes);

    assert_eq!(budget.max_requests(), 32);
    assert_eq!(budget.max_read_bytes(), 96);
    assert_eq!(budget.max_write_bytes(), 96);
    assert_eq!(
        PluginWorkspaceIoBudget::from_max_message_bytes(0),
        Err(PluginWorkspaceIoCommitError::ZeroLimit {
            field: PluginWorkspaceIoBudgetField::MaxReadBytes,
        })
    );
}

#[test]
fn workspace_observe_task_reservation_errors_project_source_rejections() {
    let identity = plugin_identity("workspace");
    let budget_error = PluginWorkspaceObserveTaskReserveError::WorkspaceIo(
        PluginWorkspaceIoCommitError::TooManyRequests {
            identity: identity.clone(),
            limit: count_limit(1),
        },
    );
    let queue_error = PluginWorkspaceObserveTaskReserveError::Queue(
        PluginWorkspaceObserveTaskQueueError::TooManyPending {
            identity,
            limit: count_limit(1),
        },
    );

    assert_eq!(
        budget_error.operational_rejection(),
        PluginOperationalImportRejection::WorkspaceIoQueue
    );
    assert_eq!(
        queue_error.operational_rejection(),
        PluginOperationalImportRejection::WorkspaceObserveTaskQueue
    );
}

#[test]
fn workspace_observe_task_handle_borrows_internal_lookup() {
    let identity = plugin_identity("workspace");
    let task_id = PluginWorkspaceObserveTaskId::try_new(7).expect("task id should validate");
    let handle = PluginWorkspaceObserveTaskHandle::new(identity.clone(), task_id);

    let lookup = handle.lookup();

    assert_eq!(handle.identity(), "workspace");
    assert_eq!(handle.identity_proof(), &identity);
    assert_eq!(handle.task_id(), task_id);
    assert_eq!(handle.lookup(), lookup);
    assert_eq!(format!("{handle:?}"), "identity=\"workspace\" task_id=7");
    assert!(lookup.matches_handle(&handle));
    assert_eq!(
        format!("{lookup:?}"),
        "PluginWorkspaceObserveTaskLookup { identity: PluginIdentity(\"workspace\"), task_id: PluginWorkspaceObserveTaskId(7) }"
    );
    assert!(lookup.matches_task(&identity, task_id));
    assert!(!lookup.matches_task(
        &PluginIdentity::try_new("other").expect("identity"),
        task_id
    ));
    assert!(!lookup.matches_task(
        &identity,
        PluginWorkspaceObserveTaskId::try_new(8).expect("task id should validate")
    ));
}

#[test]
fn workspace_observe_task_poll_states_are_closed() {
    let pending_poll = PluginWorkspaceObserveTaskPoll::Pending;
    let completed_poll = PluginWorkspaceObserveTaskPoll::Completed {
        outcome: PluginWorkspaceObserveOutcomeShape::Rejected,
    };
    let unknown_poll = PluginWorkspaceObserveTaskPoll::Unknown;

    assert_eq!(
        pending_poll.state(),
        PluginWorkspaceObserveTaskState::Pending
    );
    assert_eq!(
        pending_poll.shape(),
        PluginWorkspaceObserveTaskResultShape::Pending
    );
    assert_eq!(pending_poll.shape().state(), pending_poll.state());
    assert_eq!(pending_poll.shape().outcome_shape(), None);
    assert!(PluginWorkspaceObserveTaskState::Pending.is_pending());
    assert!(!PluginWorkspaceObserveTaskState::Pending.is_completed());
    assert!(!PluginWorkspaceObserveTaskState::Pending.is_unknown());
    assert!(pending_poll.is_pending());
    assert!(!pending_poll.is_completed());
    assert!(!pending_poll.is_unknown());
    assert_eq!(pending_poll.outcome_shape(), None);
    assert_eq!(
        completed_poll.state(),
        PluginWorkspaceObserveTaskState::Completed
    );
    assert_eq!(
        completed_poll.shape(),
        PluginWorkspaceObserveTaskResultShape::Completed {
            outcome: PluginWorkspaceObserveOutcomeShape::Rejected,
        }
    );
    assert_eq!(completed_poll.shape().state(), completed_poll.state());
    assert_eq!(
        completed_poll.shape().outcome_shape(),
        Some(PluginWorkspaceObserveOutcomeShape::Rejected)
    );
    assert!(!PluginWorkspaceObserveTaskState::Completed.is_pending());
    assert!(PluginWorkspaceObserveTaskState::Completed.is_completed());
    assert!(!PluginWorkspaceObserveTaskState::Completed.is_unknown());
    assert!(!completed_poll.is_pending());
    assert!(completed_poll.is_completed());
    assert!(!completed_poll.is_unknown());
    assert_eq!(
        completed_poll.outcome_shape(),
        Some(PluginWorkspaceObserveOutcomeShape::Rejected)
    );
    assert_eq!(
        unknown_poll.state(),
        PluginWorkspaceObserveTaskState::Unknown
    );
    assert_eq!(
        unknown_poll.shape(),
        PluginWorkspaceObserveTaskResultShape::Unknown
    );
    assert_eq!(unknown_poll.shape().state(), unknown_poll.state());
    assert_eq!(unknown_poll.shape().outcome_shape(), None);
    assert!(!PluginWorkspaceObserveTaskState::Unknown.is_pending());
    assert!(!PluginWorkspaceObserveTaskState::Unknown.is_completed());
    assert!(PluginWorkspaceObserveTaskState::Unknown.is_unknown());
    assert!(unknown_poll.is_unknown());
    assert_eq!(unknown_poll.outcome_shape(), None);
}

#[test]
fn workspace_observe_task_take_states_are_closed() {
    let outcome = PluginWorkspaceObserveOutcome::Rejected;
    let pending_take = PluginWorkspaceObserveTaskTake::Pending;
    let completed_take = PluginWorkspaceObserveTaskTake::Completed(outcome.clone());
    let unknown_take = PluginWorkspaceObserveTaskTake::Unknown;

    assert_eq!(
        pending_take.state(),
        PluginWorkspaceObserveTaskState::Pending
    );
    assert_eq!(
        pending_take.shape(),
        PluginWorkspaceObserveTaskResultShape::Pending
    );
    assert!(pending_take.is_pending());
    assert!(!pending_take.is_completed());
    assert!(!pending_take.is_unknown());
    assert_eq!(
        completed_take.state(),
        PluginWorkspaceObserveTaskState::Completed
    );
    assert_eq!(
        completed_take.shape(),
        PluginWorkspaceObserveTaskResultShape::Completed {
            outcome: PluginWorkspaceObserveOutcomeShape::Rejected,
        }
    );
    assert!(!completed_take.is_pending());
    assert!(completed_take.is_completed());
    assert!(!completed_take.is_unknown());
    assert_eq!(completed_take.outcome(), Some(&outcome));
    assert_eq!(
        unknown_take.state(),
        PluginWorkspaceObserveTaskState::Unknown
    );
    assert_eq!(
        unknown_take.shape(),
        PluginWorkspaceObserveTaskResultShape::Unknown
    );
    assert!(unknown_take.is_unknown());
    assert_eq!(unknown_take.outcome(), None);
}

#[test]
fn workspace_observe_task_state_rendering_is_stable() {
    let outcome = PluginWorkspaceObserveOutcome::Rejected;
    let completed_poll = PluginWorkspaceObserveTaskPoll::Completed {
        outcome: PluginWorkspaceObserveOutcomeShape::Rejected,
    };

    assert_eq!(PluginWorkspaceObserveTaskState::Pending.as_str(), "pending");
    assert_eq!(
        PluginWorkspaceObserveTaskState::Completed.to_string(),
        "completed"
    );
    assert_eq!(
        format!("{:?}", PluginWorkspaceObserveTaskState::Unknown),
        "unknown"
    );
    assert_eq!(
        completed_poll.shape().to_string(),
        "completed outcome=rejected"
    );
    assert_eq!(completed_poll.to_string(), "completed outcome=rejected");
    assert_eq!(
        PluginWorkspaceObserveTaskPoll::Pending.to_string(),
        "pending"
    );
    assert_eq!(
        PluginWorkspaceObserveTaskTake::Completed(outcome).to_string(),
        "completed outcome=rejected"
    );
}

#[test]
fn authorizes_only_effective_capabilities() {
    let host = PluginHostContext::for_test(
        "status",
        PluginCapabilities {
            status_publish: true,
            workspace_observe: vec![WorkspacePathGrant::new("docs")],
            ..PluginCapabilities::default()
        },
    );

    assert_eq!(host.authorize(PluginCapabilityRef::StatusPublish), Ok(()));
    assert_eq!(
        host.authorize(
            PluginCapabilityRef::workspace_observe("docs/arch.md").expect("path should be valid")
        ),
        Ok(())
    );
    assert_eq!(
        host.authorize(
            PluginCapabilityRef::workspace_artifact_write("docs/arch.md")
                .expect("path should be valid")
        ),
        Err(PluginAuthorizationError::Denied {
            identity: plugin_identity("status"),
            capability: PluginCapabilityShape::WorkspaceArtifactWrite,
        })
    );
}

#[test]
fn workspace_authorization_returns_policy_token_not_filesystem_path() {
    let host = PluginHostContext::for_test(
        "workspace",
        PluginCapabilities {
            workspace_observe: vec![WorkspacePathGrant::new("docs")],
            workspace_artifact_write: vec![WorkspacePathGrant::new("docs/generated")],
            ..PluginCapabilities::default()
        },
    );

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/arch.md").expect("path should be valid"),
        )
        .expect("read should be authorized");
    assert_eq!(read.kind(), WorkspaceAccessKind::Read);
    assert_eq!(read.identity(), "workspace");
    assert_eq!(read.identity_proof().as_str(), "workspace");
    assert_eq!(read.workspace_path().as_str(), "docs/arch.md");
    assert_eq!(read.workspace_relative_path(), "docs/arch.md");

    assert!(matches!(
        host.authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/arch.md").expect("path should be valid")
        ),
        Err(PluginAuthorizationError::Denied {
            capability: PluginCapabilityShape::WorkspaceArtifactWrite,
            ..
        })
    ));
}

#[test]
fn workspace_access_debug_redacts_paths() {
    let host = workspace_host();
    let read_path = "docs/secret-read-debug.txt";
    let write_path = "docs/secret-write-debug.txt";

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from(read_path).expect("path should validate"),
        )
        .expect("read should be authorized");
    let read_debug = format!("{read:?}");
    let read_token = read.into_policy_token();
    let read_token_debug = format!("{read_token:?}");

    let write = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from(write_path).expect("path should validate"),
        )
        .expect("write should be authorized");
    let write_debug = format!("{write:?}");
    let write_token = write.into_policy_token();
    let write_token_debug = format!("{write_token:?}");

    for debug in [
        read_debug.as_str(),
        read_token_debug.as_str(),
        write_debug.as_str(),
        write_token_debug.as_str(),
    ] {
        assert!(debug.contains("workspace"));
        assert!(debug.contains("path_byte_len"));
        assert!(!debug.contains(read_path));
        assert!(!debug.contains(write_path));
        assert!(!debug.contains("secret-read-debug"));
        assert!(!debug.contains("secret-write-debug"));
    }
}

#[test]
fn workspace_access_kind_renders_stable_text() {
    assert_eq!(WorkspaceAccessKind::Read.as_str(), "read");
    assert_eq!(WorkspaceAccessKind::Write.as_str(), "write");
    assert_eq!(WorkspaceAccessKind::Read.to_string(), "read");
    assert_eq!(WorkspaceAccessKind::Write.to_string(), "write");
}

#[test]
fn workspace_io_batch_debug_redacts_bounded_write_payload() {
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );
    let write_path = "docs/secret-batch-write.txt";
    let payload = b"secret batch payload";
    let write = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from(write_path).expect("path should validate"),
        )
        .expect("write should be authorized");
    batch
        .push_write(write, payload.to_vec())
        .expect("write should queue");

    let pending_debug = format!("{batch:?}");
    let sealed_debug = format!("{:?}", batch.seal());

    for debug in [pending_debug.as_str(), sealed_debug.as_str()] {
        assert!(debug.contains("request_shapes"));
        assert!(debug.contains("byte_len"));
        assert!(!debug.contains(write_path));
        assert!(!debug.contains("secret-batch-write"));
        assert!(!debug.contains("secret batch payload"));
    }
}

#[test]
fn workspace_policy_token_routes_reads_and_writes_through_fs_utils() {
    let root = temp_dir("workspace-policy-token");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/input.txt"), "read me").expect("fixture should be written");
    let filesystem = config(&root);
    let host = PluginHostContext::for_test(
        "workspace",
        PluginCapabilities {
            workspace_observe: vec![WorkspacePathGrant::new("docs")],
            workspace_artifact_write: vec![WorkspacePathGrant::new("docs")],
            ..PluginCapabilities::default()
        },
    );

    let read_token = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized")
        .into_policy_token();
    assert_eq!(read_token.workspace_path().as_str(), "docs/input.txt");
    assert_eq!(read_token.workspace_relative_path(), "docs/input.txt");
    let read = read_token
        .read_existing(&filesystem, read_limit(64))
        .expect("policy read should succeed");
    assert_eq!(read.bytes, b"read me");

    let write_token = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/output.txt").expect("path should validate"),
        )
        .expect("write should be authorized")
        .into_policy_token();
    assert_eq!(write_token.workspace_path().as_str(), "docs/output.txt");
    assert_eq!(write_token.workspace_relative_path(), "docs/output.txt");
    let display_path = write_token
        .write_atomic(b"written", &filesystem, write_limit(64))
        .expect("policy write should succeed");
    assert!(!display_path.as_str().contains('\n'));
    assert_eq!(
        fs::read_to_string(root.join("docs/output.txt")).expect("output should be readable"),
        "written"
    );

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_byte_limits_are_non_zero() {
    let read = PluginWorkspaceReadByteLimit::try_new(64).expect("read byte limit should validate");
    let write =
        PluginWorkspaceWriteByteLimit::try_new(32).expect("write byte limit should validate");

    assert_eq!(read.get(), 64);
    assert_eq!(write.get(), 32);
    assert_eq!(
        PluginWorkspaceReadByteLimit::try_new(0),
        Err(PluginWorkspaceIoCommitError::ZeroLimit {
            field: PluginWorkspaceIoBudgetField::MaxReadBytes,
        })
    );
    assert_eq!(
        PluginWorkspaceWriteByteLimit::try_new(0),
        Err(PluginWorkspaceIoCommitError::ZeroLimit {
            field: PluginWorkspaceIoBudgetField::MaxWriteBytes,
        })
    );
    assert_eq!(
        PluginWorkspaceIoBudget::try_new(4, 64, 64)
            .expect("budget should validate")
            .max_read_bytes_limit(),
        read
    );
    assert_eq!(
        PluginWorkspaceIoBudget::try_new(4, 64, 32)
            .expect("budget should validate")
            .max_write_bytes_limit(),
        write
    );
    assert_eq!(
        PluginWorkspaceIoBudgetField::MaxRequests.as_str(),
        "max_requests"
    );
    assert_eq!(
        PluginWorkspaceIoBudgetField::MaxReadBytes.to_string(),
        "max_read_bytes"
    );
    assert_eq!(
        PluginWorkspaceIoBudgetField::MaxWriteBytes.to_string(),
        "max_write_bytes"
    );
    assert_eq!(
        PluginWorkspaceIoIdentityMismatchSource::AccessProof.as_str(),
        "access-proof"
    );
    assert_eq!(
        PluginWorkspaceIoIdentityMismatchSource::RequestLedger.to_string(),
        "request-ledger"
    );
}

#[test]
fn workspace_write_payload_proof_is_bounded_and_redacted() {
    let payload =
        PluginWorkspaceWritePayload::try_new(b"secret payload bytes".to_vec(), write_limit(64))
            .expect("payload should fit");

    assert_eq!(payload.as_bytes(), b"secret payload bytes");
    assert_eq!(payload.len(), 20);
    assert_eq!(payload.max_size(), write_limit(64));

    let debug = format!("{payload:?}");
    assert!(debug.contains("byte_len"));
    assert!(debug.contains("max_size"));
    assert!(!debug.contains("secret payload bytes"));
    assert!(!debug.contains("secret"));

    assert_eq!(
        PluginWorkspaceWritePayload::try_new(b"large".to_vec(), write_limit(3))
            .expect_err("oversized payload should reject"),
        PluginWorkspaceIoCommitError::PayloadTooLarge {
            size: 5,
            max_size: 3,
        }
    );
}

#[test]
fn workspace_io_commit_error_kinds_are_closed() {
    let identity = plugin_identity("workspace");
    let cases = [
        (
            PluginWorkspaceIoCommitError::ZeroLimit {
                field: PluginWorkspaceIoBudgetField::MaxRequests,
            },
            PluginWorkspaceIoCommitErrorKind::ZeroLimit,
            "zero-limit",
        ),
        (
            PluginWorkspaceIoCommitError::PayloadTooLarge {
                size: 9,
                max_size: 8,
            },
            PluginWorkspaceIoCommitErrorKind::PayloadTooLarge,
            "payload-too-large",
        ),
        (
            PluginWorkspaceIoCommitError::TooManyRequests {
                identity: identity.clone(),
                limit: count_limit(1),
            },
            PluginWorkspaceIoCommitErrorKind::TooManyRequests,
            "too-many-requests",
        ),
        (
            PluginWorkspaceIoCommitError::IdentityMismatch {
                mismatch_source: PluginWorkspaceIoIdentityMismatchSource::AccessProof,
                batch_identity: identity,
                provided_identity: plugin_identity("other"),
            },
            PluginWorkspaceIoCommitErrorKind::IdentityMismatch,
            "identity-mismatch",
        ),
    ];

    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 workspace_io_batch_observes_existing_files_through_budgeted_policy() {
    let root = temp_dir("workspace-direct-observe");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/input.txt"), b"secret observe bytes")
        .expect("fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(2, 64, 64).expect("budget should validate"),
    );

    let first = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let missing = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/missing.txt").expect("path should validate"),
        )
        .expect("read should be authorized");

    let outcome = batch
        .observe_existing(first, &filesystem)
        .expect("direct observation should spend one request");
    let rejected = batch
        .observe_existing(missing, &filesystem)
        .expect("filesystem failure should stay guest-domain");

    assert_eq!(outcome.bytes(), Some(b"secret observe bytes".as_slice()));
    assert_eq!(rejected, PluginWorkspaceObserveOutcome::Rejected);
    assert_eq!(batch.request_count(), 0);
    assert_eq!(batch.non_deferred_request_count(), 2);
    assert_eq!(batch.accepted_request_count(), 2);
    let debug = format!("{outcome:?}");
    assert!(debug.contains("byte_len"));
    assert!(!debug.contains("secret observe bytes"));
    assert!(!debug.contains("input.txt"));
    assert!(!debug.contains(root.to_string_lossy().as_ref()));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_io_direct_observe_rejects_cross_plugin_access_before_budget_charge() {
    let root = temp_dir("workspace-direct-observe-identity");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host_for("other");
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized by the other plugin");

    let error = batch
        .observe_existing(access, &filesystem)
        .expect_err("direct observation should reject cross-plugin access");

    assert_eq!(
        error,
        PluginWorkspaceIoCommitError::IdentityMismatch {
            mismatch_source: PluginWorkspaceIoIdentityMismatchSource::AccessProof,
            batch_identity: plugin_identity("workspace"),
            provided_identity: plugin_identity("other"),
        }
    );
    assert_eq!(batch.request_count(), 0);
    assert_eq!(batch.non_deferred_request_count(), 0);
    assert_eq!(batch.accepted_request_count(), 0);

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_policy_token_rejects_oversized_payload() {
    let root = temp_dir("workspace-policy-reject");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = PluginHostContext::for_test(
        "workspace",
        PluginCapabilities {
            workspace_observe: vec![WorkspacePathGrant::new("docs")],
            workspace_artifact_write: vec![WorkspacePathGrant::new("docs")],
            ..PluginCapabilities::default()
        },
    );

    let write_token = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/file.txt").expect("path should validate"),
        )
        .expect("write should be authorized")
        .into_policy_token();
    assert!(matches!(
        write_token.write_atomic(b"too large", &filesystem, write_limit(3)),
        Err(PluginWorkspaceWriteError::PayloadTooLarge {
            size: 9,
            max_size: 3,
        })
    ));
    assert!(!root.join("docs/file.txt").exists());

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_io_batch_defers_filesystem_effects_until_successful_return() {
    let root = temp_dir("workspace-io-deferred-write");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );

    let write = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/output.txt").expect("path should validate"),
        )
        .expect("write should be authorized");
    batch
        .push_write(write, b"deferred".to_vec())
        .expect("write should queue");

    assert_eq!(
        batch.execution_model(),
        PluginWorkspaceIoExecutionModel::DeferredSynchronous
    );
    assert_eq!(batch.identity(), "workspace");
    assert_eq!(batch.identity_proof().as_str(), "workspace");
    assert_eq!(batch.request_count(), 1);
    assert!(!root.join("docs/output.txt").exists());

    let batch = batch.seal();
    assert_eq!(batch.identity_proof().as_str(), "workspace");

    let report = batch.execute_synchronous(&filesystem);

    assert_eq!(report.identity(), "workspace");
    assert_eq!(report.identity_proof().as_str(), "workspace");
    assert_eq!(report.completions().len(), 1);
    assert_eq!(
        report.completions()[0].workspace_path().as_str(),
        "docs/output.txt"
    );
    assert_eq!(
        fs::read_to_string(root.join("docs/output.txt")).expect("output should be readable"),
        "deferred"
    );
    let write = report.completions()[0]
        .as_write()
        .expect("completion should be a write");
    assert!(write.outcome().is_ok());
    let debug = format!("{report:?}");
    assert!(debug.contains("display_path_byte_len"));
    assert!(!debug.contains(root.to_string_lossy().as_ref()));
    assert!(!debug.contains("output.txt"));
    assert!(!debug.contains("deferred"));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn failed_guest_update_discards_workspace_io_without_touching_filesystem() {
    let root = temp_dir("workspace-io-discard");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );

    let write = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/output.txt").expect("path should validate"),
        )
        .expect("write should be authorized");
    batch
        .push_write(write, b"discarded".to_vec())
        .expect("write should queue");

    assert_eq!(batch.identity(), "workspace");
    assert_eq!(batch.identity_proof().as_str(), "workspace");
    let report = batch.discard();

    assert_eq!(report.identity(), "workspace");
    assert_eq!(report.identity_proof().as_str(), "workspace");
    assert_eq!(report.discarded_requests(), 1);
    assert!(!root.join("docs/output.txt").exists());

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn sealed_workspace_io_reports_bounded_reads_without_debug_disclosing_bytes() {
    let root = temp_dir("workspace-io-read");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/input.txt"), "secret read bytes").expect("fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    batch.push_read(read).expect("read should queue");

    let report = batch.seal().execute_synchronous(&filesystem);
    let completion = &report.completions()[0];

    assert_eq!(completion.kind(), WorkspaceAccessKind::Read);
    assert_eq!(completion.workspace_path().as_str(), "docs/input.txt");
    assert_eq!(completion.workspace_relative_path(), "docs/input.txt");
    let read = completion.as_read().expect("completion should be a read");
    match read.outcome() {
        Ok(success) => {
            assert_eq!(success.bytes(), b"secret read bytes");
        }
        other => panic!("expected successful read, got {other:?}"),
    }
    let debug = format!("{report:?}");
    assert!(debug.contains("byte_len"));
    assert!(!debug.contains("secret read bytes"));
    assert!(!debug.contains(root.to_string_lossy().as_ref()));
    assert!(!debug.contains("input.txt"));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn failed_workspace_io_reports_do_not_disclose_host_paths() {
    let root = temp_dir("workspace-io-redacted-error");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/missing.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    batch.push_read(read).expect("read should queue");

    let report = batch.seal().execute_synchronous(&filesystem);
    let debug = format!("{report:?}");
    let display = report.completions()[0]
        .as_read()
        .expect("completion should be a read")
        .outcome()
        .as_ref()
        .expect_err("read should fail")
        .to_string();

    assert!(debug.contains("Read"));
    assert_eq!(display, "plugin workspace read denied");
    assert!(!debug.contains(root.to_string_lossy().as_ref()));
    assert!(!debug.contains("missing.txt"));
    assert!(!display.contains(root.to_string_lossy().as_ref()));
    assert!(!display.contains("missing.txt"));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_outcome_projects_read_success_without_path_metadata() {
    let root = temp_dir("workspace-observe-success");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/secret.txt"), b"secret observed bytes")
        .expect("fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/secret.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    batch.push_read(read).expect("read should queue");
    let report = batch.seal().execute_synchronous(&filesystem);
    let completion = &report.completions()[0];

    let outcome = PluginWorkspaceObserveOutcome::try_from(completion)
        .expect("read completion should project to observe outcome");

    assert!(outcome.is_ok());
    assert_eq!(
        outcome.shape(),
        PluginWorkspaceObserveOutcomeShape::Bytes {
            byte_len: "secret observed bytes".len()
        }
    );
    assert!(outcome.shape().is_ok());
    assert_eq!(outcome.bytes(), Some(b"secret observed bytes".as_slice()));
    let debug = format!("{outcome:?}");
    assert!(debug.contains("byte_len"));
    assert!(!debug.contains("secret observed bytes"));
    assert!(!debug.contains("secret.txt"));
    assert!(!debug.contains(root.to_string_lossy().as_ref()));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_outcome_projects_read_failures_as_closed_rejections() {
    let root = temp_dir("workspace-observe-rejected");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/missing.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    batch.push_read(read).expect("read should queue");
    let report = batch.seal().execute_synchronous(&filesystem);
    let completion = &report.completions()[0];

    let outcome = PluginWorkspaceObserveOutcome::try_from(completion)
        .expect("read completion should project to observe outcome");

    assert!(!outcome.is_ok());
    assert_eq!(
        outcome.shape(),
        PluginWorkspaceObserveOutcomeShape::Rejected
    );
    assert!(!outcome.shape().is_ok());
    assert_eq!(outcome.bytes(), None);
    assert_eq!(
        format!("{outcome:?}"),
        "PluginWorkspaceObserveOutcome { shape: Rejected }"
    );

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_outcome_rejects_write_completions() {
    let root = temp_dir("workspace-observe-write");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let batch = sealed_write_batch(&host, "workspace", "docs/out.txt", b"artifact");
    let report = batch.execute_synchronous(&filesystem);
    let completion = &report.completions()[0];

    let error = PluginWorkspaceObserveOutcome::try_from(completion)
        .expect_err("write completion should not project to observe outcome");

    assert_eq!(
        error,
        PluginWorkspaceObserveProjectionError::WrongOperation {
            kind: WorkspaceAccessKind::Write,
        }
    );
    assert_eq!(error.kind(), WorkspaceAccessKind::Write);
    assert_eq!(
        error.to_string(),
        "workspace observe projection requires read completion, got write"
    );

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_queue_executes_reads_into_retained_guest_outcomes() {
    let root = temp_dir("workspace-observe-task");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/input.txt"), b"secret task bytes")
        .expect("fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let identity = access.identity_proof().clone();

    let task_handle = queue
        .push(access, read_limit(64))
        .expect("task should queue");
    let task_id = task_handle.task_id();

    assert_eq!(task_id.get(), 1);
    assert_eq!(queue.pending_len(), 1);
    assert_eq!(queue.retained_len(), 0);
    assert_eq!(
        queue.poll(&task_handle),
        PluginWorkspaceObserveTaskPoll::Pending
    );
    assert_eq!(
        queue.state(&task_handle),
        PluginWorkspaceObserveTaskState::Pending
    );
    assert_eq!(
        queue.take(&task_handle),
        PluginWorkspaceObserveTaskTake::Pending
    );

    let completion = queue
        .execute_next(&filesystem)
        .expect("task should execute");

    assert_eq!(completion.identity_proof().as_str(), "workspace");
    assert_eq!(completion.task_id(), task_id);
    let completion_shape = completion.shape();
    assert_eq!(
        completion_shape,
        PluginWorkspaceObserveTaskCompletionShape::new(
            task_handle.clone(),
            PluginWorkspaceObserveOutcomeShape::Bytes {
                byte_len: "secret task bytes".len(),
            },
        )
    );
    assert_eq!(completion_shape.identity_proof(), &identity);
    assert_eq!(completion_shape.task_id(), task_id);
    assert_eq!(completion_shape.handle(), completion.handle());
    assert_eq!(
        completion_shape.outcome_shape(),
        PluginWorkspaceObserveOutcomeShape::Bytes {
            byte_len: "secret task bytes".len(),
        }
    );
    assert!(completion_shape.is_ok());
    assert_eq!(
        completion_shape.to_string(),
        "identity=\"workspace\" task_id=1 outcome=bytes byte_len=17"
    );
    assert_eq!(
        completion.outcome().bytes(),
        Some(b"secret task bytes".as_slice())
    );
    assert_eq!(
        queue.poll(&task_handle),
        PluginWorkspaceObserveTaskPoll::Completed {
            outcome: completion.outcome().shape(),
        }
    );
    assert_eq!(
        queue.poll(&task_handle).outcome_shape(),
        Some(PluginWorkspaceObserveOutcomeShape::Bytes {
            byte_len: "secret task bytes".len(),
        })
    );
    assert_eq!(
        queue.state(&task_handle),
        PluginWorkspaceObserveTaskState::Completed
    );
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(queue.retained_len(), 1);
    let state_debug = format!("{:?}", queue.state(&task_handle));
    assert_eq!(state_debug, "completed");
    assert!(!state_debug.contains("secret task bytes"));
    assert!(!state_debug.contains("input.txt"));
    let debug = format!("{queue:?}");
    assert_count_shaped_observe_task_queue_debug(&debug, &root);
    let completion_debug = format!("{completion:?}");
    assert!(
        completion_debug.contains("identity=\"workspace\" task_id=1 outcome=bytes byte_len=17")
    );
    assert!(!completion_debug.contains("secret task bytes"));
    assert!(!completion_debug.contains("input.txt"));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_queue_executes_up_to_budget_without_draining_tail() {
    let root = temp_dir("workspace-observe-task-execute-up-to");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    for (path, bytes) in [
        ("docs/one.txt", b"one".as_slice()),
        ("docs/two.txt", b"two".as_slice()),
        ("docs/three.txt", b"three".as_slice()),
    ] {
        fs::write(root.join(path), bytes).expect("fixture should be written");
    }
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let mut handles = Vec::new();

    for path in ["docs/one.txt", "docs/two.txt", "docs/three.txt"] {
        let access = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from(path).expect("path should validate"),
            )
            .expect("read should be authorized");
        handles.push(
            queue
                .push(access, read_limit(64))
                .expect("task should queue"),
        );
    }

    let completions = queue.execute_up_to(
        &filesystem,
        NonZeroUsize::new(2).expect("execution budget should be non-zero"),
    );

    assert_eq!(completions.len(), 2);
    assert_eq!(completions[0].outcome().bytes(), Some(b"one".as_slice()));
    assert_eq!(completions[1].outcome().bytes(), Some(b"two".as_slice()));
    assert_eq!(queue.pending_len(), 1);
    assert_eq!(queue.retained_len(), 2);
    assert_eq!(
        queue.poll(&handles[2]),
        PluginWorkspaceObserveTaskPoll::Pending
    );

    let tail = queue.execute_all(&filesystem);

    assert_eq!(tail.len(), 1);
    assert_eq!(tail[0].outcome().bytes(), Some(b"three".as_slice()));
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(queue.retained_len(), 3);

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_queue_bounds_pending_and_retained_results() {
    let root = temp_dir("workspace-observe-task-bounds");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/one.txt"), b"one").expect("first fixture should be written");
    fs::write(root.join("docs/two.txt"), b"two").expect("second fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(1, 1).expect("limit should validate"),
    );

    let first = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/one.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let identity = first.identity_proof().clone();
    let first_handle = queue
        .push(first, read_limit(64))
        .expect("first task should queue");
    let extra = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/two.txt").expect("path should validate"),
        )
        .expect("read should be authorized");

    let error = queue
        .push(extra, read_limit(64))
        .expect_err("second pending task should exceed cap");
    assert_eq!(
        error,
        PluginWorkspaceObserveTaskQueueError::TooManyPending {
            identity,
            limit: count_limit(1),
        }
    );
    let event = error
        .operational_event()
        .expect("task queue saturation should be observable");
    assert_eq!(event.identity(), Some("workspace"));
    assert_eq!(
        event.to_string(),
        "plugin \"workspace\" saturated workspace-observe-task queue at 1"
    );
    assert_eq!(
        event.kind(),
        crate::plugin::PluginOperationalEventKind::QueueSaturated {
            queue: PluginOperationalQueue::WorkspaceObserveTask,
            limit: count_limit(1),
        }
    );

    let _first_completion = queue
        .execute_next(&filesystem)
        .expect("first task should execute");
    let second = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/two.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let second_handle = queue
        .push(second, read_limit(64))
        .expect("second task should queue after first completes");
    let second_completion = queue
        .execute_next(&filesystem)
        .expect("second task should execute");

    assert_eq!(queue.retained_len(), 1);
    assert_eq!(
        queue.poll(&first_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    assert_eq!(
        queue.poll(&second_handle),
        PluginWorkspaceObserveTaskPoll::Completed {
            outcome: PluginWorkspaceObserveOutcomeShape::Bytes { byte_len: 3 },
        }
    );
    assert_eq!(second_completion.outcome().bytes(), Some(b"two".as_slice()));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_queue_limit_fields_are_closed() {
    assert_eq!(
        PluginWorkspaceObserveTaskQueueLimit::try_new(0, 1),
        Err(PluginWorkspaceObserveTaskQueueError::ZeroLimit {
            field: PluginWorkspaceObserveTaskQueueLimitField::MaxPending,
        })
    );
    assert_eq!(
        PluginWorkspaceObserveTaskQueueLimit::try_new(1, 0),
        Err(PluginWorkspaceObserveTaskQueueError::ZeroLimit {
            field: PluginWorkspaceObserveTaskQueueLimitField::MaxRetained,
        })
    );
    assert_eq!(
        PluginWorkspaceObserveTaskQueueLimitField::MaxPending.as_str(),
        "max_pending"
    );
    assert_eq!(
        PluginWorkspaceObserveTaskQueueLimitField::MaxRetained.to_string(),
        "max_retained"
    );
    assert_eq!(
        PluginWorkspaceObserveTaskQueueLimit::try_new(0, 1)
            .expect_err("zero pending limit should fail")
            .to_string(),
        "plugin workspace observe task queue max_pending must be non-zero"
    );
    assert_eq!(
        PluginWorkspaceObserveTaskReservationMismatch::IdentityMismatch.as_str(),
        "identity-mismatch"
    );
    assert_eq!(
        PluginWorkspaceObserveTaskReservationMismatch::QueueMismatch.to_string(),
        "queue-mismatch"
    );
}

#[test]
fn workspace_observe_task_queue_error_kinds_are_closed() {
    let identity = plugin_identity("workspace");
    let cases = [
        (
            PluginWorkspaceObserveTaskQueueError::ZeroLimit {
                field: PluginWorkspaceObserveTaskQueueLimitField::MaxPending,
            },
            PluginWorkspaceObserveTaskQueueErrorKind::ZeroLimit,
            "zero-limit",
        ),
        (
            PluginWorkspaceObserveTaskQueueError::TooManyPending {
                identity: identity.clone(),
                limit: count_limit(1),
            },
            PluginWorkspaceObserveTaskQueueErrorKind::TooManyPending,
            "too-many-pending",
        ),
        (
            PluginWorkspaceObserveTaskQueueError::TaskIdExhausted {
                identity: identity.clone(),
            },
            PluginWorkspaceObserveTaskQueueErrorKind::TaskIdExhausted,
            "task-id-exhausted",
        ),
        (
            PluginWorkspaceObserveTaskQueueError::ReservationMismatch {
                identity,
                reason: PluginWorkspaceObserveTaskReservationMismatch::QueueMismatch,
            },
            PluginWorkspaceObserveTaskQueueErrorKind::ReservationMismatch,
            "reservation-mismatch",
        ),
    ];

    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 workspace_observe_task_batch_enqueues_after_successful_return() {
    let root = temp_dir("workspace-observe-task-batch");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/one.txt"), b"one").expect("first fixture should be written");
    fs::write(root.join("docs/two.txt"), b"two").expect("second fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let mut reserved_task_ids = Vec::new();

    for path in ["docs/one.txt", "docs/two.txt"] {
        let access = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from(path).expect("path should validate"),
            )
            .expect("read should be authorized");
        reserved_task_ids.push(
            batch
                .push_reserved(access, &mut workspace_io, &mut queue)
                .expect("task handle should reserve")
                .task_id(),
        );
    }

    let sealed = batch.seal();
    assert_eq!(sealed.identity(), "workspace");
    assert_eq!(sealed.identity_proof(), &identity);
    assert_eq!(sealed.task_count(), 2);
    assert_eq!(queue.reserved_pending_len(), 2);
    assert_eq!(queue.pending_len(), 0);

    let report = sealed
        .enqueue(&mut queue)
        .expect("sealed task batch should enqueue");

    assert_eq!(report.identity_proof(), &identity);
    assert_eq!(
        report
            .task_handles()
            .iter()
            .map(PluginWorkspaceObserveTaskHandle::task_id)
            .collect::<Vec<_>>(),
        reserved_task_ids
    );
    assert_eq!(queue.pending_len(), 2);
    assert_eq!(queue.reserved_pending_len(), 0);
    let completions = queue.execute_all(&filesystem);
    assert_eq!(completions.len(), 2);
    assert_eq!(completions[0].outcome().bytes(), Some(b"one".as_slice()));
    assert_eq!(completions[1].outcome().bytes(), Some(b"two".as_slice()));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_batch_retains_update_read_byte_limit() {
    let root = temp_dir("workspace-observe-task-read-limit");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/large.txt"), b"secret task bytes")
        .expect("fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity,
        PluginWorkspaceIoBudget::try_new(1, 4, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/large.txt").expect("path should validate"),
        )
        .expect("read should be authorized");

    let task_handle = batch
        .push_reserved(access, &mut workspace_io, &mut queue)
        .expect("task handle should reserve");
    let task_id = task_handle.task_id();
    let enqueue = batch
        .seal()
        .enqueue(&mut queue)
        .expect("sealed task should enqueue");
    assert_eq!(
        enqueue.task_handles()[0].identity_proof().as_str(),
        "workspace"
    );
    assert_eq!(enqueue.task_handles()[0].task_id(), task_id);
    let completion = queue
        .execute_next(&filesystem)
        .expect("task should execute through retained read cap");

    assert_eq!(completion.task_id(), task_id);
    assert_eq!(
        completion.outcome(),
        &PluginWorkspaceObserveOutcome::Rejected
    );
    let debug = format!("{completion:?}");
    assert!(!debug.contains("secret task bytes"));
    assert!(!debug.contains("large.txt"));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_batch_debug_redacts_unsealed_paths() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity,
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let secret_path = "docs/secret-unsealed-task-path.txt";
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from(secret_path).expect("path should validate"),
        )
        .expect("read should be authorized");

    let _task_handle = batch
        .push_reserved(access, &mut workspace_io, &mut queue)
        .expect("task handle should reserve");
    let pending_debug = format!("{batch:?}");

    assert_eq!(queue.reserved_pending_len(), 1);
    assert!(pending_debug.contains("PendingPluginWorkspaceObserveTaskBatch"));
    assert!(pending_debug.contains("task_count"));
    assert!(!pending_debug.contains("task_shapes"));
    assert!(!pending_debug.contains("path_byte_len"));
    assert!(!pending_debug.contains("task_id"));
    assert!(!pending_debug.contains(secret_path));
    assert!(!pending_debug.contains("secret-unsealed-task-path"));

    let sealed = batch.seal();
    let sealed_debug = format!("{sealed:?}");

    assert!(sealed_debug.contains("SealedPluginWorkspaceObserveTaskBatch"));
    assert!(sealed_debug.contains("task_count"));
    assert!(!sealed_debug.contains("task_shapes"));
    assert!(!sealed_debug.contains("path_byte_len"));
    assert!(!sealed_debug.contains("task_id"));
    assert!(!sealed_debug.contains(secret_path));
    assert!(!sealed_debug.contains("secret-unsealed-task-path"));
    drop(sealed);
    assert_eq!(queue.reserved_pending_len(), 0);
}

#[test]
fn workspace_observe_task_reservation_holds_queue_capacity_until_enqueue() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(2, 1).expect("limit should validate"),
    );
    for path in ["docs/one.txt", "docs/two.txt"] {
        let access = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from(path).expect("path should validate"),
            )
            .expect("read should be authorized");
        let _reserved_task_handle = batch
            .push_reserved(access, &mut workspace_io, &mut queue)
            .expect("task handle should reserve");
    }
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(queue.reserved_pending_len(), 2);
    let sealed = batch.seal();
    let competing_access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/competing.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let error = queue
        .push(competing_access, read_limit(64))
        .expect_err("reserved slots should count against queue capacity");

    assert_eq!(
        error,
        PluginWorkspaceObserveTaskQueueError::TooManyPending {
            identity,
            limit: count_limit(2),
        }
    );
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(queue.reserved_pending_len(), 2);

    let report = sealed
        .enqueue(&mut queue)
        .expect("reserved batch should still enqueue after competing push rejection");

    assert_eq!(
        report
            .task_handles()
            .iter()
            .map(|handle| handle.task_id().get())
            .collect::<Vec<_>>(),
        vec![1, 2]
    );
    assert_eq!(queue.pending_len(), 2);
    assert_eq!(queue.reserved_pending_len(), 0);
    assert_eq!(queue.retained_len(), 0);
}

proptest! {
    #[test]
    fn workspace_observe_task_queue_rejections_leave_reservations_unchanged(limit in 1usize..8) {
        let host = workspace_host();
        let identity = plugin_identity("workspace");
        let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
        let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
            identity.clone(),
            PluginWorkspaceIoBudget::try_new(limit + 1, 64, 64)
                .expect("budget should validate"),
        );
        let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
            PluginWorkspaceObserveTaskQueueLimit::try_new(limit, 1)
                .expect("limit should validate"),
        );

        for index in 0..limit {
            let path = format!("docs/{index}.txt");
            let access = host
                .authorize_workspace_observe(
                    WorkspacePathRef::try_from(path.as_str()).expect("path should validate"),
                )
                .expect("read should be authorized");
            let _task_handle = batch
                .push_reserved(access, &mut workspace_io, &mut queue)
                .expect("task handle should reserve before queue cap");
        }

        let next_id = queue.next_id_for_test();
        let extra = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from("docs/extra.txt").expect("path should validate"),
            )
            .expect("extra read should be authorized");
        let error = batch
            .push_reserved(extra, &mut workspace_io, &mut queue)
            .expect_err("queue cap should reject without mutation");

        prop_assert_eq!(
            error,
            PluginWorkspaceObserveTaskReserveError::Queue(
                PluginWorkspaceObserveTaskQueueError::TooManyPending {
                    identity,
                    limit: count_limit(limit),
                }
            )
        );
        prop_assert_eq!(workspace_io.accepted_request_count(), limit);
        prop_assert_eq!(queue.pending_len(), 0);
        prop_assert_eq!(queue.reserved_pending_len(), limit);
        prop_assert_eq!(queue.next_id_for_test(), next_id);
        prop_assert_eq!(batch.seal().task_count(), limit);
    }

    #[test]
    fn workspace_observe_task_budget_rejections_do_not_reserve_task_ids(limit in 1usize..8) {
        let host = workspace_host();
        let identity = plugin_identity("workspace");
        let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
        let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
            identity.clone(),
            PluginWorkspaceIoBudget::try_new(limit, 64, 64).expect("budget should validate"),
        );
        let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
            PluginWorkspaceObserveTaskQueueLimit::try_new(limit + 1, 1)
                .expect("limit should validate"),
        );

        for index in 0..limit {
            let path = format!("docs/{index}.txt");
            let access = host
                .authorize_workspace_observe(
                    WorkspacePathRef::try_from(path.as_str()).expect("path should validate"),
                )
                .expect("read should be authorized");
            let _task_handle = batch
                .push_reserved(access, &mut workspace_io, &mut queue)
                .expect("task handle should reserve before budget cap");
        }

        let next_id = queue.next_id_for_test();
        let extra = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from("docs/extra.txt").expect("path should validate"),
            )
            .expect("extra read should be authorized");
        let error = batch
            .push_reserved(extra, &mut workspace_io, &mut queue)
            .expect_err("budget cap should reject before queue reservation");

        prop_assert_eq!(
            error,
            PluginWorkspaceObserveTaskReserveError::WorkspaceIo(
                PluginWorkspaceIoCommitError::TooManyRequests {
                    identity,
                    limit: count_limit(limit),
                }
            )
        );
        prop_assert_eq!(workspace_io.accepted_request_count(), limit);
        prop_assert_eq!(queue.pending_len(), 0);
        prop_assert_eq!(queue.reserved_pending_len(), limit);
        prop_assert_eq!(queue.next_id_for_test(), next_id);
        prop_assert_eq!(batch.seal().task_count(), limit);
    }
}

#[test]
fn sealed_workspace_observe_task_batch_rejects_cross_queue_without_losing_batch() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(2, 64, 64).expect("budget should validate"),
    );
    let mut source_queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(2, 1).expect("limit should validate"),
    );
    let mut other_queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(2, 1).expect("limit should validate"),
    );

    for path in [
        "docs/secret-cross-queue-one.txt",
        "docs/secret-cross-queue-two.txt",
    ] {
        let access = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from(path).expect("path should validate"),
            )
            .expect("read should be authorized");
        let _task_handle = batch
            .push_reserved(access, &mut workspace_io, &mut source_queue)
            .expect("task handle should reserve");
    }
    let sealed = batch.seal();

    let error = sealed
        .enqueue(&mut other_queue)
        .expect_err("reservations from another queue should reject");

    assert_eq!(
        error.queue_error(),
        &PluginWorkspaceObserveTaskQueueError::ReservationMismatch {
            identity: identity.clone(),
            reason: PluginWorkspaceObserveTaskReservationMismatch::QueueMismatch,
        }
    );
    assert_eq!(error.batch().identity_proof(), &identity);
    assert_eq!(error.batch().task_count(), 2);
    let debug = format!("{error:?}");
    assert!(debug.contains("PluginWorkspaceObserveTaskEnqueueError"));
    assert!(debug.contains("task_count"));
    assert!(!debug.contains("task_shapes"));
    assert!(!debug.contains("path_byte_len"));
    assert!(!debug.contains("secret-cross-queue"));
    assert_eq!(source_queue.pending_len(), 0);
    assert_eq!(source_queue.reserved_pending_len(), 2);
    assert_eq!(other_queue.pending_len(), 0);
    assert_eq!(other_queue.reserved_pending_len(), 0);

    let parts = error.into_parts();
    assert_eq!(
        parts.queue_error,
        PluginWorkspaceObserveTaskQueueError::ReservationMismatch {
            identity: identity.clone(),
            reason: PluginWorkspaceObserveTaskReservationMismatch::QueueMismatch,
        }
    );

    let retry = parts
        .batch
        .enqueue(&mut source_queue)
        .expect("returned sealed batch should retry on the owning queue");

    assert_eq!(retry.identity_proof(), &identity);
    assert_eq!(retry.task_count(), 2);
    let retry_debug = format!("{retry:?}");
    assert!(retry_debug.contains("task_count"));
    assert!(!retry_debug.contains("task_handles"));
    assert_eq!(
        retry
            .task_handles()
            .iter()
            .map(|handle| handle.task_id().get())
            .collect::<Vec<_>>(),
        vec![1, 2]
    );
    assert_eq!(source_queue.pending_len(), 2);
    assert_eq!(source_queue.reserved_pending_len(), 0);
}

#[test]
fn sealed_workspace_observe_task_enqueue_error_can_discard_retry_batch() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let secret_path = "docs/secret-discarded-retry-task.txt";
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let mut source_queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(1, 1).expect("limit should validate"),
    );
    let mut other_queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(1, 1).expect("limit should validate"),
    );
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from(secret_path).expect("path should validate"),
        )
        .expect("read should be authorized");
    let _task_handle = batch
        .push_reserved(access, &mut workspace_io, &mut source_queue)
        .expect("task handle should reserve");
    let sealed = batch.seal();

    let error = sealed
        .enqueue(&mut other_queue)
        .expect_err("wrong queue should keep the batch retryable");
    let debug = format!("{error:?}");

    assert!(debug.contains("PluginWorkspaceObserveTaskEnqueueError"));
    assert!(debug.contains("task_count"));
    assert!(!debug.contains("task_shapes"));
    assert!(!debug.contains("path_byte_len"));
    assert!(!debug.contains(secret_path));
    assert!(!debug.contains("secret-discarded-retry-task"));
    assert_eq!(source_queue.reserved_pending_len(), 1);
    assert_eq!(source_queue.pending_len(), 0);
    assert_eq!(other_queue.reserved_pending_len(), 0);
    assert_eq!(other_queue.pending_len(), 0);

    let report = error.discard_batch();

    assert_eq!(report.identity_proof(), &identity);
    assert_eq!(report.identity(), "workspace");
    assert_eq!(report.discarded_tasks(), 1);
    assert_eq!(source_queue.reserved_pending_len(), 0);
    assert_eq!(source_queue.pending_len(), 0);
    assert_eq!(other_queue.reserved_pending_len(), 0);
    assert_eq!(other_queue.pending_len(), 0);
}

#[test]
fn workspace_observe_task_queue_accepts_empty_sealed_batches_as_noops() {
    let identity = plugin_identity("workspace");
    let batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone()).seal();
    let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(1, 1).expect("limit should validate"),
    );

    assert!(batch.is_empty());
    let report = batch
        .enqueue(&mut queue)
        .expect("empty task batch should enqueue as a no-op");

    assert_eq!(report.identity_proof(), &identity);
    assert!(report.task_handles().is_empty());
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(queue.reserved_pending_len(), 0);
    assert!(queue.is_empty());
}

#[test]
fn workspace_observe_task_cancel_all_leaves_retry_batch_caller_owned() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let secret_path = "docs/secret-retry-after-cancel-task.txt";
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(2, 64, 64).expect("budget should validate"),
    );
    let mut source_queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(2, 1).expect("limit should validate"),
    );
    let mut other_queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(2, 1).expect("limit should validate"),
    );
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from(secret_path).expect("path should validate"),
        )
        .expect("read should be authorized");
    let reserved_handle = batch
        .push_reserved(access, &mut workspace_io, &mut source_queue)
        .expect("task handle should reserve");
    let reserved_id = reserved_handle.task_id();
    let error = batch
        .seal()
        .enqueue(&mut other_queue)
        .expect_err("wrong queue should keep the batch retryable");

    let reports = source_queue.cancel_all();

    assert!(reports.is_empty());
    assert_eq!(reserved_id.get(), 1);
    assert_eq!(source_queue.reserved_pending_len(), 1);
    assert!(!source_queue.is_empty());
    assert_eq!(other_queue.reserved_pending_len(), 0);
    assert_eq!(error.batch().task_count(), 1);
    let debug = format!("{error:?}");
    assert!(!debug.contains(secret_path));
    assert!(!debug.contains("secret-retry-after-cancel-task"));

    let retry = error
        .into_batch()
        .enqueue(&mut source_queue)
        .expect("caller-owned retry batch should still enter its queue");

    assert_eq!(retry.task_handles()[0].task_id(), reserved_id);
    assert_eq!(source_queue.pending_len(), 1);
    assert_eq!(source_queue.reserved_pending_len(), 0);

    let reports = source_queue.cancel_all();

    assert_eq!(reports.len(), 1);
    assert_eq!(reports[0].identity_proof(), &identity);
    assert_eq!(reports[0].canceled_pending(), 1);
    assert_eq!(reports[0].discarded_retained(), 0);
    assert!(source_queue.is_empty());
    let new_access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/after-cancel.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let new_handle = source_queue
        .push(new_access, read_limit(64))
        .expect("canceled task ids should not be reused");
    assert_eq!(new_handle.task_id().get(), 2);
}

#[test]
fn workspace_observe_task_discard_releases_reserved_queue_capacity() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity,
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(1, 1).expect("limit should validate"),
    );
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/reserved.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let reserved_handle = batch
        .push_reserved(access, &mut workspace_io, &mut queue)
        .expect("task handle should reserve");
    let reserved_id = reserved_handle.task_id();

    assert_eq!(reserved_id.get(), 1);
    assert_eq!(queue.reserved_pending_len(), 1);
    assert_eq!(queue.pending_len(), 0);

    let discard = batch.discard();

    assert_eq!(discard.discarded_tasks(), 1);
    assert_eq!(queue.reserved_pending_len(), 0);
    let direct_access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/direct.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let direct_handle = queue
        .push(direct_access, read_limit(64))
        .expect("released reservation should free queue capacity");
    assert_eq!(direct_handle.task_id().get(), 2);
    assert_eq!(queue.pending_len(), 1);
}

#[test]
fn workspace_observe_task_batch_rejects_access_from_another_plugin_identity() {
    let host = workspace_host_for("other");
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/other.txt").expect("path should validate"),
        )
        .expect("read should be authorized");

    let error = batch
        .push_reserved(access, &mut workspace_io, &mut queue)
        .expect_err("task batch should reject cross-plugin access");

    assert_eq!(
        error,
        PluginWorkspaceObserveTaskReserveError::WorkspaceIo(
            PluginWorkspaceIoCommitError::IdentityMismatch {
                mismatch_source: PluginWorkspaceIoIdentityMismatchSource::AccessProof,
                batch_identity: identity,
                provided_identity: plugin_identity("other"),
            }
        )
    );
    assert_eq!(workspace_io.accepted_request_count(), 0);
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(batch.seal().task_count(), 0);
}

#[test]
fn workspace_observe_task_batch_rejects_workspace_io_from_another_plugin_identity() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("other"),
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let next_id = queue.next_id_for_test();
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");

    let error = batch
        .push_reserved(access, &mut workspace_io, &mut queue)
        .expect_err("task batch should reject cross-plugin workspace I/O ledger");

    assert_eq!(
        error,
        PluginWorkspaceObserveTaskReserveError::WorkspaceIo(
            PluginWorkspaceIoCommitError::IdentityMismatch {
                mismatch_source: PluginWorkspaceIoIdentityMismatchSource::RequestLedger,
                batch_identity: identity,
                provided_identity: plugin_identity("other"),
            }
        )
    );
    assert_eq!(workspace_io.accepted_request_count(), 0);
    assert_eq!(queue.pending_len(), 0);
    assert_eq!(queue.reserved_pending_len(), 0);
    assert_eq!(queue.next_id_for_test(), next_id);
    assert_eq!(batch.seal().task_count(), 0);
}

#[test]
fn workspace_observe_task_batch_exhaustion_check_is_all_or_nothing() {
    let host = workspace_host();
    let identity = plugin_identity("workspace");
    let mut batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should validate"),
    );
    let mut queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(
        PluginWorkspaceObserveTaskQueueLimit::try_new(4, 1).expect("limit should validate"),
    );
    queue.set_next_id_for_test(Some(
        PluginWorkspaceObserveTaskId::try_new(u64::MAX - 1).expect("task id should validate"),
    ));
    for path in ["docs/near-end.txt", "docs/final.txt"] {
        let access = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from(path).expect("path should validate"),
            )
            .expect("read should be authorized");
        let _reserved_task_handle = batch
            .push_reserved(access, &mut workspace_io, &mut queue)
            .expect("task handle should reserve");
    }
    let sealed = batch.seal();

    let report = sealed
        .enqueue(&mut queue)
        .expect("batch should use the final task ids");

    assert_eq!(
        report
            .task_handles()
            .iter()
            .map(|handle| handle.task_id().get())
            .collect::<Vec<_>>(),
        vec![u64::MAX - 1, u64::MAX]
    );
    assert_eq!(queue.next_id_for_test(), None);
    assert_eq!(queue.pending_len(), 2);

    let mut extra_batch = PendingPluginWorkspaceObserveTaskBatch::new(identity.clone());
    let mut extra_workspace_io = PendingPluginWorkspaceIoBatch::from_validated_budget(
        identity.clone(),
        PluginWorkspaceIoBudget::try_new(1, 64, 64).expect("budget should validate"),
    );
    let extra_access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/after-end.txt").expect("path should validate"),
        )
        .expect("read should be authorized");

    let error = extra_batch
        .push_reserved(extra_access, &mut extra_workspace_io, &mut queue)
        .expect_err("exhausted ids should reject before mutation");

    assert_eq!(
        error,
        PluginWorkspaceObserveTaskReserveError::Queue(
            PluginWorkspaceObserveTaskQueueError::TaskIdExhausted { identity }
        )
    );
    assert_eq!(extra_workspace_io.accepted_request_count(), 0);
    assert_eq!(extra_batch.seal().task_count(), 0);
    assert_eq!(queue.pending_len(), 2);
}

#[test]
fn workspace_observe_task_ids_are_scoped_by_identity() {
    let root = temp_dir("workspace-observe-task-identity");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/input.txt"), b"secret owned task bytes")
        .expect("fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let other = plugin_identity("other");
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let access = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let owner_handle = queue
        .push(access, read_limit(64))
        .expect("task should queue");
    let task_id = owner_handle.task_id();
    let other_handle = task_handle(&other, task_id);

    assert_eq!(
        queue.take(&owner_handle),
        PluginWorkspaceObserveTaskTake::Pending
    );
    assert_eq!(
        queue.state(&owner_handle),
        PluginWorkspaceObserveTaskState::Pending
    );
    assert_eq!(
        queue.take(&other_handle),
        PluginWorkspaceObserveTaskTake::Unknown
    );
    assert_eq!(
        queue.state(&other_handle),
        PluginWorkspaceObserveTaskState::Unknown
    );
    assert_eq!(
        queue.poll(&other_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );

    let _completion = queue
        .execute_next(&filesystem)
        .expect("task should execute");

    assert_eq!(
        queue.state(&owner_handle),
        PluginWorkspaceObserveTaskState::Completed
    );
    assert_eq!(
        queue.take(&other_handle),
        PluginWorkspaceObserveTaskTake::Unknown
    );
    let poll = queue.poll(&owner_handle);
    let poll_debug = format!("{poll:?}");
    assert!(poll_debug.contains("completed"));
    assert!(poll_debug.contains("byte_len"));
    assert!(!poll_debug.contains("PluginWorkspaceObserveBytes"));
    assert!(!poll_debug.contains("secret owned task bytes"));
    assert!(!poll_debug.contains("input.txt"));

    let take = queue.take(&owner_handle);
    let take_debug = format!("{take:?}");
    assert!(take_debug.contains("completed"));
    assert!(take_debug.contains("byte_len"));
    assert!(!take_debug.contains("PluginWorkspaceObserveBytes"));
    assert!(!take_debug.contains("secret owned task bytes"));
    assert!(!take_debug.contains("input.txt"));
    assert!(take.is_completed());
    assert_eq!(
        take.outcome()
            .and_then(PluginWorkspaceObserveOutcome::bytes),
        Some(b"secret owned task bytes".as_slice())
    );
    assert_eq!(
        take.into_outcome()
            .and_then(PluginWorkspaceObserveOutcome::into_bytes),
        Some(b"secret owned task bytes".to_vec())
    );
    assert_eq!(
        queue.poll(&owner_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    assert_eq!(
        queue.state(&owner_handle),
        PluginWorkspaceObserveTaskState::Unknown
    );
    assert_eq!(
        queue.take(&owner_handle),
        PluginWorkspaceObserveTaskTake::Unknown
    );

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_queue_cancels_revoked_identity_state() {
    let root = temp_dir("workspace-observe-task-cancel");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(root.join("docs/retained.txt"), b"retained")
        .expect("retained fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let revoked = plugin_identity("workspace");
    let kept_host = workspace_host_for("other");
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let retained = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/retained.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let pending = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/pending.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let kept_pending = kept_host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/kept.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let retained_handle = queue
        .push(retained, read_limit(64))
        .expect("retained task should queue");
    let _retained_completion = queue
        .execute_next(&filesystem)
        .expect("retained task should execute");
    let pending_handle = queue
        .push(pending, read_limit(64))
        .expect("pending task should queue");
    let kept_handle = queue
        .push(kept_pending, read_limit(64))
        .expect("kept task should queue");

    let report = queue.cancel_identity(&revoked);

    assert_eq!(report.identity(), "workspace");
    assert_eq!(report.identity_proof().as_str(), "workspace");
    assert_eq!(report.canceled_pending(), 1);
    assert_eq!(report.discarded_retained(), 1);
    assert_eq!(
        queue.poll(&retained_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    assert_eq!(
        queue.poll(&pending_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    assert_eq!(
        queue.poll(&kept_handle),
        PluginWorkspaceObserveTaskPoll::Pending
    );
    assert_eq!(queue.pending_len(), 1);
    assert_eq!(queue.retained_len(), 0);

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_observe_task_queue_cancel_all_reports_by_identity() {
    let root = temp_dir("workspace-observe-task-cancel-all");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    fs::write(
        root.join("docs/secret-retained-task.txt"),
        b"secret retained task bytes",
    )
    .expect("retained fixture should be written");
    let filesystem = config(&root);
    let host = workspace_host();
    let other_host = workspace_host_for("other");
    let mut queue = PluginWorkspaceObserveTaskQueue::default();
    let retained = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/secret-retained-task.txt")
                .expect("path should validate"),
        )
        .expect("read should be authorized");
    let retained_handle = queue
        .push(retained, read_limit(64))
        .expect("retained task should queue");
    let _retained_completion = queue
        .execute_next(&filesystem)
        .expect("retained task should execute");
    let pending = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/secret-pending-task.txt")
                .expect("path should validate"),
        )
        .expect("read should be authorized");
    let pending_handle = queue
        .push(pending, read_limit(64))
        .expect("pending task should queue");
    let other_pending = other_host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/other-pending-task.txt")
                .expect("path should validate"),
        )
        .expect("other read should be authorized");
    let other_handle = queue
        .push(other_pending, read_limit(64))
        .expect("other task should queue");

    let reports = queue.cancel_all();

    assert_eq!(reports.len(), 2);
    assert_eq!(reports[0].identity(), "workspace");
    assert_eq!(reports[0].canceled_pending(), 1);
    assert_eq!(reports[0].discarded_retained(), 1);
    assert_eq!(reports[1].identity(), "other");
    assert_eq!(reports[1].canceled_pending(), 1);
    assert_eq!(reports[1].discarded_retained(), 0);
    assert!(queue.is_empty());
    assert_eq!(
        queue.poll(&retained_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    assert_eq!(
        queue.poll(&pending_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    assert_eq!(
        queue.poll(&other_handle),
        PluginWorkspaceObserveTaskPoll::Unknown
    );
    let after_cancel = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/after-cancel-all.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let after_cancel_handle = queue
        .push(after_cancel, read_limit(64))
        .expect("canceled task ids should not be reused");
    assert_eq!(after_cancel_handle.task_id().get(), 4);
    assert_eq!(queue.pending_len(), 1);
    let debug = format!("{reports:?}");
    assert!(!debug.contains("secret-retained-task"));
    assert!(!debug.contains("secret-pending-task"));
    assert!(!debug.contains("secret retained task bytes"));
    assert!(!debug.contains(root.to_string_lossy().as_ref()));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_io_batch_rejects_over_budget_without_partial_queueing() {
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(1, 64, 3).expect("budget should be valid"),
    );

    let oversized = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/oversized.txt").expect("path should validate"),
        )
        .expect("write should be authorized");
    assert_eq!(
        batch
            .push_write(oversized, b"large".to_vec())
            .expect_err("oversized payload should fail"),
        PluginWorkspaceIoCommitError::PayloadTooLarge {
            size: 5,
            max_size: 3,
        }
    );
    assert_eq!(batch.request_count(), 0);

    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    batch.push_read(read).expect("read should fit");
    let extra = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/other.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    assert_eq!(
        batch
            .push_read(extra)
            .expect_err("second request should exceed count limit"),
        PluginWorkspaceIoCommitError::TooManyRequests {
            identity: plugin_identity("workspace"),
            limit: count_limit(1),
        }
    );
    assert_eq!(batch.request_count(), 1);
}

proptest! {
    #[test]
    fn workspace_io_request_limit_rejections_leave_pending_batch_unchanged(limit in 1usize..16) {
        let host = workspace_host();
        let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
            plugin_identity("workspace"),
            PluginWorkspaceIoBudget::try_new(limit, 64, 64).expect("budget should be valid"),
        );

        for index in 0..limit {
            let path = format!("docs/{index}.txt");
            let read = host
                .authorize_workspace_observe(
                    WorkspacePathRef::try_from(path.as_str()).expect("path should validate"),
                )
                .expect("read should be authorized");
            batch.push_read(read).expect("read should fit before limit");
        }
        prop_assert_eq!(batch.request_count(), limit);

        let extra = host
            .authorize_workspace_observe(
                WorkspacePathRef::try_from("docs/extra.txt").expect("path should validate"),
            )
            .expect("extra read should be authorized");
        let error = batch
            .push_read(extra)
            .expect_err("extra request should reject");

        prop_assert_eq!(
            error,
            PluginWorkspaceIoCommitError::TooManyRequests {
                identity: plugin_identity("workspace"),
                limit: count_limit(limit),
            }
        );
        prop_assert_eq!(batch.request_count(), limit);
    }
}

#[test]
fn workspace_io_shared_budget_counts_deferred_requests_from_batch_storage() {
    let host = workspace_host();
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(2, 64, 64).expect("budget should be valid"),
    );

    let deferred = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/deferred.txt").expect("path should validate"),
        )
        .expect("deferred read should be authorized");
    batch.push_read(deferred).expect("deferred read should fit");

    assert_eq!(batch.request_count(), 1);
    assert_eq!(batch.non_deferred_request_count(), 0);
    assert_eq!(batch.accepted_request_count(), 1);

    let direct_limit = batch
        .charge_direct_observation()
        .expect("direct read should share remaining budget");

    assert_eq!(direct_limit, read_limit(64));
    assert_eq!(batch.request_count(), 1);
    assert_eq!(batch.non_deferred_request_count(), 1);
    assert_eq!(batch.accepted_request_count(), 2);

    let extra_deferred = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/extra-deferred.txt").expect("path should validate"),
        )
        .expect("extra deferred read should be authorized");
    assert_eq!(
        batch
            .push_read(extra_deferred)
            .expect_err("shared budget should reject extra deferred read"),
        PluginWorkspaceIoCommitError::TooManyRequests {
            identity: plugin_identity("workspace"),
            limit: count_limit(2),
        }
    );
    assert_eq!(
        batch
            .charge_direct_observation()
            .expect_err("shared budget should reject extra direct read"),
        PluginWorkspaceIoCommitError::TooManyRequests {
            identity: plugin_identity("workspace"),
            limit: count_limit(2),
        }
    );
    assert_eq!(batch.request_count(), 1);
    assert_eq!(batch.non_deferred_request_count(), 1);
    assert_eq!(batch.accepted_request_count(), 2);
}

#[test]
fn workspace_io_batch_rejects_access_from_another_plugin_identity() {
    let host = workspace_host_for("other");
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity("workspace"),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget should be valid"),
    );
    let read = host
        .authorize_workspace_observe(
            WorkspacePathRef::try_from("docs/input.txt").expect("path should validate"),
        )
        .expect("read should be authorized");
    let write = host
        .authorize_workspace_artifact_write(
            WorkspacePathRef::try_from("docs/output.txt").expect("path should validate"),
        )
        .expect("write should be authorized");

    assert_eq!(
        batch
            .push_read(read)
            .expect_err("read access from another plugin should reject"),
        PluginWorkspaceIoCommitError::IdentityMismatch {
            mismatch_source: PluginWorkspaceIoIdentityMismatchSource::AccessProof,
            batch_identity: plugin_identity("workspace"),
            provided_identity: plugin_identity("other"),
        }
    );
    assert_eq!(
        batch
            .push_write(write, b"payload".to_vec())
            .expect_err("write access from another plugin should reject"),
        PluginWorkspaceIoCommitError::IdentityMismatch {
            mismatch_source: PluginWorkspaceIoIdentityMismatchSource::AccessProof,
            batch_identity: plugin_identity("workspace"),
            provided_identity: plugin_identity("other"),
        }
    );
    assert_eq!(batch.request_count(), 0);
}

#[test]
fn workspace_worker_queue_limit_fields_are_closed() {
    let error =
        PluginWorkspaceIoWorkerQueueLimit::try_new(0).expect_err("zero limit should reject");

    assert_eq!(
        error,
        PluginWorkspaceIoWorkerQueueError::ZeroLimit {
            field: PluginWorkspaceIoWorkerQueueLimitField::MaxBatches,
        }
    );
    assert_eq!(
        error.kind(),
        PluginWorkspaceIoWorkerQueueErrorKind::ZeroLimit
    );
    assert_eq!(
        PluginWorkspaceIoWorkerQueueLimitField::MaxBatches.as_str(),
        "max_batches"
    );
    assert_eq!(
        PluginWorkspaceIoWorkerQueueLimitField::MaxBatches.to_string(),
        "max_batches"
    );
    assert_eq!(
        format!("{:?}", PluginWorkspaceIoWorkerQueueLimitField::MaxBatches),
        "max_batches"
    );
    assert_eq!(
        error.to_string(),
        "plugin workspace I/O queue limit must be non-zero"
    );
}

#[test]
fn workspace_worker_queue_error_kinds_are_closed() {
    let identity = plugin_identity("workspace");
    let cases = [
        (
            PluginWorkspaceIoWorkerQueueError::ZeroLimit {
                field: PluginWorkspaceIoWorkerQueueLimitField::MaxBatches,
            },
            PluginWorkspaceIoWorkerQueueErrorKind::ZeroLimit,
            "zero-limit",
        ),
        (
            PluginWorkspaceIoWorkerQueueError::TooManyBatches {
                identity,
                limit: count_limit(1),
            },
            PluginWorkspaceIoWorkerQueueErrorKind::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 workspace_worker_queue_is_bounded() {
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::from_validated_limit(
        PluginWorkspaceIoWorkerQueueLimit::try_new(1).expect("limit"),
    );
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/one.txt",
            b"one",
        ))
        .expect("first batch");

    let error = queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/two.txt",
            b"two",
        ))
        .expect_err("second batch should exceed limit");

    assert_eq!(
        error.queue_error(),
        &PluginWorkspaceIoWorkerQueueError::TooManyBatches {
            identity: plugin_identity("workspace"),
            limit: count_limit(1),
        }
    );
    assert_eq!(
        error.queue_error().kind(),
        PluginWorkspaceIoWorkerQueueErrorKind::TooManyBatches
    );
    assert_eq!(error.batch().identity(), "workspace");
    assert_eq!(error.batch().request_count(), 1);
    let event = error
        .queue_error()
        .operational_event()
        .expect("queue event");
    assert_eq!(event.identity(), Some("workspace"));
    assert_eq!(
        event.to_string(),
        "plugin \"workspace\" saturated workspace-io queue at 1"
    );
    assert_eq!(
        event.kind(),
        crate::plugin::PluginOperationalEventKind::QueueSaturated {
            queue: PluginOperationalQueue::WorkspaceIo,
            limit: count_limit(1),
        }
    );
    assert_eq!(queue.len(), 1);
}

#[test]
fn workspace_worker_ignores_empty_sealed_batches() {
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::from_validated_limit(
        PluginWorkspaceIoWorkerQueueLimit::try_new(1).expect("limit"),
    );

    let empty = empty_workspace_io_batch("workspace");
    assert!(empty.is_empty());
    empty
        .enqueue(&mut queue)
        .expect("empty batch should be accepted as a no-op");
    assert_eq!(queue.len(), 0);

    sealed_write_batch(&host, "workspace", "docs/kept.txt", b"kept")
        .enqueue(&mut queue)
        .expect("non-empty batch should fit");
    empty_workspace_io_batch("workspace")
        .enqueue(&mut queue)
        .expect("empty batch should not saturate a full worker queue");
    assert_eq!(queue.len(), 1);

    let error = sealed_write_batch(
        &host,
        "workspace",
        "docs/secret-overflow.txt",
        b"secret overflow bytes",
    )
    .enqueue(&mut queue)
    .expect_err("non-empty batch should still observe queue capacity");
    assert_eq!(error.batch().request_count(), 1);
    let debug = format!("{error:?}");
    assert!(!debug.contains("secret-overflow"));
    assert!(!debug.contains("secret overflow bytes"));
}

#[test]
fn sealed_workspace_io_worker_enqueue_error_can_retry_batch() {
    let root = temp_dir("workspace-worker-retry");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::from_validated_limit(
        PluginWorkspaceIoWorkerQueueLimit::try_new(1).expect("limit"),
    );
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/one.txt",
            b"one",
        ))
        .expect("first batch");

    let error = sealed_write_batch(
        &host,
        "workspace",
        "docs/secret-retry-io.txt",
        b"secret retry bytes",
    )
    .enqueue(&mut queue)
    .expect_err("full queue should keep the batch retryable");

    assert_eq!(
        error.queue_error(),
        &PluginWorkspaceIoWorkerQueueError::TooManyBatches {
            identity: plugin_identity("workspace"),
            limit: count_limit(1),
        }
    );
    assert_eq!(error.batch().request_count(), 1);
    let debug = format!("{error:?}");
    assert!(!debug.contains("secret-retry-io"));
    assert!(!debug.contains("secret retry bytes"));

    let first = queue
        .execute_next(&filesystem)
        .expect("first batch should execute");
    assert_eq!(
        first.completions()[0].workspace_relative_path(),
        "docs/one.txt"
    );

    let parts = error.into_parts();
    assert_eq!(
        parts.queue_error,
        PluginWorkspaceIoWorkerQueueError::TooManyBatches {
            identity: plugin_identity("workspace"),
            limit: count_limit(1),
        }
    );
    let retry = parts.batch;
    retry
        .enqueue(&mut queue)
        .expect("returned sealed batch should retry after capacity frees");
    let reports = queue.execute_all(&filesystem);

    assert_eq!(reports.len(), 1);
    assert_eq!(
        reports[0].completions()[0].workspace_relative_path(),
        "docs/secret-retry-io.txt"
    );
    assert_eq!(
        fs::read_to_string(root.join("docs/secret-retry-io.txt")).expect("retry"),
        "secret retry bytes"
    );

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn sealed_workspace_io_worker_enqueue_error_can_discard_retry_batch() {
    let root = temp_dir("workspace-worker-discard-retry");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::from_validated_limit(
        PluginWorkspaceIoWorkerQueueLimit::try_new(1).expect("limit"),
    );
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/kept.txt",
            b"kept",
        ))
        .expect("first batch");

    let error = queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/secret-discarded-retry-io.txt",
            b"secret discarded retry bytes",
        ))
        .expect_err("full queue should keep the batch retryable");
    let report = error.discard_batch();
    let reports = queue.execute_all(&filesystem);

    assert_eq!(report.identity(), "workspace");
    assert_eq!(report.discarded_requests(), 1);
    assert_eq!(reports.len(), 1);
    assert_eq!(
        reports[0].completions()[0].workspace_relative_path(),
        "docs/kept.txt"
    );
    assert!(root.join("docs/kept.txt").exists());
    assert!(!root.join("docs/secret-discarded-retry-io.txt").exists());

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_worker_cancel_all_leaves_retry_batch_caller_owned() {
    let root = temp_dir("workspace-worker-cancel-all-retry");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::from_validated_limit(
        PluginWorkspaceIoWorkerQueueLimit::try_new(1).expect("limit"),
    );
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/queued-cancel-all.txt",
            b"queued",
        ))
        .expect("queued batch should fit");
    let error = queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/secret-retry-after-cancel-io.txt",
            b"secret retry bytes",
        ))
        .expect_err("full queue should keep the batch retryable");

    let reports = queue.cancel_all();

    assert_eq!(reports.len(), 1);
    assert_eq!(reports[0].identity(), "workspace");
    assert_eq!(reports[0].canceled_batches(), 1);
    assert_eq!(reports[0].canceled_requests(), 1);
    assert!(queue.is_empty());
    assert_eq!(error.batch().request_count(), 1);
    let debug = format!("{error:?}");
    assert!(!debug.contains("secret-retry-after-cancel-io"));
    assert!(!debug.contains("secret retry bytes"));

    error
        .into_batch()
        .enqueue(&mut queue)
        .expect("caller-owned retry batch should still enter the queue");
    let executed = queue.execute_all(&filesystem);

    assert_eq!(executed.len(), 1);
    assert_eq!(
        executed[0].completions()[0].workspace_relative_path(),
        "docs/secret-retry-after-cancel-io.txt"
    );
    assert!(!root.join("docs/queued-cancel-all.txt").exists());
    assert_eq!(
        fs::read_to_string(root.join("docs/secret-retry-after-cancel-io.txt"))
            .expect("retry should execute"),
        "secret retry bytes"
    );

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_worker_cancels_revoked_identity_before_filesystem_work() {
    let root = temp_dir("workspace-worker-cancel");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::default();
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/canceled.txt",
            b"no write",
        ))
        .expect("batch");

    let report = queue.cancel_identity(&plugin_identity("workspace"));
    let reports = queue.execute_all(&filesystem);

    assert_eq!(report.identity(), "workspace");
    assert_eq!(report.identity_proof().as_str(), "workspace");
    assert_eq!(report.canceled_batches(), 1);
    assert_eq!(report.canceled_requests(), 1);
    assert!(reports.is_empty());
    assert!(!root.join("docs/canceled.txt").exists());

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_worker_cancel_all_reports_by_identity_without_filesystem_work() {
    let root = temp_dir("workspace-worker-cancel-all");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let other_host = workspace_host_for("other");
    let mut queue = PluginWorkspaceIoWorkerQueue::default();
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/secret-cancel-all-one.txt",
            b"secret canceled one",
        ))
        .expect("first batch");
    queue
        .push(sealed_write_batch(
            &other_host,
            "other",
            "docs/other-cancel-all.txt",
            b"other canceled",
        ))
        .expect("second batch");
    queue
        .push(sealed_write_batch(
            &host,
            "workspace",
            "docs/secret-cancel-all-two.txt",
            b"secret canceled two",
        ))
        .expect("third batch");

    let reports = queue.cancel_all();
    let executed = queue.execute_all(&filesystem);

    assert_eq!(reports.len(), 2);
    assert_eq!(reports[0].identity(), "workspace");
    assert_eq!(reports[0].canceled_batches(), 2);
    assert_eq!(reports[0].canceled_requests(), 2);
    assert_eq!(reports[1].identity(), "other");
    assert_eq!(reports[1].canceled_batches(), 1);
    assert_eq!(reports[1].canceled_requests(), 1);
    assert!(executed.is_empty());
    assert!(queue.is_empty());
    assert!(!root.join("docs/secret-cancel-all-one.txt").exists());
    assert!(!root.join("docs/secret-cancel-all-two.txt").exists());
    assert!(!root.join("docs/other-cancel-all.txt").exists());
    let debug = format!("{reports:?}");
    assert!(!debug.contains("secret-cancel-all"));
    assert!(!debug.contains("secret canceled"));
    assert!(!debug.contains(root.to_string_lossy().as_ref()));

    let _cleanup = fs::remove_dir_all(root);
}

#[test]
fn workspace_worker_executes_completions_in_fifo_order() {
    let root = temp_dir("workspace-worker-fifo");
    fs::create_dir(root.join("docs")).expect("docs directory should be created");
    let filesystem = config(&root);
    let host = workspace_host();
    let mut queue = PluginWorkspaceIoWorkerQueue::default();
    queue
        .push(sealed_write_batch(&host, "workspace", "docs/a.txt", b"a"))
        .expect("first batch");
    queue
        .push(sealed_write_batch(&host, "workspace", "docs/b.txt", b"b"))
        .expect("second batch");

    let reports = queue.execute_all(&filesystem);

    assert_eq!(reports.len(), 2);
    assert_eq!(
        reports[0].completions()[0].workspace_relative_path(),
        "docs/a.txt"
    );
    assert_eq!(
        reports[1].completions()[0].workspace_relative_path(),
        "docs/b.txt"
    );
    assert_eq!(fs::read_to_string(root.join("docs/a.txt")).expect("a"), "a");
    assert_eq!(fs::read_to_string(root.join("docs/b.txt")).expect("b"), "b");

    let _cleanup = fs::remove_dir_all(root);
}

fn temp_dir_path(name: &str) -> PathBuf {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system time should be after unix epoch")
        .as_nanos();
    std::env::temp_dir().join(format!(
        "alma-plugin-host-{name}-{}-{nanos}",
        std::process::id()
    ))
}

fn temp_dir(name: &str) -> PathBuf {
    let path = temp_dir_path(name);
    fs::create_dir(&path).expect("temporary directory should be created");
    path
}

fn config(root: &Path) -> FilesystemConfig {
    FilesystemConfig {
        workspace_root: root.canonicalize().expect("root should canonicalize"),
        max_file_bytes: 128,
    }
}

fn workspace_host() -> PluginHostContext {
    workspace_host_for("workspace")
}

fn workspace_host_for(identity: &str) -> PluginHostContext {
    PluginHostContext::for_test(
        identity,
        PluginCapabilities {
            workspace_observe: vec![WorkspacePathGrant::new("docs")],
            workspace_artifact_write: vec![WorkspacePathGrant::new("docs")],
            ..PluginCapabilities::default()
        },
    )
}

fn plugin_identity(identity: &str) -> PluginIdentity {
    PluginIdentity::try_new(identity).expect("test identity should be valid")
}

fn read_limit(max_read_bytes: u64) -> PluginWorkspaceReadByteLimit {
    PluginWorkspaceReadByteLimit::try_new(max_read_bytes)
        .expect("test read limit should be non-zero")
}

fn write_limit(max_write_bytes: u64) -> PluginWorkspaceWriteByteLimit {
    PluginWorkspaceWriteByteLimit::try_new(max_write_bytes)
        .expect("test write limit should be non-zero")
}

fn sealed_write_batch(
    host: &PluginHostContext,
    identity: &str,
    path: &str,
    bytes: &[u8],
) -> SealedPluginWorkspaceIoBatch {
    let mut batch = PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity(identity),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget"),
    );
    let access = host
        .authorize_workspace_artifact_write(WorkspacePathRef::try_from(path).expect("path"))
        .expect("write");
    batch.push_write(access, bytes.to_vec()).expect("push");
    batch.seal()
}

fn empty_workspace_io_batch(identity: &str) -> SealedPluginWorkspaceIoBatch {
    PendingPluginWorkspaceIoBatch::from_validated_budget(
        plugin_identity(identity),
        PluginWorkspaceIoBudget::try_new(4, 64, 64).expect("budget"),
    )
    .seal()
}