use std::path::Path;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::Weak;
use std::time::Duration;

use anyhow::Context;
use codex_core::NotSubmittedReason;
use codex_core::StartIfIdleSubmission;
use codex_core::TurnInput;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionEventSink;
use codex_extension_api::ExtensionFuture;
use codex_extension_api::ExtensionRegistry;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_extension_api::ExtensionWarning;
use codex_extension_api::NoopExtensionEventSink;
use codex_extension_api::ThreadIdleCause;
use codex_extension_api::ThreadIdleInput;
use codex_extension_api::ThreadLifecycleContributor;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
use codex_protocol::user_input::UserInput;
use codex_queue_extension::QueueServiceError;
use codex_queue_extension::QueuedItemService;
use codex_state::SqliteConfig;
use codex_state::StateRuntime;
use codex_thread_store::LocalQueueStore;
use codex_thread_store::QueueStore;
use codex_thread_store::ThreadStoreError;
use codex_utils_absolute_path::test_support::PathExt;
use core_test_support::hooks::trust_discovered_hooks;
use core_test_support::responses;
use core_test_support::responses::start_mock_server;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event_match;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::sync::oneshot;

const TINY_PNG_BYTES: &[u8] = &[
    137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 13, 73, 72, 68, 82, 0, 0, 0, 1, 0, 0, 0, 1, 8, 6, 0,
    0, 0, 31, 21, 196, 137, 0, 0, 0, 11, 73, 68, 65, 84, 120, 156, 99, 96, 0, 2, 0, 0, 5, 0, 1,
    122, 94, 171, 63, 0, 0, 0, 0, 73, 69, 78, 68, 174, 66, 96, 130,
];
const TINY_PNG_DATA_URL: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVR4nGNgAAIAAAUAAXpeqz8AAAAASUVORK5CYII=";

#[derive(Default)]
struct RecordingEventSink {
    events: Mutex<Vec<Event>>,
}

impl ExtensionEventSink for RecordingEventSink {
    fn emit(&self, event: Event) {
        self.events
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(event);
    }

    fn emit_warning(&self, _warning: ExtensionWarning) {}
}

#[derive(Default)]
struct InstalledQueue(OnceLock<Arc<QueuedItemService>>);

impl ThreadLifecycleContributor<codex_core::config::Config> for InstalledQueue {
    fn on_thread_idle<'a>(&'a self, input: ThreadIdleInput<'a>) -> ExtensionFuture<'a, ()> {
        match self.0.get() {
            Some(service) => <QueuedItemService as ThreadLifecycleContributor<
                codex_core::config::Config,
            >>::on_thread_idle(service.as_ref(), input),
            None => Box::pin(async {}),
        }
    }
}

fn registered_queue_extensions() -> (
    Arc<InstalledQueue>,
    Arc<ExtensionRegistry<codex_core::config::Config>>,
) {
    let installed = Arc::new(InstalledQueue::default());
    let mut extensions = ExtensionRegistryBuilder::new();
    extensions.thread_lifecycle_contributor(installed.clone());
    (installed, Arc::new(extensions.build()))
}

fn install_registered_queue(
    test: &TestCodex,
    installed: &InstalledQueue,
) -> anyhow::Result<Arc<QueuedItemService>> {
    let service = Arc::new(QueuedItemService::new(
        loaded_thread_queue(test)?,
        Arc::downgrade(&test.thread_manager),
        Arc::new(NoopExtensionEventSink),
    ));
    assert!(installed.0.set(Arc::clone(&service)).is_ok());
    Ok(service)
}

