alien-commands 2.0.1

//! Integration tests for alien-commands
//!
//! These tests focus on the 8 core command scenarios:
//! (PUSH, PULL) × (SMALL PARAMS, LARGE PARAMS) × (SMALL RESPONSE, LARGE RESPONSE)
//!
//! 1. (Push, Small Params, Small Response): inline params auto-dispatched, inline response
//! 2. (Push, Small Params, Large Response): inline params auto-dispatched, storage response
//! 3. (Push, Large Params, Small Response): storage params auto-dispatched after upload, inline response
//! 4. (Push, Large Params, Large Response): storage params auto-dispatched after upload, storage response
//! 5. (Pull, Small Params, Small Response): inline params acquired via lease, inline response
//! 6. (Pull, Small Params, Large Response): inline params acquired via lease, storage response
//! 7. (Pull, Large Params, Small Response): storage params acquired via lease after upload, inline response
//! 8. (Pull, Large Params, Large Response): storage params acquired via lease after upload, storage response
//!
//! Additional component tests verify basic API functionality and runtime integration.

#[cfg(feature = "test-utils")]
mod tests {
    use std::time::Duration;

    use alien_commands::{
        runtime::{decode_params, parse_envelope},
        test_utils::{
            dispatcher::MockDispatcherAssertions, server::TestCommandServerAssertions, *,
        },
        types::*,
    };
    use alien_core::{MessagePayload, QueueMessage};
    use chrono::Utc;

    // ===========================================
    // CORE SCENARIOS: (PUSH, PULL) × (SMALL PARAMS, LARGE PARAMS) × (SMALL RESPONSE, LARGE RESPONSE)
    // ===========================================

    /// Core Scenario 1: Push + Small Params + Small Response
    /// Inline params auto-dispatched, inline response
    #[tokio::test]
    async fn test_core_push_small_params_small_response() {
        let server = TestCommandServer::new().await;

        // 1. Client creates small inline command (auto-dispatched immediately)
        let request = test_inline_create_command("push-agent", "generate-report");
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::Dispatched); // Auto-dispatched
        assert!(response.storage_upload.is_none());

        // 2. Verify envelope was dispatched to mock dispatcher (simulates push to agent)
        let mock_dispatcher = server
            .mock_dispatcher()
            .expect("Should have mock dispatcher");
        mock_dispatcher.assert_has_dispatched().await;
        let dispatched = mock_dispatcher.get_latest().await.unwrap();
        assert_eq!(dispatched.envelope.command_id, response.command_id);
        assert_eq!(dispatched.envelope.command, "generate-report");
        assert!(matches!(
            dispatched.envelope.params,
            BodySpec::Inline { .. }
        ));

        // 3. Simulate agent receiving envelope and processing
        let params = decode_params(&dispatched.envelope).await.unwrap();
        assert!(params.is_object()); // Should have JSON params

        // 4. Agent submits response
        let agent_response = test_success_response(b"report generated");
        server
            .submit_command_response(&response.command_id, agent_response)
            .await
            .unwrap();

