{"record":{"id":"402d2ce90f7914b9","repo":"openai/codex","slug":"queued-submission-payload-is-invalid-0","errorCode":null,"errorMessage":"queued submission payload is invalid: {0}","messagePattern":"queued submission payload is invalid: (.+?)","errorType":"validation","errorClass":"QueueServiceError","httpStatus":null,"severity":"error","filePath":"codex-rs/ext/queue/src/service.rs","lineNumber":50,"sourceCode":"use codex_thread_store::ThreadStoreError;\nuse thiserror::Error;\nuse tokio::sync::Mutex;\nuse tokio::sync::OwnedMutexGuard;\nuse tokio::sync::broadcast::error::TryRecvError;\nuse uuid::Uuid;\n\n/// One user message waiting to start on its thread.\n#[derive(Clone, Debug, PartialEq)]\npub struct QueuedItem {\n    pub id: String,\n    pub input: TurnInput,\n}\n\n#[derive(Debug, Error)]\npub enum QueueServiceError {\n    #[error(\"queue storage failed: {0}\")]\n    Storage(#[from] ThreadStoreError),\n    #[error(\"queued submission payload is invalid: {0}\")]\n    InvalidPayload(#[from] serde_json::Error),\n    #[error(\"local queued attachment is invalid: {0}\")]\n    InvalidAttachment(#[from] std::io::Error),\n    #[error(\"Core failed to submit queued user message: {0}\")]\n    CoreSubmissionError(#[from] CodexErr),\n    #[error(\"only user input can be added to the user-message queue\")]\n    InvalidInput,\n    #[error(\n        \"queued user input exceeds the maximum length of {MAX_USER_INPUT_TEXT_CHARS} characters ({actual_chars} provided)\"\n    )]\n    InputTooLarge { actual_chars: usize },\n}\n\n#[derive(Clone)]\npub struct QueuedItemService {\n    queue: Arc<dyn QueueStore>,\n    thread_manager: Weak<ThreadManager>,\n    event_sink: Arc<dyn ExtensionEventSink>,","sourceCodeStart":32,"sourceCodeEnd":68,"githubUrl":"https://github.com/openai/codex/blob/339751715c64496cb86246bfb3935f40e309dd3d/codex-rs/ext/queue/src/service.rs#L32-L68","documentation":"Wraps serde_json::Error via #[from] (codex-rs/ext/queue/src/service.rs:50-51). It has two directions: serializing a TurnInput when enqueueing or updating (service.rs:270 and 321, practically unreachable for well-formed values) and deserializing a stored payload back into TurnInput in queued_item_from_record (service.rs:560-566). In practice it fires on the read side: a queue record whose payload no longer matches the TurnInput schema, typically written by an older or newer binary. The dispatcher itself treats such records as garbage, discarding them with a warning (service.rs:419-426), but list(), update(), and start() propagate the error.","triggerScenarios":"Calling list(), list_page(), update(), or start() while the store contains a payload that fails to deserialize as TurnInput; version skew between the writer and reader of the queue table; hand-edited or migrated queue rows.","commonSituations":"Upgrading codex while old queued messages exist and the TurnInput shape changed; two different versions running against the same state database; test fixtures with hand-written payloads.","solutions":["Run one version at a time against a given state database so writer and reader agree on the schema.","Drain or delete stale queued items; the automatic dispatcher already discards invalid ones when the thread goes idle, and queue.delete(thread_id, item_id) removes specific rows.","Read the serde error message; it names the field that failed to deserialize.","For queues that must survive upgrades, add a migration or clear the thread queue table."],"exampleFix":"// before\nlet items = service.list(thread_id).await?; // fails if any stored payload is stale\n\n// after - isolate and drop unreadable rows (same policy as the dispatcher)\nfor record in store.list_page(thread_id, 0, MAX_QUEUE_ITEMS).await? {\n    if serde_json::from_str::<TurnInput>(&record.payload).is_err() {\n        tracing::warn!(id = %record.id, \"dropping unreadable queued item\");\n        store.delete(thread_id, record.id).await?;\n    }\n}\nlet items = service.list(thread_id).await?;","handlingStrategy":"fallback","validationCode":"// Rust - probe a stored payload before trusting it\nfn payload_parses(payload: &str) -> bool {\n    serde_json::from_str::<TurnInput>(payload).is_ok()\n}","typeGuard":null,"tryCatchPattern":"match service.list(thread_id).await {\n    Ok(items) => { /* ... */ }\n    Err(QueueServiceError::InvalidPayload(serde_err)) => {\n        // fall back to store-level listing and drop rows that do not parse\n        // (the dispatcher uses the same warn-and-delete policy, service.rs:419-426)\n        tracing::warn!(%serde_err, \"dropping unreadable queue rows\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Run a single codex version against each state database.","After upgrades with items still queued, expect the dispatcher to discard old-shape rows.","Never hand-edit queue payloads.","Wrap list() so one bad row degrades gracefully instead of failing the whole queue view."],"tags":["rust","codex","queue","serde","json","version-skew","schema-drift"],"backgroundTag":"json-deserialization-failed","analyzedSha":"339751715c64496cb86246bfb3935f40e309dd3d","analyzedAt":"2026-08-25T05:35:09.876Z","schemaVersion":2},"datasetVersion":"2026-08-25T06:17:31.827Z"}