fn write_rejecting_prompt_hook(home: &Path) {
    let script_path = home.join("queue_prompt_hook.py");
    let log_path = home.join("queue_prompt_hook.log");
    let script = format!(
        r#"import json
from pathlib import Path
import sys

payload = json.load(sys.stdin)
with Path(r"{log_path}").open("a", encoding="utf-8") as log:
    log.write(payload["prompt"] + "\n")
if payload["prompt"] == "blocked":
    print(json.dumps({{"decision": "block", "reason": "blocked by queue hook"}}))
"#,
        log_path = log_path.display(),
    );
    std::fs::write(&script_path, script)
        .unwrap_or_else(|error| panic!("write queue hook script: {error}"));
    let hooks = serde_json::json!({
        "hooks": {
            "UserPromptSubmit": [{
                "hooks": [{
                    "type": "command",
                    "command": format!("python3 {}", script_path.display()),
                }]
            }]
        }
    });
    std::fs::write(home.join("hooks.json"), hooks.to_string())
        .unwrap_or_else(|error| panic!("write queue hooks: {error}"));
}

async fn test_queue() -> anyhow::Result<(Arc<dyn QueueStore>, TempDir)> {
    let home = tempfile::tempdir()?;
    let sqlite = SqliteConfig::new_for_testing(home.path().abs());
    let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string()).await?;
    let queue: Arc<dyn QueueStore> = Arc::new(LocalQueueStore::new(runtime));
    Ok((queue, home))
}

fn loaded_thread_queue(test: &TestCodex) -> anyhow::Result<Arc<dyn QueueStore>> {
    let runtime = test.codex.state_db().context("state runtime unavailable")?;
    Ok(Arc::new(LocalQueueStore::new(runtime)))
}

fn user_input(text: &str) -> TurnInput {
    TurnInput::UserInput {
        content: vec![UserInput::Text {
            text: text.to_string(),
            text_elements: Vec::new(),
        }],
        client_id: None,
    }
}

fn structured_user_input(text: &str) -> TurnInput {
    TurnInput::UserInput {
        content: vec![UserInput::Text {
            text: text.to_string(),
            text_elements: Vec::new(),
        }],
        client_id: Some("stable-client-message".to_string()),
    }
}

fn user_input_with_media(text: &str, image: UserInput, audio: UserInput) -> TurnInput {
    let mut input = structured_user_input(text);
    if let TurnInput::UserInput { content, .. } = &mut input {
        content.extend([image, audio]);
    }
    input
}

async fn emit_idle(service: &QueuedItemService, thread_id: ThreadId) {
    emit_idle_with_cause(service, thread_id, ThreadIdleCause::Completed).await;
}

async fn emit_idle_with_cause(
    service: &QueuedItemService,
    thread_id: ThreadId,
    cause: ThreadIdleCause,
) {
    let session_store = ExtensionData::new("session");
    let thread_store = ExtensionData::new(thread_id.to_string());
    <QueuedItemService as ThreadLifecycleContributor<()>>::on_thread_idle(
        service,
        ThreadIdleInput {
            cause,
            session_store: &session_store,
            thread_store: &thread_store,
        },
    )
    .await;
}

#[tokio::test]
async fn queued_input_and_unique_event_ids_round_trip() -> anyhow::Result<()> {
    let (queue, _home) = test_queue().await?;
    let sink = Arc::new(RecordingEventSink::default());
    let event_sink: Arc<dyn ExtensionEventSink> = sink.clone();
    let service = QueuedItemService::new(queue, Weak::new(), event_sink);
    let thread_id = ThreadId::new();
    let input = structured_user_input("structured message");

    let first = service.enqueue(thread_id, input.clone()).await?;
    let second = service.enqueue(thread_id, user_input("next")).await?;
    let TurnInput::UserInput {
        client_id: Some(generated_client_id),
        ..
    } = &second.input
    else {
        anyhow::bail!("queued message did not receive a client message id");
    };
    assert_eq!(
        7,
        uuid::Uuid::parse_str(generated_client_id)?.get_version_num()
    );
    assert_eq!(vec![first.clone(), second], service.list(thread_id).await?);
    assert_eq!(input, first.input);

    let events = sink
        .events
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    assert_eq!(2, events.len());
    for event in events.iter() {
        let EventMsg::ThreadQueueChanged(change) = &event.msg else {
            anyhow::bail!("event is not a queue change");
        };
        assert_eq!(
            serde_json::json!({ "threadId": thread_id }),
            serde_json::to_value(change)?
        );
    }
    assert_ne!(events[0].id, events[1].id);
    Ok(())
}