        // 5. Client polls for completion
        let final_status = server
            .wait_for_completion(&response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let final_response = final_status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            assert_inline_body(&body, b"report generated");
        }
    }

    /// Core Scenario 5: Pull + Small Params + Small Response
    /// Inline params acquired via lease, inline response
    #[tokio::test]
    async fn test_core_pull_small_params_small_response() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // 1. Client creates command (in Pull mode, stays Pending until lease)
        let request = test_inline_create_command("pull-agent", "sync-data");
        let create_response = server.create_command(request).await.unwrap();
        assert_eq!(create_response.state, CommandState::Pending); // Pending until lease

        // 2. Agent polls for lease (moves to Dispatched)
        let lease = server
            .acquire_single_lease("pull-agent")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(lease.command_id, create_response.command_id);
        assert!(matches!(lease.envelope.params, BodySpec::Inline { .. }));

        // Verify state moved to Dispatched after lease
        let status = server
            .get_command_status(&create_response.command_id)
            .await
            .unwrap();
        assert_eq!(status.state, CommandState::Dispatched);

        // 3. Agent processes envelope (simulated)
        let params = decode_params(&lease.envelope).await.unwrap();
        assert!(params.is_object()); // Should have JSON params

        // 4. Agent submits response
        let agent_response = test_json_success_response(&serde_json::json!({
            "status": "synced",
            "command_id": lease.command_id
        }));
        server
            .submit_command_response(&lease.command_id, agent_response)
            .await
            .unwrap();

        // 5. Client checks completion
        let final_status = server
            .wait_for_completion(&create_response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let response = final_status.response.unwrap();
        assert!(response.is_success());
        if let CommandResponse::Success { response: body } = response {
            let body_data = body.decode_inline().unwrap();
            let json: serde_json::Value = serde_json::from_slice(&body_data).unwrap();
            assert_eq!(json["status"], "synced");
        }
    }

    /// Core Scenario 4: Push + Large Params + Large Response
    /// Storage params auto-dispatched after upload, storage response
    #[tokio::test]
    async fn test_core_push_large_params_large_response() {
        let server = TestCommandServer::new().await; // Use default 150KB limit

        // 1. Client creates large command
        let large_params = vec![b'X'; 160000]; // 160KB > 150KB inline limit
        let request = test_storage_create_command("push-agent", "process-bulk", large_params.len());
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::PendingUpload);
        assert!(response.storage_upload.is_some());

        // 2. Client uploads large params using the presigned URL mechanism
        let storage_upload = response.storage_upload.unwrap();
        storage_upload
            .put_request
            .execute(Some(large_params.clone().into()))
            .await
            .unwrap();

        let upload_complete = test_upload_complete_request(160000);
        server
            .upload_complete(&response.command_id, upload_complete)
            .await
            .unwrap();

        // 3. Command should be auto-dispatched after upload
        let mock_dispatcher = server
            .mock_dispatcher()
            .expect("Should have mock dispatcher");
        assert!(mock_dispatcher.has_dispatched().await);
        let dispatched = mock_dispatcher.get_latest().await.unwrap();
        assert_eq!(dispatched.envelope.command_id, response.command_id);
        assert!(matches!(
            dispatched.envelope.params,
            BodySpec::Storage { .. }
        ));

        // 4. Agent submits large response (> 150KB to force storage)
        let large_response_data = vec![b'R'; 160000]; // 160KB > 150KB inline limit

        // Agent uses the presigned upload request from the envelope
        dispatched
            .envelope
            .response_handling
            .storage_upload_request
            .execute(Some(large_response_data.clone().into()))
            .await
            .unwrap();

        let agent_response = CommandResponse::success_storage(large_response_data.len() as u64);
        server
            .submit_command_response(&response.command_id, agent_response)
            .await
            .unwrap();

        // 5. Client gets final result with large storage response
        let final_status = server
            .wait_for_completion(&response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let final_response = final_status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            assert_storage_body(&body, Some(160000));
            // Verify we can download and the content matches what the agent uploaded
            assert_storage_body_content(&body, &large_response_data).await;
        }

        // Should have storage objects from both large params and response
        assert!(server.storage_object_count().await > 1);
    }

    /// Core Scenario 8: Pull + Large Params + Large Response
    /// Storage params acquired via lease after upload, storage response
    #[tokio::test]
    async fn test_core_pull_large_params_large_response() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // 1. Client creates large command
        let large_params = vec![b'Y'; 160000]; // 160KB > 150KB inline limit
        let request = test_storage_create_command("pull-agent", "bulk-process", large_params.len());
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::PendingUpload);
        assert!(response.storage_upload.is_some());

        // 2. Client uploads large params using the presigned URL mechanism
        let storage_upload = response.storage_upload.unwrap();
        storage_upload
            .put_request
            .execute(Some(large_params.clone().into()))
            .await
            .unwrap();

        let upload_complete = test_upload_complete_request(160000);
        let upload_response = server
            .upload_complete(&response.command_id, upload_complete)
            .await
            .unwrap();
        // In Pull mode, after upload the state is Pending (waiting for lease)
        assert_eq!(upload_response.state, CommandState::Pending);

        // 3. Agent polls for lease (moves to Dispatched)
        let lease = server
            .acquire_single_lease("pull-agent")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(lease.command_id, response.command_id);
        assert!(matches!(lease.envelope.params, BodySpec::Storage { .. }));

        // Verify state moved to Dispatched after lease
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        assert_eq!(status.state, CommandState::Dispatched);

        // 4. Agent simulates processing large params
        let params_bytes = alien_commands::runtime::decode_params_bytes(&lease.envelope)
            .await
            .unwrap();
        assert_eq!(params_bytes.len(), 160000); // Should have reconstructed the large params
        assert_eq!(params_bytes, vec![b'Y'; 160000]); // Should match original data

        // 5. Agent submits large response (upload to storage since > 150KB)
        let large_response_data = vec![b'Z'; 160000]; // 160KB > 150KB inline limit

        // Agent uses the presigned upload request from the envelope
        lease
            .envelope
            .response_handling
            .storage_upload_request
            .execute(Some(large_response_data.clone().into()))
            .await
            .unwrap();

        let agent_response = CommandResponse::success_storage(large_response_data.len() as u64);
        server
            .submit_command_response(&lease.command_id, agent_response)
            .await
            .unwrap();

        // 6. Verify completion with storage response
        let final_status = server
            .wait_for_completion(&response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let final_response = final_status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            assert_storage_body(&body, Some(160000));
            // Verify we can download and the content matches what the agent uploaded
            assert_storage_body_content(&body, &large_response_data).await;
        }

        // Should have storage objects from the large payloads
        assert!(server.storage_object_count().await > 0);
    }

    /// Core Scenario 2: Push + Small Params + Large Response
    /// Inline params auto-dispatched, storage response
    #[tokio::test]
    async fn test_core_push_small_params_large_response() {
        let server = TestCommandServer::new().await; // Use default 150KB limit

        // 1. Client creates small inline command (auto-dispatched immediately)
        let request = test_inline_create_command("push-agent", "generate-large-report");
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::Dispatched); // Auto-dispatched
        assert!(response.storage_upload.is_none());

        // 2. Verify envelope was dispatched to mock dispatcher (simulates push to agent)
        let mock_dispatcher = server
            .mock_dispatcher()
            .expect("Should have mock dispatcher");
        mock_dispatcher.assert_has_dispatched().await;
        let dispatched = mock_dispatcher.get_latest().await.unwrap();
        assert_eq!(dispatched.envelope.command_id, response.command_id);
        assert!(matches!(
            dispatched.envelope.params,
            BodySpec::Inline { .. }
        ));

        // 3. Agent submits large response (> 150KB to force storage)
        let large_response_data = vec![b'L'; 160000]; // 160KB > 150KB inline limit

        // Agent uses the presigned upload request from the envelope
        dispatched
            .envelope
            .response_handling
            .storage_upload_request
            .execute(Some(large_response_data.clone().into()))
            .await
            .unwrap();

        let agent_response = CommandResponse::success_storage(large_response_data.len() as u64);
        server
            .submit_command_response(&response.command_id, agent_response)
            .await
            .unwrap();

        // 4. Client gets final result with large storage response
        let final_status = server
            .wait_for_completion(&response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let final_response = final_status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            assert_storage_body(&body, Some(160000));
            // Verify we can download and the content matches what we uploaded
            assert_storage_body_content(&body, &large_response_data).await;
        }

        // Should have storage object from large response
        assert!(server.storage_object_count().await > 0);
    }

    /// Core Scenario 3: Push + Large Params + Small Response
    /// Storage params auto-dispatched after upload, inline response
    #[tokio::test]
    async fn test_core_push_large_params_small_response() {
        let server = TestCommandServer::new().await; // Use default 150KB limit

        // 1. Client creates large command
        let large_params = vec![b'X'; 160000]; // 160KB > 150KB inline limit
        let request = test_storage_create_command("push-agent", "process-data", large_params.len());
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::PendingUpload);
        assert!(response.storage_upload.is_some());

        // 2. Client uploads large params using the presigned URL mechanism
        let storage_upload = response.storage_upload.unwrap();
        storage_upload
            .put_request
            .execute(Some(large_params.clone().into()))
            .await
            .unwrap();

        let upload_complete = test_upload_complete_request(160000);
        server
            .upload_complete(&response.command_id, upload_complete)
            .await
            .unwrap();

        // 3. Command should be auto-dispatched after upload
        let mock_dispatcher = server
            .mock_dispatcher()
            .expect("Should have mock dispatcher");
        assert!(mock_dispatcher.has_dispatched().await);
        let dispatched = mock_dispatcher.get_latest().await.unwrap();
        assert_eq!(dispatched.envelope.command_id, response.command_id);
        assert!(matches!(
            dispatched.envelope.params,
            BodySpec::Storage { .. }
        ));

        // 4. Agent submits small inline response
        let agent_response = test_success_response(b"ok");
        server
            .submit_command_response(&response.command_id, agent_response)
            .await
            .unwrap();

        // 5. Client gets final result with inline response
        let final_status = server
            .wait_for_completion(&response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let final_response = final_status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            assert_inline_body(&body, b"ok");
        }

        // Should have storage object from large params but not response
        assert!(server.storage_object_count().await > 0);
    }

    /// Core Scenario 6: Pull + Small Params + Large Response
    /// Inline params acquired via lease, storage response
    #[tokio::test]
    async fn test_core_pull_small_params_large_response() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // 1. Client creates command (in Pull mode, stays Pending until lease)
        let request = test_inline_create_command("pull-agent", "generate-large");
        let create_response = server.create_command(request).await.unwrap();
        assert_eq!(create_response.state, CommandState::Pending); // Pending until lease

        // 2. Agent polls for lease (moves to Dispatched)
        let lease = server
            .acquire_single_lease("pull-agent")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(lease.command_id, create_response.command_id);
        assert!(matches!(lease.envelope.params, BodySpec::Inline { .. }));

        // Verify state moved to Dispatched after lease
        let status = server
            .get_command_status(&create_response.command_id)
            .await
            .unwrap();
        assert_eq!(status.state, CommandState::Dispatched);

        // 3. Agent processes envelope (simulated)
        let params = decode_params(&lease.envelope).await.unwrap();
        assert!(params.is_object()); // Should have JSON params

        // 4. Agent submits large response (> 150KB to force storage)
        let large_response_data = vec![b'M'; 160000]; // 160KB > 150KB inline limit

        // Agent uses the presigned upload request from the envelope
        lease
            .envelope
            .response_handling
            .storage_upload_request
            .execute(Some(large_response_data.clone().into()))
            .await
            .unwrap();

        let agent_response = CommandResponse::success_storage(large_response_data.len() as u64);
        server
            .submit_command_response(&lease.command_id, agent_response)
            .await
            .unwrap();

        // 5. Client checks completion with large storage response
        let final_status = server
            .wait_for_completion(&create_response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let response = final_status.response.unwrap();
        assert!(response.is_success());
        if let CommandResponse::Success { response: body } = response {
            assert_storage_body(&body, Some(160000));
            // Verify we can download and the content matches what the agent uploaded
            assert_storage_body_content(&body, &large_response_data).await;
        }

        // Should have storage object from large response
        assert!(server.storage_object_count().await > 0);
    }

    /// Core Scenario 7: Pull + Large Params + Small Response
    /// Storage params acquired via lease after upload, inline response
    #[tokio::test]
    async fn test_core_pull_large_params_small_response() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // 1. Client creates large command
        let large_params = vec![b'Y'; 160000]; // 160KB > 150KB inline limit
        let request = test_storage_create_command("pull-agent", "process-bulk", large_params.len());
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::PendingUpload);
        assert!(response.storage_upload.is_some());

        // 2. Client uploads large params using the presigned URL mechanism
        let storage_upload = response.storage_upload.unwrap();
        storage_upload
            .put_request
            .execute(Some(large_params.clone().into()))
            .await
            .unwrap();

        let upload_complete = test_upload_complete_request(160000);
        let upload_response = server
            .upload_complete(&response.command_id, upload_complete)
            .await
            .unwrap();
        // In Pull mode, after upload the state is Pending (waiting for lease)
        assert_eq!(upload_response.state, CommandState::Pending);

        // 3. Agent polls for lease (moves to Dispatched)
        let lease = server
            .acquire_single_lease("pull-agent")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(lease.command_id, response.command_id);
        assert!(matches!(lease.envelope.params, BodySpec::Storage { .. }));

        // Verify state moved to Dispatched after lease
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        assert_eq!(status.state, CommandState::Dispatched);

        // 4. Agent simulates processing large params
        let params_bytes = alien_commands::runtime::decode_params_bytes(&lease.envelope)
            .await
            .unwrap();
        assert_eq!(params_bytes.len(), 160000); // Should have reconstructed the large params
        assert_eq!(params_bytes, vec![b'Y'; 160000]); // Should match original data

        // 5. Agent submits small inline response
        let agent_response = test_json_success_response(&serde_json::json!({
            "status": "processed",
            "command_id": lease.command_id
        }));
        server
            .submit_command_response(&lease.command_id, agent_response)
            .await
            .unwrap();

        // 6. Verify completion with inline response
        let final_status = server
            .wait_for_completion(&response.command_id, Duration::from_secs(5))
            .await
            .unwrap();
        assert_eq!(final_status.state, CommandState::Succeeded);

        let final_response = final_status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            let body_data = body.decode_inline().unwrap();
            let json: serde_json::Value = serde_json::from_slice(&body_data).unwrap();
            assert_eq!(json["status"], "processed");
        }

        // Should have storage object from large params but not response
        assert!(server.storage_object_count().await > 0);
    }

    // ===============================================
    // TARGET ROUTING
    // ===============================================

    /// Status responses and lease envelopes carry the resolved target
    /// (single-target shorthand: no targetResourceId in the request).
    #[tokio::test]
    async fn test_status_and_envelope_carry_resolved_target() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        let request = test_inline_create_command("target-agent", "targeted-command");
        let response = server.create_command(request).await.unwrap();

        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        assert_eq!(status.target, server.default_target);

        let lease = server
            .acquire_single_lease("target-agent")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(lease.envelope.target, server.default_target);
    }

    /// An explicitly requested target that doesn't exist is rejected with the
    /// stable COMMAND_TARGET_NOT_FOUND code.
    #[tokio::test]
    async fn test_create_with_unknown_target_rejected() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        let mut request = test_inline_create_command("target-agent", "targeted-command");
        request.target_resource_id = Some("no-such-resource".to_string());

        let err = server.create_command(request).await.unwrap_err();
        assert_eq!(err.code, "COMMAND_TARGET_NOT_FOUND");
    }

    /// With two registered targets, shorthand creation (no targetResourceId)
    /// is rejected with the stable COMMAND_TARGET_AMBIGUOUS code.
    #[tokio::test]
    async fn test_create_shorthand_with_two_targets_ambiguous() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;
        server
            .registry
            .register_target("second-daemon", CommandTargetType::Daemon)
            .await
            .unwrap();

        let request = test_inline_create_command("target-agent", "targeted-command");
        let err = server.create_command(request).await.unwrap_err();
        assert_eq!(err.code, "COMMAND_TARGET_AMBIGUOUS");
    }

    /// Each target leases only its own commands: two targets, two commands,
    /// each lease scan returns only the requester's command.
    #[tokio::test]
    async fn test_lease_scans_only_requesting_targets_prefix() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;
        server
            .registry
            .register_target("second-daemon", CommandTargetType::Daemon)
            .await
            .unwrap();
        let second_target = CommandTarget::new("second-daemon", CommandTargetType::Daemon);

        // Command for the default target.
        let mut request_a = test_inline_create_command("target-agent", "for-default");
        request_a.target_resource_id = Some(server.default_target.resource_id.clone());
        let cmd_a = server.create_command(request_a).await.unwrap();

        // Command for the second target.
        let mut request_b = test_inline_create_command("target-agent", "for-second");
        request_b.target_resource_id = Some("second-daemon".to_string());
        let cmd_b = server.create_command(request_b).await.unwrap();

        // Default target leases only its own command, even asking for many.
        let default_leases = server
            .acquire_lease(
                "target-agent",
                LeaseRequest {
                    deployment_id: "target-agent".to_string(),
                    target: server.default_target.clone(),
                    max_leases: 10,
                    lease_seconds: 60,
                },
            )
            .await
            .unwrap();
        assert_eq!(default_leases.leases.len(), 1);
        assert_eq!(default_leases.leases[0].command_id, cmd_a.command_id);
        assert_eq!(
            default_leases.leases[0].envelope.target,
            server.default_target
        );

        // Second target leases only its own command.
        let second_leases = server
            .acquire_lease(
                "target-agent",
                LeaseRequest {
                    deployment_id: "target-agent".to_string(),
                    target: second_target.clone(),
                    max_leases: 10,
                    lease_seconds: 60,
                },
            )
            .await
            .unwrap();
        assert_eq!(second_leases.leases.len(), 1);
        assert_eq!(second_leases.leases[0].command_id, cmd_b.command_id);
        assert_eq!(second_leases.leases[0].envelope.target, second_target);
    }

    /// Expiry-driven redelivery: a leased command whose lease TTL runs out
    /// WITHOUT a response (crashed receiver, network partition — no explicit
    /// release) must become leasable again, and the redelivered envelope must
    /// carry an incremented attempt so handlers can detect redelivery
    /// (`ctx.attempt > 1` is the documented at-least-once signal on both
    /// receiver twins).
    #[tokio::test]
    async fn test_expired_lease_redelivers_with_incremented_attempt() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        let mut request = test_inline_create_command("expiry-agent", "expiry-redelivery");
        request.target_resource_id = Some(server.default_target.resource_id.clone());
        let created = server.create_command(request).await.unwrap();

        let lease_request = LeaseRequest {
            deployment_id: "expiry-agent".to_string(),
            target: server.default_target.clone(),
            max_leases: 1,
            lease_seconds: 1,
        };

        // First lease: attempt 1. No response is ever submitted.
        let first = server
            .acquire_lease("expiry-agent", lease_request.clone())
            .await
            .unwrap();
        assert_eq!(first.leases.len(), 1);
        assert_eq!(first.leases[0].command_id, created.command_id);
        assert_eq!(first.leases[0].attempt, 1);

        // While the lease is live, the command must NOT be re-leasable.
        let while_held = server
            .acquire_lease("expiry-agent", lease_request.clone())
            .await
            .unwrap();
        assert!(
            while_held.leases.is_empty(),
            "a live lease must not be double-leased"
        );

        // Let the 1s lease TTL expire without any response.
        tokio::time::sleep(std::time::Duration::from_millis(2500)).await;

        // Redelivery: same command, incremented attempt.
        let second = server
            .acquire_lease("expiry-agent", lease_request)
            .await
            .unwrap();
        assert_eq!(
            second.leases.len(),
            1,
            "an expired lease must make the command leasable again"
        );
        assert_eq!(second.leases[0].command_id, created.command_id);
        assert_eq!(
            second.leases[0].attempt, 2,
            "expiry-driven redelivery must increment the attempt"
        );
        assert_eq!(
            second.leases[0].envelope.attempt, 2,
            "the redelivered envelope must carry the incremented attempt"
        );
    }

    /// Lease-served envelopes carry manager URLs relative to the exact lease
    /// endpoint. The pull consumer resolves them against the endpoint it
    /// reached, preserving both a network-corrected origin and any reverse-
    /// proxy prefix. Signed query parameters must survive untouched.
    #[tokio::test]
    async fn test_leased_envelope_manager_urls_are_path_relative() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        let mut request = test_inline_create_command("relative-agent", "relative-urls");
        request.target_resource_id = Some(server.default_target.resource_id.clone());
        let created = server.create_command(request).await.unwrap();

        let leases = server
            .acquire_lease(
                "relative-agent",
                LeaseRequest {
                    deployment_id: "relative-agent".to_string(),
                    target: server.default_target.clone(),
                    max_leases: 1,
                    lease_seconds: 60,
                },
            )
            .await
            .unwrap();
        assert_eq!(leases.leases.len(), 1);
        let envelope = &leases.leases[0].envelope;

        let submit = &envelope.response_handling.submit_response_url;
        assert!(
            submit.starts_with(&format!("{}/response?", created.command_id)),
            "submit URL must be relative to the lease endpoint, got '{submit}'"
        );
        assert!(
            submit.contains("response_token="),
            "relativization must preserve the signed query, got '{submit}'"
        );
    }

    /// Defense-in-depth: a pending-index entry under target A's prefix whose
    /// stored command metadata says target B is corruption — the lease call
    /// must fail loudly instead of misdelivering the command.
    #[tokio::test]
    async fn test_lease_target_mismatch_in_pending_index_is_loud_error() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;
        server
            .registry
            .register_target("second-daemon", CommandTargetType::Daemon)
            .await
            .unwrap();

        // Create a command registered to second-daemon.
        let mut request = test_inline_create_command("target-agent", "for-second");
        request.target_resource_id = Some("second-daemon".to_string());
        let cmd = server.create_command(request).await.unwrap();

        // Corrupt the index: plant the command under the DEFAULT target's prefix.
        let corrupt_key = format!(
            "target:target-agent:{}:pending:{}:{}",
            server.default_target.resource_id,
            chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            cmd.command_id
        );
        alien_bindings::traits::Kv::put(server.kv.as_ref(), &corrupt_key, vec![], None)
            .await
            .unwrap();

        // Leasing as the default target must fail loudly, not deliver the command.
        let result = server
            .acquire_lease(
                "target-agent",
                LeaseRequest {
                    deployment_id: "target-agent".to_string(),
                    target: server.default_target.clone(),
                    max_leases: 1,
                    lease_seconds: 60,
                },
            )
            .await;
        let err = result.unwrap_err();
        assert!(
            err.message.contains(&cmd.command_id) || err.message.contains("target"),
            "expected loud target-mismatch error, got: {}",
            err.message
        );
    }

    /// Idempotency keys are scoped per target: same key on two different
    /// targets creates two distinct commands; same key + same target replays
    /// the same command.
    #[tokio::test]
    async fn test_idempotency_scoped_per_target() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;
        server
            .registry
            .register_target("second-daemon", CommandTargetType::Daemon)
            .await
            .unwrap();

        let make_request = |target: &str| {
            let mut request = test_inline_create_command("target-agent", "idem-command");
            request.target_resource_id = Some(target.to_string());
            request.idempotency_key = Some("same-key".to_string());
            request
        };

        let default_id = server.default_target.resource_id.clone();
        let first = server
            .create_command(make_request(&default_id))
            .await
            .unwrap();
        let second = server
            .create_command(make_request("second-daemon"))
            .await
            .unwrap();
        // Same key, different target: distinct commands.
        assert_ne!(first.command_id, second.command_id);

        // Same key, same target: replays the same command.
        let replay = server
            .create_command(make_request(&default_id))
            .await
            .unwrap();
        assert_eq!(replay.command_id, first.command_id);
    }

    /// One deployment, two command-capable targets of different types
    /// (Worker + Daemon) sharing the exact same command name: the Worker's
    /// command routes Push (mock dispatcher receives it), the Daemon's lands
    /// only in the Daemon's own pending index (leasable there, absent from
    /// the Worker's). The two never cross.
    #[tokio::test]
    async fn test_worker_and_daemon_share_command_name_route_independently() {
        // Push-capable dispatcher: the default auto-registered target is a
        // Worker in Push mode (see TestCommandServerBuilder::build).
        let server = TestCommandServer::new().await;
        server
            .registry
            .register_target("shared-daemon", CommandTargetType::Daemon)
            .await
            .unwrap();
        let daemon_target = CommandTarget::new("shared-daemon", CommandTargetType::Daemon);

        // Command addressed to the Worker.
        let mut worker_request = test_inline_create_command("target-agent", "shared-command");
        worker_request.target_resource_id = Some(server.default_target.resource_id.clone());
        let worker_response = server.create_command(worker_request).await.unwrap();
        assert_eq!(worker_response.state, CommandState::Dispatched);

        // Command addressed to the Daemon, same command name.
        let mut daemon_request = test_inline_create_command("target-agent", "shared-command");
        daemon_request.target_resource_id = Some("shared-daemon".to_string());
        let daemon_response = server.create_command(daemon_request).await.unwrap();
        assert_eq!(daemon_response.state, CommandState::Pending);

        // The Worker's command reached the mock dispatcher (push); the
        // Daemon's did not — exactly one dispatch, and it's the Worker's.
        let mock_dispatcher = server
            .mock_dispatcher()
            .expect("Should have mock dispatcher");
        mock_dispatcher.assert_dispatch_count(1).await;
        let dispatched = mock_dispatcher.get_latest().await.unwrap();
        assert_eq!(dispatched.envelope.command_id, worker_response.command_id);
        assert_eq!(dispatched.envelope.target, server.default_target);

        // The Daemon's command sits only in ITS OWN pending index: leasing as
        // the Daemon target returns exactly the Daemon's command.
        let daemon_leases = server
            .acquire_lease(
                "target-agent",
                LeaseRequest {
                    deployment_id: "target-agent".to_string(),
                    target: daemon_target.clone(),
                    max_leases: 10,
                    lease_seconds: 60,
                },
            )
            .await
            .unwrap();
        assert_eq!(daemon_leases.leases.len(), 1);
        assert_eq!(
            daemon_leases.leases[0].command_id,
            daemon_response.command_id
        );
        assert_eq!(daemon_leases.leases[0].envelope.target, daemon_target);

        // Leasing as the Worker target never surfaces the Daemon's command
        // (the Worker's pending index is untouched — its command was pushed,
        // not enqueued, and the Daemon's command was never indexed there).
        let worker_leases = server
            .acquire_lease(
                "target-agent",
                LeaseRequest {
                    deployment_id: "target-agent".to_string(),
                    target: server.default_target.clone(),
                    max_leases: 10,
                    lease_seconds: 60,
                },
            )
            .await
            .unwrap();
        assert_eq!(worker_leases.leases.len(), 0);
    }

    // ===============================================
    // ESSENTIAL COMPONENT TESTS
    // ===============================================

    /// Test basic API operations
    #[tokio::test]
    async fn test_basic_api_operations() {
        let server = TestCommandServer::new().await;

        // Test create command with inline payload
        let request = test_inline_create_command("api-agent", "test-command");
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::Dispatched);
        assert!(response.command_id.starts_with("cmd_"));
        assert!(response.storage_upload.is_none());

        // Test status check
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        assert_eq!(status.command_id, response.command_id);
        assert_eq!(status.state, CommandState::Dispatched);
        assert_eq!(status.attempt, 1);

        // Test create large command requiring storage
        let large_request = test_storage_create_command("storage-agent", "upload-command", 200_000);
        let response = server.create_command(large_request).await.unwrap();
        assert_eq!(response.state, CommandState::PendingUpload);
        assert!(response.storage_upload.is_some());

        // Test upload completion
        let upload_complete = test_upload_complete_request(200_000);
        let complete_response = server
            .upload_complete(&response.command_id, upload_complete)
            .await
            .unwrap();
        assert_eq!(complete_response.state, CommandState::Dispatched);
    }

    /// A transport error is an ambiguous acknowledgement: the target may have
    /// accepted the envelope before the response was lost. The command must
    /// stay Dispatched so a late response remains valid. Creation still returns
    /// the durable command ID, avoiding an unknown orphan and an immediate
    /// duplicate when default clients retry without an idempotency key.
    #[tokio::test]
    async fn ambiguous_push_failure_keeps_dispatched_and_accepts_late_response() {
        let server = TestCommandServer::new().await;
        let dispatcher = server
            .mock_dispatcher()
            .expect("push test server uses the mock dispatcher");
        dispatcher.set_should_fail(true).await;

        let created = server
            .create_command(test_inline_create_command(
                "push-agent",
                "ambiguous-dispatch",
            ))
            .await
            .expect("ambiguous acknowledgement must still return the durable command ID");
        assert_eq!(created.state, CommandState::Dispatched);
        dispatcher.assert_dispatch_count(0).await;

        let status = server
            .get_command_status(&created.command_id)
            .await
            .expect("command status");
        assert_eq!(status.command_id, created.command_id);
        assert_eq!(status.state, CommandState::Dispatched);
        assert_eq!(status.attempt, 1);

        server
            .submit_command_response(
                &created.command_id,
                test_success_response(b"completed after ambiguous acknowledgement"),
            )
            .await
            .expect("late response remains valid");
        server.assert_command_succeeded(&created.command_id).await;
    }

    #[tokio::test]
    async fn definite_push_rejection_becomes_terminal_delivery_failure() {
        let server = TestCommandServer::new().await;
        let dispatcher = server
            .mock_dispatcher()
            .expect("push test server uses the mock dispatcher");
        dispatcher.set_should_reject(true).await;

        let created = server
            .create_command(test_inline_create_command(
                "push-agent",
                "definite-rejection",
            ))
            .await
            .expect("definite rejection must return the durable terminal command ID");
        assert_eq!(created.state, CommandState::Failed);

        let status = server
            .get_command_status(&created.command_id)
            .await
            .expect("command status");
        assert_eq!(status.state, CommandState::Failed);
        let Some(CommandResponse::Error { code, message, .. }) = status.response else {
            panic!("definite delivery rejection must persist an error response");
        };
        assert_eq!(code, "DELIVERY_FAILED");
        assert_eq!(message, "Worker runtime did not accept command delivery");

        let late = server
            .submit_command_response(
                &created.command_id,
                test_success_response(b"must not replace delivery failure"),
            )
            .await;
        assert!(
            late.is_ok(),
            "terminal duplicate submissions are idempotent"
        );
        server.assert_command_failed(&created.command_id).await;
    }

    /// Test lease operations
    #[tokio::test]
    async fn test_lease_operations() {
        // Lease operations require Pull mode (Push mode dispatches immediately)
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // Create command (stays Pending in Pull mode)
        let request = test_inline_create_command("lease-agent", "lease-command");
        let response = server.create_command(request).await.unwrap();
        assert_eq!(response.state, CommandState::Pending);

        // Acquire lease
        let lease = server
            .acquire_single_lease("lease-agent")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(lease.command_id, response.command_id);
        assert_eq!(lease.attempt, 1);
        assert!(lease.lease_expires_at > Utc::now());

        // Verify envelope details
        assert_envelope_command_id(&lease.envelope, &response.command_id);
        assert_envelope_command(&lease.envelope, "lease-command");

        // Test no available leases
        let empty_lease_request = LeaseRequest {
            deployment_id: "nonexistent-agent".to_string(),
            target: server.default_target.clone(),
            max_leases: 1,
            lease_seconds: 60,
        };
        let empty_response = server
            .acquire_lease("nonexistent-agent", empty_lease_request)
            .await
            .unwrap();
        assert_eq!(empty_response.leases.len(), 0);

        // Release lease
        server
            .release_lease(&lease.command_id, &lease.lease_id)
            .await
            .unwrap();
        server
            .assert_command_state(&response.command_id, CommandState::Pending)
            .await;
    }

    #[tokio::test]
    async fn max_timeout_lease_receives_response_credentials_beyond_lease_headroom() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;
        let created = server
            .create_command(test_inline_create_command("max-timeout", "run"))
            .await
            .unwrap();
        let lease = server
            .acquire_lease(
                "max-timeout",
                LeaseRequest {
                    deployment_id: "max-timeout".to_string(),
                    target: server.default_target.clone(),
                    max_leases: 1,
                    lease_seconds: 3660,
                },
            )
            .await
            .unwrap()
            .leases
            .into_iter()
            .next()
            .expect("max-timeout lease");
        assert_eq!(lease.command_id, created.command_id);

        let submit = url::Url::parse("http://manager.invalid")
            .unwrap()
            .join(&lease.envelope.response_handling.submit_response_url)
            .unwrap();
        let expires = submit
            .query_pairs()
            .find_map(|(key, value)| (key == "expires").then(|| value.parse::<i64>().unwrap()))
            .expect("response token expiry");
        let required = Utc::now() + chrono::Duration::seconds(3660);
        assert!(
            expires > required.timestamp(),
            "response token must outlive max execution plus lease headroom"
        );
        assert!(
            lease
                .envelope
                .response_handling
                .storage_upload_request
                .expiration
                > required,
            "response upload credential must outlive max execution plus lease headroom"
        );
    }

    #[tokio::test]
    async fn delayed_storage_command_lease_refreshes_expired_params_get() {
        let server = TestCommandServer::builder().with_pull_mode().build().await;
        let created = server
            .create_command(test_storage_create_command(
                "delayed-storage",
                "run",
                200_000,
            ))
            .await
            .unwrap();
        server
            .upload_complete(&created.command_id, test_upload_complete_request(200_000))
            .await
            .unwrap();

        let mut stored = server
            .command_server
            .get_params(&created.command_id)
            .await
            .unwrap()
            .expect("stored params");
        let BodySpec::Storage {
            storage_get_request: Some(request),
            ..
        } = &mut stored
        else {
            panic!("storage params with GET request");
        };
        request.expiration = Utc::now() - chrono::Duration::hours(1);
        server
            .command_server
            .store_params(&created.command_id, &stored)
            .await
            .unwrap();

        let lease = server
            .acquire_single_lease("delayed-storage")
            .await
            .unwrap()
            .expect("delayed command lease");
        let BodySpec::Storage {
            storage_get_request: Some(fresh),
            ..
        } = lease.envelope.params
        else {
            panic!("leased storage params with fresh GET request");
        };
        assert!(
            fresh.expiration > Utc::now() + chrono::Duration::minutes(50),
            "leasing must replace the stale upload-time params URL"
        );
    }

    /// Test response submission and idempotency
    #[tokio::test]
    async fn test_response_operations() {
        // Lease operations require Pull mode (Push mode dispatches immediately)
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // Create command and acquire lease
        let request = test_inline_create_command("response-agent", "response-command");
        let response = server.create_command(request).await.unwrap();
        let lease = server
            .acquire_single_lease("response-agent")
            .await
            .unwrap()
            .unwrap();

        // Submit response
        let agent_response = test_json_success_response(&serde_json::json!({
            "result": "success",
            "data": [1, 2, 3]
        }));
        server
            .submit_command_response(&lease.command_id, agent_response)
            .await
            .unwrap();
        server.assert_command_succeeded(&response.command_id).await;

        // Verify response data
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        let final_response = status.response.unwrap();
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            let decoded = body.decode_inline().unwrap();
            let json: serde_json::Value = serde_json::from_slice(&decoded).unwrap();
            assert_eq!(json["result"], "success");
        }

        // Test duplicate response submission (should be idempotent)
        let duplicate_response = test_success_response(b"second response");
        let result = server
            .submit_command_response(&lease.command_id, duplicate_response)
            .await;
        assert!(result.is_ok()); // Should not error, just ignore

        // Original response should still be there
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        let final_response = status.response.unwrap();
        if let CommandResponse::Success { response: body } = final_response {
            let decoded = body.decode_inline().unwrap();
            let json: serde_json::Value = serde_json::from_slice(&decoded).unwrap();
            assert_eq!(json["result"], "success"); // Not changed
        }
    }

    /// Regression: a failure partway through response cleanup must not strand the
    /// command as non-terminal with an invisible response, and must not fail the
    /// `submit_command_response` call either.
    ///
    /// `submit_command_response` stores the response blob, commits the terminal
    /// registry state (the source of truth), then cleans up the lease and pending
    /// index. Here the KV is armed to fail the pending-index scan performed by
    /// that cleanup. Cleanup is best-effort — a leftover pending-index/lease entry
    /// is reaped by `acquire_lease`'s terminal-state check on its next scan — so
    /// the submit call must still return `Ok`, only logging a warning, and the
    /// command must still be observable as `Succeeded` with its stored response.
    /// Under the old ordering (cleanup first, state last) the same fault left the
    /// command stuck as `Dispatched` with a stored-but-invisible response; under a
    /// prior version of the new ordering, the cleanup error was still propagated
    /// with `?`, failing the call despite the terminal state already being safe.
    #[tokio::test]
    #[tracing_test::traced_test]
    async fn test_response_cleanup_failure_keeps_command_terminal() {
        let server = TestCommandServer::builder()
            .with_pull_mode()
            .with_fault_injection()
            .build()
            .await;
        let fault_kv = server
            .fault_kv
            .clone()
            .expect("fault injection was requested");

        // Create + lease the command. Lease acquisition scans the same pending
        // prefix, so it must run before the fault is armed.
        let request = test_inline_create_command("cleanup-agent", "cleanup-command");
        let response = server.create_command(request).await.unwrap();
        let lease = server
            .acquire_single_lease("cleanup-agent")
            .await
            .unwrap()
            .unwrap();

        // Arm the fault so the pending-index cleanup scan fails during submit.
        fault_kv.arm_pending_scan_failure();

        let agent_response = test_json_success_response(&serde_json::json!({ "result": "ok" }));
        let submit_result = server
            .submit_command_response(&lease.command_id, agent_response)
            .await;
        assert!(
            submit_result.is_ok(),
            "cleanup failures are best-effort and must not fail submit_command_response \
             once the terminal state is committed"
        );
        assert!(
            logs_contain("Failed to clean up pending index"),
            "a cleanup failure must still be logged as a warning"
        );

        // Despite the cleanup failure, the terminal state was committed first, so
        // the command is visible as Succeeded with its stored response rather than
        // stranded as Dispatched with an invisible one.
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        assert_eq!(
            status.state,
            CommandState::Succeeded,
            "command must be terminal even though response cleanup failed"
        );
        let final_response = status
            .response
            .expect("stored response must be visible on the terminal command");
        assert!(final_response.is_success());
        if let CommandResponse::Success { response: body } = final_response {
            let decoded = body.decode_inline().unwrap();
            let json: serde_json::Value = serde_json::from_slice(&decoded).unwrap();
            assert_eq!(json["result"], "ok");
        }
    }

    /// Test runtime envelope parsing
    #[tokio::test]
    async fn test_runtime_integration() {
        // Test envelope parsing from queue message
        let envelope = test_simple_envelope("cmd_runtime_test", "runtime-command");
        let envelope_json = serde_json::to_value(&envelope).unwrap();
        let queue_message = QueueMessage {
            id: "msg_123".to_string(),
            payload: MessagePayload::Json(envelope_json),
            receipt_handle: "handle_123".to_string(),
            timestamp: Utc::now(),
            source: "test-queue".to_string(),
            attributes: std::collections::HashMap::new(),
            attempt_count: Some(1),
        };

        let parsed = parse_envelope(&queue_message).unwrap();
        assert!(parsed.is_some());
        let parsed_envelope = parsed.unwrap();
        assert_eq!(parsed_envelope.command_id, "cmd_runtime_test");
        assert_eq!(parsed_envelope.command, "runtime-command");

        // Test non-command message
        let non_arc_message = QueueMessage {
            id: "msg_456".to_string(),
            payload: MessagePayload::Json(serde_json::json!({"regular": "message"})),
            receipt_handle: "handle_456".to_string(),
            timestamp: Utc::now(),
            source: "test-queue".to_string(),
            attributes: std::collections::HashMap::new(),
            attempt_count: Some(1),
        };
        let parsed = parse_envelope(&non_arc_message).unwrap();
        assert!(parsed.is_none());

        // Test params decoding
        let params_json = serde_json::json!({"key": "value", "number": 42});
        let params_bytes = serde_json::to_vec(&params_json).unwrap();
        let test_envelope = test_envelope(
            "cmd_params",
            "params-command",
            BodySpec::inline(&params_bytes),
        );
        let decoded_params = decode_params(&test_envelope).await.unwrap();
        assert_eq!(decoded_params["key"], "value");
        assert_eq!(decoded_params["number"], 42);
    }

    /// Test error handling and edge cases
    #[tokio::test]
    async fn test_error_handling() {
        let server = TestCommandServer::new().await;

        // Test invalid command (empty command name)
        let invalid_request = CreateCommandRequest {
            deployment_id: "error-agent".to_string(),
            command: "".to_string(), // Invalid: empty command name
            params: BodySpec::inline(b"{}"),
            deadline: None,
            idempotency_key: None,
            target_resource_id: None,
        };
        let result = server.create_command(invalid_request).await;
        assert!(result.is_err());

        // Test invalid command (empty deployment_id)
        let invalid_request = CreateCommandRequest {
            deployment_id: "".to_string(), // Invalid: empty deployment_id
            command: "test".to_string(),
            params: BodySpec::inline(b"{}"),
            deadline: None,
            idempotency_key: None,
            target_resource_id: None,
        };
        let result = server.create_command(invalid_request).await;
        assert!(result.is_err());

        // Test operations on non-existent commands
        let upload_complete = test_upload_complete_request(1000);
        assert!(server
            .upload_complete("nonexistent", upload_complete)
            .await
            .is_err());
        assert!(server.get_command_status("nonexistent").await.is_err());
        assert!(server
            .submit_command_response("nonexistent", test_success_response(b"test"))
            .await
            .is_err());

        // Test command expiration (past deadline)
        let past_deadline = Utc::now() - chrono::Duration::minutes(1);
        let expired_request = test_create_command_with_deadline(
            "expired-agent",
            "expired-command",
            BodySpec::inline(b"{}"),
            past_deadline,
        );
        assert!(server.create_command(expired_request).await.is_err());
    }

    /// Test error response handling
    #[tokio::test]
    async fn test_error_response_handling() {
        // Lease operations require Pull mode (Push mode dispatches immediately)
        let server = TestCommandServer::builder().with_pull_mode().build().await;

        // Create command
        let request = test_inline_create_command("error-agent", "error-command");
        let response = server.create_command(request).await.unwrap();

        // Acquire lease
        let lease = server
            .acquire_single_lease("error-agent")
            .await
            .unwrap()
            .unwrap();

        // Submit error response
        let agent_response = test_error_response("PROCESSING_FAILED", "Something went wrong");
        server
            .submit_command_response(&lease.command_id, agent_response)
            .await
            .unwrap();

        // Verify command failed
        let status = server
            .get_command_status(&response.command_id)
            .await
            .unwrap();
        assert_eq!(status.state, CommandState::Failed);

        let final_response = status.response.unwrap();
        assert!(final_response.is_error());
        if let CommandResponse::Error { code, message, .. } = final_response {
            assert_eq!(code, "PROCESSING_FAILED");
            assert_eq!(message, "Something went wrong");
        }
    }
}

/// Two racing submitters with OPPOSITE outcomes (a redelivered execution
/// racing the original whose lease expired): exactly one wins the terminal
/// transition, the loser is swallowed as a duplicate, and the recorded state
/// matches the served response — a terminal record is never overwritten.
#[cfg(feature = "test-utils")]
#[tokio::test]
async fn concurrent_opposite_submits_keep_state_and_response_consistent() {
    use alien_commands::test_utils::*;
    use alien_commands::types::{CommandResponse, CommandState};

    let server = TestCommandServer::builder().with_pull_mode().build().await;

    let request = test_inline_create_command("pull-agent", "flaky-op");
    let created = server.create_command(request).await.unwrap();
    let lease = server
        .acquire_single_lease("pull-agent")
        .await
        .unwrap()
        .unwrap();

    let success = test_json_success_response(&serde_json::json!({ "ok": true }));
    let failure = CommandResponse::Error {
        code: "HANDLER_ERROR".to_string(),
        message: "boom".to_string(),
        details: None,
    };

    // Race the two submissions.
    let (a, b) = tokio::join!(
        server.submit_command_response(&lease.command_id, success),
        server.submit_command_response(&lease.command_id, failure),
    );
    // Both calls succeed: the loser is silently treated as a duplicate.
    a.unwrap();
    b.unwrap();

    let status = server
        .get_command_status(&created.command_id)
        .await
        .unwrap();
    assert!(status.state.is_terminal());
    let response = status.response.expect("terminal command serves a response");
    match (&status.state, &response) {
        (CommandState::Succeeded, CommandResponse::Success { .. }) => {}
        (CommandState::Failed, CommandResponse::Error { .. }) => {}
        (state, response) => {
            panic!("torn terminal record: state {state:?} does not match response {response:?}")
        }
    }
}