#[tokio::test]
async fn editing_reordering_and_deleting_preserve_queue_identity() -> anyhow::Result<()> {
    let (queue, _home) = test_queue().await?;
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let thread_id = ThreadId::new();
    let first = service
        .enqueue(thread_id, structured_user_input("first"))
        .await?;
    let second = service.enqueue(thread_id, user_input("second")).await?;
    let edited = service
        .update(thread_id, first.id.clone(), user_input("edited"))
        .await?
        .context("queued item missing")?;
    assert_eq!(first.id, edited.id);
    let TurnInput::UserInput { client_id, .. } = &edited.input else {
        anyhow::bail!("edited queue item does not contain user input");
    };
    assert_eq!(Some("stable-client-message"), client_id.as_deref());
    assert!(matches!(
        service.reorder(thread_id, vec![first.id.clone()]).await,
        Err(QueueServiceError::Storage(
            ThreadStoreError::InvalidRequest { .. }
        ))
    ));
    service
        .reorder(thread_id, vec![second.id.clone(), first.id.clone()])
        .await?;
    assert_eq!(
        vec![second.clone(), edited.clone()],
        service.list(thread_id).await?
    );
    assert!(service.delete(thread_id, second.id).await?);
    assert_eq!(vec![edited], service.list(thread_id).await?);
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn starting_a_selected_item_preserves_the_remaining_queue() -> anyhow::Result<()> {
    let server = start_mock_server().await;
    let response =
        responses::mount_sse_once(&server, responses::sse_completed("selected-turn")).await;
    let test = test_codex().build_with_auto_env(&server).await?;
    let thread_id = test.session_configured.thread_id;
    let queue = loaded_thread_queue(&test)?;
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let first = service.enqueue(thread_id, user_input("first")).await?;
    let second = service
        .enqueue(thread_id, structured_user_input("second"))
        .await?;

    let submission = service
        .start(test.codex.as_ref(), second.id.clone(), /*trace*/ None)
        .await?;

    assert!(matches!(submission, StartIfIdleSubmission::Started { .. }));
    assert_eq!(vec![first], service.list(thread_id).await?);
    wait_for_event_match(test.codex.as_ref(), |event| match event {
        EventMsg::TurnComplete(_) => Some(()),
        _ => None,
    })
    .await;
    assert_eq!(
        Some("second"),
        response
            .single_request()
            .message_input_texts("user")
            .last()
            .map(String::as_str)
    );
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn starting_a_selected_item_while_active_leaves_it_queued() -> anyhow::Result<()> {
    let (release_response, response_gate) = oneshot::channel();
    let (server, _completions) = start_streaming_sse_server(vec![vec![
        StreamingSseChunk {
            gate: None,
            body: responses::sse(vec![responses::ev_response_created("resp-1")]),
        },
        StreamingSseChunk {
            gate: Some(response_gate),
            body: responses::sse(vec![responses::ev_completed("resp-1")]),
        },
    ]])
    .await;
    let test = test_codex().build_with_streaming_server(&server).await?;
    let thread_id = test.session_configured.thread_id;
    let service = QueuedItemService::new(
        loaded_thread_queue(&test)?,
        Weak::new(),
        Arc::new(NoopExtensionEventSink),
    );
    let queued = service
        .enqueue(thread_id, user_input("stay queued"))
        .await?;

    let active_turn = test
        .codex
        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
            text: "active turn".to_string(),
            text_elements: Vec::new(),
        }]))
        .await?;
    assert!(matches!(active_turn, TurnInputSubmission::Started { .. }));
    tokio::time::timeout(
        Duration::from_secs(5),
        server.wait_for_request_count(/*count*/ 1),
    )
    .await?;

    let error = service
        .start(test.codex.as_ref(), queued.id.clone(), /*trace*/ None)
        .await
        .expect_err("active turn should reject explicit queue start");
    assert!(matches!(
        error,
        QueueServiceError::NotStarted(NotSubmittedReason::NotIdle)
    ));
    assert_eq!(vec![queued], service.list(thread_id).await?);

    release_response
        .send(())
        .expect("active response gate should remain open");
    wait_for_event_match(test.codex.as_ref(), |event| {
        matches!(event, EventMsg::TurnComplete(_)).then_some(())
    })
    .await;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stopped_turns_leave_queued_messages_in_place() -> anyhow::Result<()> {
    let server = start_mock_server().await;
    let test = test_codex().build_with_auto_env(&server).await?;
    let thread_id = test.session_configured.thread_id;
    let queue = loaded_thread_queue(&test)?;
    let service = QueuedItemService::new(
        queue,
        Arc::downgrade(&test.thread_manager),
        Arc::new(NoopExtensionEventSink),
    );
    let queued = service
        .enqueue(thread_id, user_input("keep queued after stopping"))
        .await?;

    for cause in [ThreadIdleCause::Interrupted, ThreadIdleCause::Failed] {
        emit_idle_with_cause(&service, thread_id, cause).await;

        assert_eq!(vec![queued.clone()], service.list(thread_id).await?);
    }
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn registered_queue_lifecycle_starts_messages_in_fifo_order() -> anyhow::Result<()> {
    let server = start_mock_server().await;
    let responses = responses::mount_sse_sequence(
        &server,
        ["turn-a", "turn-b", "turn-c"]
            .into_iter()
            .map(responses::sse_completed)
            .collect(),
    )
    .await;
    let (installed, extensions) = registered_queue_extensions();
    let test = test_codex()
        .with_extensions(extensions)
        .with_config(|config| config.include_environment_context = false)
        .build_with_auto_env(&server)
        .await?;
    let thread_id = test.session_configured.thread_id;
    let staging = QueuedItemService::new(
        loaded_thread_queue(&test)?,
        Weak::new(),
        Arc::new(NoopExtensionEventSink),
    );
    for prompt in ["B", "C"] {
        staging.enqueue(thread_id, user_input(prompt)).await?;
    }
    let queue = install_registered_queue(&test, installed.as_ref())?;

    tokio::time::timeout(Duration::from_secs(10), async {
        test.submit_text_turn("A").await?;
        for _ in 0..2 {
            wait_for_event_match(test.codex.as_ref(), |event| {
                matches!(event, EventMsg::TurnComplete(_)).then_some(())
            })
            .await;
        }
        anyhow::Ok(())
    })
    .await??;

    let prompts = responses
        .requests()
        .into_iter()
        .map(|request| {
            request
                .message_input_texts("user")
                .pop()
                .context("model request has no user input")
        })
        .collect::<anyhow::Result<Vec<_>>>()?;
    assert_eq!(vec!["A", "B", "C"], prompts);
    assert!(queue.list(thread_id).await?.is_empty());
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rejected_queue_messages_are_consumed_without_retrying_or_blocking_followups()
-> anyhow::Result<()> {
    let server = start_mock_server().await;
    let responses = responses::mount_sse_sequence(
        &server,
        ["initial-turn", "following-turn"]
            .into_iter()
            .map(responses::sse_completed)
            .collect(),
    )
    .await;
    let (installed, extensions) = registered_queue_extensions();
    let test = test_codex()
        .with_extensions(extensions)
        .with_pre_build_hook(write_rejecting_prompt_hook)
        .with_config(trust_discovered_hooks)
        .with_config(|config| config.include_environment_context = false)
        .build_with_auto_env(&server)
        .await?;
    let thread_id = test.session_configured.thread_id;
    let staging = QueuedItemService::new(
        loaded_thread_queue(&test)?,
        Weak::new(),
        Arc::new(NoopExtensionEventSink),
    );
    for prompt in ["blocked", "C"] {
        staging.enqueue(thread_id, user_input(prompt)).await?;
    }
    let queue = install_registered_queue(&test, installed.as_ref())?;

    tokio::time::timeout(Duration::from_secs(10), async {
        test.submit_text_turn("A").await?;
        for _ in 0..2 {
            wait_for_event_match(test.codex.as_ref(), |event| {
                matches!(event, EventMsg::TurnComplete(_)).then_some(())
            })
            .await;
        }
        anyhow::Ok(())
    })
    .await??;

    let prompts = responses
        .requests()
        .into_iter()
        .map(|request| {
            request
                .message_input_texts("user")
                .pop()
                .context("model request has no user input")
        })
        .collect::<anyhow::Result<Vec<_>>>()?;
    assert_eq!(vec!["A", "C"], prompts);
    let hook_log = std::fs::read_to_string(test.codex_home_path().join("queue_prompt_hook.log"))?;
    assert_eq!(
        vec!["A", "blocked", "C"],
        hook_log.lines().collect::<Vec<_>>()
    );
    assert!(queue.list(thread_id).await?.is_empty());

    let rejected = staging.enqueue(thread_id, user_input("blocked")).await?;
    let submission = tokio::time::timeout(
        Duration::from_secs(10),
        queue.start(test.codex.as_ref(), rejected.id, /*trace*/ None),
    )
    .await?
    .expect("explicitly started input should be submitted");
    assert!(matches!(submission, StartIfIdleSubmission::Started { .. }));
    wait_for_event_match(test.codex.as_ref(), |event| {
        matches!(event, EventMsg::TurnComplete(_)).then_some(())
    })
    .await;
    assert!(queue.list(thread_id).await?.is_empty());
    let hook_log = std::fs::read_to_string(test.codex_home_path().join("queue_prompt_hook.log"))?;
    assert_eq!(
        vec!["A", "blocked", "C", "blocked"],
        hook_log.lines().collect::<Vec<_>>()
    );
    assert_eq!(2, responses.requests().len());
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn local_attachments_are_snapshotted_before_enqueue_and_update() -> anyhow::Result<()> {
    let (queue, home) = test_queue().await?;
    let service = QueuedItemService::new(
        Arc::clone(&queue),
        Weak::new(),
        Arc::new(NoopExtensionEventSink),
    );
    let thread_id = ThreadId::new();
    let image_path = home.path().join("queued.png");
    let audio_path = home.path().join("queued.mp3");
    std::fs::write(&image_path, TINY_PNG_BYTES)?;
    std::fs::write(&audio_path, b"audio")?;
    let queued_input = user_input_with_media(
        "queued attachments",
        UserInput::LocalImage {
            path: image_path.clone(),
            detail: Some(ImageDetail::Original),
        },
        UserInput::LocalAudio {
            path: audio_path.clone(),
        },
    );
    let expected_queued_input = user_input_with_media(
        "queued attachments",
        UserInput::Image {
            image_url: TINY_PNG_DATA_URL.to_string(),
            detail: Some(ImageDetail::Original),
        },
        UserInput::Audio {
            audio_url: "data:audio/mpeg;base64,YXVkaW8=".to_string(),
        },
    );

    let queued = service.enqueue(thread_id, queued_input).await?;
    std::fs::remove_file(&image_path)?;
    std::fs::remove_file(&audio_path)?;

    assert_eq!(expected_queued_input, queued.input);
    assert_eq!(vec![queued.clone()], service.list(thread_id).await?);
    let queued_record = queue
        .list_page(thread_id, /*offset*/ 0, /*limit*/ 1)
        .await?
        .into_iter()
        .next()
        .context("snapshotted queued attachments were not persisted")?;
    let persisted_queued_input: TurnInput = serde_json::from_str(&queued_record.payload)?;
    assert_eq!(expected_queued_input, persisted_queued_input);

    let edited_image_path = home.path().join("edited.png");
    let edited_audio_path = home.path().join("edited.m4a");
    std::fs::write(&edited_image_path, TINY_PNG_BYTES)?;
    std::fs::write(&edited_audio_path, b"edited audio")?;
    let edited_input = user_input_with_media(
        "edited attachments",
        UserInput::LocalImage {
            path: edited_image_path.clone(),
            detail: Some(ImageDetail::High),
        },
        UserInput::LocalAudio {
            path: edited_audio_path.clone(),
        },
    );
    let expected_edited_input = user_input_with_media(
        "edited attachments",
        UserInput::Image {
            image_url: TINY_PNG_DATA_URL.to_string(),
            detail: Some(ImageDetail::High),
        },
        UserInput::Audio {
            audio_url: "data:audio/mp4;base64,ZWRpdGVkIGF1ZGlv".to_string(),
        },
    );

    let edited = service
        .update(thread_id, queued.id.clone(), edited_input)
        .await?
        .context("snapshotted queued attachments were not updated")?;
    std::fs::remove_file(&edited_image_path)?;
    std::fs::remove_file(&edited_audio_path)?;

    assert_eq!(expected_edited_input, edited.input);
    assert_eq!(vec![edited.clone()], service.list(thread_id).await?);
    let edited_record = queue
        .list_page(thread_id, /*offset*/ 0, /*limit*/ 1)
        .await?
        .into_iter()
        .next()
        .context("snapshotted edited attachments were not persisted")?;
    let persisted_edited_input: TurnInput = serde_json::from_str(&edited_record.payload)?;
    assert_eq!(expected_edited_input, persisted_edited_input);

    Ok(())
}

#[tokio::test]
async fn invalid_local_attachments_do_not_mutate_queue() -> anyhow::Result<()> {
    let (queue, home) = test_queue().await?;
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let thread_id = ThreadId::new();
    let existing = service.enqueue(thread_id, user_input("existing")).await?;
    let missing_path = home.path().join("missing.png");
    let invalid_input = user_input_with_media(
        "invalid attachments",
        UserInput::LocalImage {
            path: missing_path,
            detail: Some(ImageDetail::Original),
        },
        UserInput::Audio {
            audio_url: "data:audio/mpeg;base64,YXVkaW8=".to_string(),
        },
    );

    assert!(matches!(
        service
            .enqueue(thread_id, invalid_input.clone())
            .await,
        Err(QueueServiceError::InvalidAttachment(error))
            if error.kind() == std::io::ErrorKind::NotFound
    ));
    assert!(matches!(
        service
            .update(
                thread_id,
                existing.id.clone(),
                invalid_input,
            )
            .await,
        Err(QueueServiceError::InvalidAttachment(error))
            if error.kind() == std::io::ErrorKind::NotFound
    ));
    assert_eq!(vec![existing], service.list(thread_id).await?);

    Ok(())
}

#[tokio::test]
async fn non_user_input_cannot_enter_the_user_message_queue() -> anyhow::Result<()> {
    let (queue, _home) = test_queue().await?;
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let error = service
        .enqueue(
            ThreadId::new(),
            TurnInput::ResponseItem(ResponseItem::Other),
        )
        .await
        .expect_err("response item should not enter the user queue");
    assert!(matches!(error, QueueServiceError::InvalidInput));
    Ok(())
}

#[tokio::test]
async fn queued_text_limit_is_enforced_across_all_input_items() -> anyhow::Result<()> {
    let (queue, _home) = test_queue().await?;
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let thread_id = ThreadId::new();
    let existing = service.enqueue(thread_id, user_input("existing")).await?;
    let mut oversized = user_input(&"x".repeat(MAX_USER_INPUT_TEXT_CHARS / 2));
    if let TurnInput::UserInput { content, .. } = &mut oversized {
        content.push(UserInput::Text {
            text: "y".repeat(MAX_USER_INPUT_TEXT_CHARS / 2 + 1),
            text_elements: Vec::new(),
        });
    }

    assert!(matches!(
        service
            .enqueue(thread_id, oversized.clone())
            .await,
        Err(QueueServiceError::InputTooLarge { actual_chars })
            if actual_chars == MAX_USER_INPUT_TEXT_CHARS + 1
    ));
    assert!(matches!(
        service
            .update(
                thread_id,
                existing.id.clone(),
                oversized,
            )
            .await,
        Err(QueueServiceError::InputTooLarge { actual_chars })
            if actual_chars == MAX_USER_INPUT_TEXT_CHARS + 1
    ));
    assert_eq!(vec![existing], service.list(thread_id).await?);

    Ok(())
}

#[tokio::test]
async fn queued_text_limit_counts_characters_and_ignores_non_text_items() -> anyhow::Result<()> {
    let (queue, _home) = test_queue().await?;
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let mut input = structured_user_input(&"é".repeat(MAX_USER_INPUT_TEXT_CHARS));
    if let TurnInput::UserInput { content, .. } = &mut input {
        content.push(UserInput::Mention {
            name: "Demo App".to_string(),
            path: "app://demo-app".to_string(),
        });
    }

    let queued = service.enqueue(ThreadId::new(), input.clone()).await?;
    assert_eq!(input, queued.input);

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn invalid_head_is_skipped_and_a_live_user_turn_is_accepted() -> anyhow::Result<()> {
    let server = start_mock_server().await;
    let response =
        responses::mount_sse_once(&server, responses::sse_completed("queued-turn")).await;
    let test = test_codex()
        .with_config(|config| config.include_environment_context = false)
        .build_with_auto_env(&server)
        .await?;
    let thread_id = test.session_configured.thread_id;
    let queue = loaded_thread_queue(&test)?;
    queue
        .enqueue(thread_id, r#"{"unsupported":true}"#.to_string())
        .await?;
    let service = QueuedItemService::new(
        queue,
        Arc::downgrade(&test.thread_manager),
        Arc::new(NoopExtensionEventSink),
    );

    service
        .enqueue(thread_id, structured_user_input("durable follow-up"))
        .await?;
    emit_idle(&service, thread_id).await;
    let client_id = wait_for_event_match(test.codex.as_ref(), |event| match event {
        EventMsg::ItemCompleted(event) => match &event.item {
            TurnItem::UserMessage(item) => Some(item.client_id.clone()),
            _ => None,
        },
        _ => None,
    })
    .await;
    assert_eq!(Some("stable-client-message".to_string()), client_id);
    wait_for_event_match(test.codex.as_ref(), |event| match event {
        EventMsg::TurnComplete(_) => Some(()),
        _ => None,
    })
    .await;
    assert!(service.list(thread_id).await?.is_empty());
    let request = response.single_request();
    assert_eq!(
        vec!["durable follow-up"],
        request.message_input_texts("user")
    );
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resumed_idle_dispatches_input_without_a_loaded_manager() -> anyhow::Result<()> {
    let server = start_mock_server().await;
    responses::mount_sse_once(&server, responses::sse_completed("resumed-turn")).await;
    let test = test_codex().build_with_auto_env(&server).await?;
    let thread_id = test.session_configured.thread_id;
    let queue = loaded_thread_queue(&test)?;
    QueuedItemService::new(
        Arc::clone(&queue),
        Weak::new(),
        Arc::new(NoopExtensionEventSink),
    )
    .enqueue(thread_id, user_input("queued while unloaded"))
    .await?;
    let service = QueuedItemService::new(
        queue,
        Arc::downgrade(&test.thread_manager),
        Arc::new(NoopExtensionEventSink),
    );
    emit_idle(&service, thread_id).await;
    assert!(service.list(thread_id).await?.is_empty());
    Ok(())
}