/// The deadline reaper terminates commands nothing else would ever touch:
/// a Pending pull command past its deadline (with its pending-index entry
/// cleaned) — the same path also covers PendingUpload commands whose params
/// upload never completed, since the reaper transitions ANY non-terminal
/// state.
#[cfg(feature = "test-utils")]
#[tokio::test]
async fn deadline_reaper_expires_overdue_commands() {
    use alien_commands::test_utils::*;
    use alien_commands::types::CommandState;

    let server = TestCommandServer::builder().with_pull_mode().build().await;

    // A multi-second deadline matters here: the index entry gets a KV TTL,
    // and the regression this guards is the TTL expiring AT the deadline —
    // hiding the entry from the reaper's scan exactly when it became due.
    let mut request = test_inline_create_command("pull-agent", "slow-op");
    request.deadline = Some(chrono::Utc::now() + chrono::Duration::seconds(2));
    let created = server.create_command(request).await.unwrap();
    assert_eq!(created.state, CommandState::Pending);

    tokio::time::sleep(std::time::Duration::from_millis(2600)).await;
    let expired = server.command_server.reap_expired_commands().await.unwrap();
    assert_eq!(expired, 1, "the overdue command must be reaped");

    let status = server
        .get_command_status(&created.command_id)
        .await
        .unwrap();
    assert_eq!(status.state, CommandState::Expired);

    // The pending index entry is gone: a poller can no longer lease it.
    let lease = server.acquire_single_lease("pull-agent").await.unwrap();
    assert!(lease.is_none(), "expired command must not be leasable");

    // A second reap pass is a no-op (index entry deleted).
    assert_eq!(
        server.command_server.reap_expired_commands().await.unwrap(),
        0
    );
}