fix: restore console input and attachment workflows

Restore durable busy Submit queues and TUI Enter delivery, expose Web Queue beside Notify, and separate the Composer toolbar from its editor without persistent success notices.

Support pasted image attachments and restore canonical Worker Session uploaded-file operations, including retention, binding, cleanup and authenticated upload context.

Validated Worker, Protocol and Session Store suites, 278 TUI tests, 59 Web tests and Svelte check. Runtime suite could not complete due to external Workdir transport timeouts.

crates/protocol/src/lib.rs

@@ -219,9 +219,9 @@ impl From<WorkerStatus> for WorkerStateSnapshot {
 #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "method", content = "params", rename_all = "snake_case")]
 pub enum Method {
-    /// Durably accept typed input for immediate activation while the Worker is Idle.
-    /// Running or Paused Workers reject Submit; use Notify for advisory information
-    /// that should be incorporated into work already in progress.
+    /// Durably accept typed input for immediate activation or the session FIFO.
+    /// Running or Paused Workers queue Submit for a later turn; use Notify for
+    /// advisory information to incorporate into work already in progress.
     ///
     /// `submission_request_id` is generated by the authenticated caller and is
     /// used only for idempotent retry. Worker allocates the durable
@@ -786,8 +786,7 @@ pub enum ToolResultDisposition {
 #[serde(rename_all = "snake_case")]
 pub enum SubmissionDisposition {
     Started,
-    /// Compatibility value for durable receipts created before Submit became
-    /// Idle-only. New submissions are never accepted with this disposition.
+    /// Input is durably held in the session FIFO until eligible for activation.
     Queued,
 }
 
@@ -933,10 +932,9 @@ pub struct SessionToolAttachment {
 #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "event", content = "data", rename_all = "snake_case")]
 pub enum Event {
-    /// Durable Submit acceptance. New submissions are accepted only while Idle,
-    /// and a `Started` receipt follows the pending activation checkpoint; the
-    /// corresponding [`Event::UserMessage`] marks the later history commit.
-    /// `Queued` remains readable for receipts persisted by older runtimes.
+    /// Durable Submit acceptance. A `Queued` receipt follows the durable FIFO
+    /// checkpoint. A `Started` receipt follows the pending activation checkpoint;
+    /// the corresponding [`Event::UserMessage`] marks the later history commit.
     /// Repeating the same request id and exact payload returns the same receipt
     /// without appending or activating twice.
     SubmissionAccepted {

crates/session-store/src/worker_metadata.rs

@@ -810,6 +810,112 @@ where
         self.session_store
             .read_paste_artifact(session_id, artifact_id)
     }
+    fn write_uploaded_file(
+        &self,
+        session_id: SessionId,
+        file_name: &str,
+        media_type: &str,
+        content: &[u8],
+        limits: crate::UploadedFileLimits,
+    ) -> Result<protocol::UploadedFileRef, crate::StoreError> {
+        self.session_store
+            .write_uploaded_file(session_id, file_name, media_type, content, limits)
+    }
+    fn write_uploaded_file_with_context(
+        &self,
+        session_id: SessionId,
+        file_name: &str,
+        media_type: &str,
+        content: &[u8],
+        context: &crate::UploadedFileUploadContext,
+        limits: crate::UploadedFileLimits,
+    ) -> Result<protocol::UploadedFileRef, crate::StoreError> {
+        self.session_store.write_uploaded_file_with_context(
+            session_id, file_name, media_type, content, context, limits,
+        )
+    }
+    fn read_uploaded_file(
+        &self,
+        session_id: SessionId,
+        reference: &protocol::UploadedFileRef,
+    ) -> Result<Vec<u8>, crate::StoreError> {
+        self.session_store.read_uploaded_file(session_id, reference)
+    }
+    fn read_uploaded_file_by_id(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+    ) -> Result<(protocol::UploadedFileRef, Vec<u8>), crate::StoreError> {
+        self.session_store
+            .read_uploaded_file_by_id(session_id, artifact_id)
+    }
+    fn bind_uploaded_file(
+        &self,
+        session_id: SessionId,
+        reference: &protocol::UploadedFileRef,
+        source_entry_id: &str,
+    ) -> Result<protocol::UploadedFileRef, crate::StoreError> {
+        self.session_store
+            .bind_uploaded_file(session_id, reference, source_entry_id)
+    }
+    fn pin_uploaded_file(
+        &self,
+        session_id: SessionId,
+        reference: &protocol::UploadedFileRef,
+        owner_id: &str,
+    ) -> Result<(), crate::StoreError> {
+        self.session_store
+            .pin_uploaded_file(session_id, reference, owner_id)
+    }
+    fn release_uploaded_file_pin(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+        owner_id: &str,
+    ) -> Result<(), crate::StoreError> {
+        self.session_store
+            .release_uploaded_file_pin(session_id, artifact_id, owner_id)
+    }
+    fn finalize_uploaded_file_binding(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+        source_entry_id: &str,
+    ) -> Result<(), crate::StoreError> {
+        self.session_store
+            .finalize_uploaded_file_binding(session_id, artifact_id, source_entry_id)
+    }
+    fn reconcile_uploaded_file_pins(
+        &self,
+        session_id: SessionId,
+        live_owner_ids: &[String],
+    ) -> Result<u64, crate::StoreError> {
+        self.session_store
+            .reconcile_uploaded_file_pins(session_id, live_owner_ids)
+    }
+    fn delete_uploaded_file(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+    ) -> Result<bool, crate::StoreError> {
+        self.session_store
+            .delete_uploaded_file(session_id, artifact_id)
+    }
+    fn delete_uncommitted_uploaded_files(
+        &self,
+        session_id: SessionId,
+    ) -> Result<u64, crate::StoreError> {
+        self.session_store
+            .delete_uncommitted_uploaded_files(session_id)
+    }
+    fn copy_committed_uploaded_files(
+        &self,
+        source_session_id: SessionId,
+        target_session_id: SessionId,
+    ) -> Result<u64, crate::StoreError> {
+        self.session_store
+            .copy_committed_uploaded_files(source_session_id, target_session_id)
+    }
     fn append_trace(
         &self,
         session_id: SessionId,

crates/session-store/src/worker_session_store.rs

@@ -13,8 +13,16 @@ use crate::event_trace::TraceEntry;
 use crate::paste_artifact::{read_from_dir, write_to_dir};
 use crate::segment_log::LogEntry;
 use crate::store::{Store, StoreError};
-use crate::{PasteArtifactLimits, SegmentId, SessionId};
-use protocol::PasteArtifactRef;
+use crate::uploaded_file::{
+    bind_uploaded_file, clear_uploaded_file_binding, delete_uncommitted_uploaded_files,
+    delete_uploaded_file, finalize_uploaded_file_binding, list_uploaded_file_refs,
+    pin_uploaded_file, read_uploaded_file, read_uploaded_file_by_id, reconcile_uploaded_file_pins,
+    release_uploaded_file_pin, uploaded_file_has_pending_owner, write_uploaded_file,
+};
+use crate::{
+    PasteArtifactLimits, SegmentId, SessionId, UploadedFileLimits, UploadedFileUploadContext,
+};
+use protocol::{PasteArtifactRef, UploadedFileRef};
 use serde::{Deserialize, Serialize};
 use std::fs::{self, File, OpenOptions};
 use std::io::{Read, Seek, SeekFrom, Write};
@@ -207,6 +215,40 @@ impl WorkerSessionStore {
             .join(format!("{segment_id}.trace.jsonl"))
     }
 
+    fn uploaded_file_is_referenced(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+    ) -> Result<bool, StoreError> {
+        fn segments_contain(segments: &[protocol::Segment], artifact_id: &str) -> bool {
+            segments.iter().any(|segment| {
+                matches!(
+                    segment,
+                    protocol::Segment::UploadedFile { file }
+                        if file.artifact_id == artifact_id
+                )
+            })
+        }
+
+        for segment_id in self.list_segments(session_id)? {
+            for entry in self.read_all(session_id, segment_id)? {
+                let referenced = match entry {
+                    LogEntry::AnnotatedUserInput { segments, .. } => {
+                        segments_contain(&segments, artifact_id)
+                    }
+                    LogEntry::InputSegmentsCheckpoint { user_segments, .. } => user_segments
+                        .iter()
+                        .any(|segments| segments_contain(segments, artifact_id)),
+                    _ => false,
+                };
+                if referenced {
+                    return Ok(true);
+                }
+            }
+        }
+        Ok(false)
+    }
+
     fn append_log_entry(&self, path: &Path, entry: &LogEntry) -> Result<(), StoreError> {
         let _guard = self
             .append_lock
@@ -402,6 +444,200 @@ impl Store for WorkerSessionStore {
         read_from_dir(&self.root.join(PASTE_ARTIFACTS_DIR), artifact_id)
     }
 
+    fn write_uploaded_file(
+        &self,
+        session_id: SessionId,
+        file_name: &str,
+        media_type: &str,
+        content: &[u8],
+        limits: UploadedFileLimits,
+    ) -> Result<UploadedFileRef, StoreError> {
+        self.ensure_session(session_id, true)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        write_uploaded_file(
+            &self.root.join(PASTE_ARTIFACTS_DIR),
+            file_name,
+            media_type,
+            content,
+            None,
+            limits,
+        )
+    }
+
+    fn write_uploaded_file_with_context(
+        &self,
+        session_id: SessionId,
+        file_name: &str,
+        media_type: &str,
+        content: &[u8],
+        context: &UploadedFileUploadContext,
+        limits: UploadedFileLimits,
+    ) -> Result<UploadedFileRef, StoreError> {
+        self.ensure_session(session_id, true)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        write_uploaded_file(
+            &self.root.join(PASTE_ARTIFACTS_DIR),
+            file_name,
+            media_type,
+            content,
+            Some(context),
+            limits,
+        )
+    }
+
+    fn read_uploaded_file(
+        &self,
+        session_id: SessionId,
+        reference: &UploadedFileRef,
+    ) -> Result<Vec<u8>, StoreError> {
+        self.ensure_session(session_id, false)?;
+        read_uploaded_file(&self.root.join(PASTE_ARTIFACTS_DIR), reference)
+    }
+
+    fn read_uploaded_file_by_id(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+    ) -> Result<(UploadedFileRef, Vec<u8>), StoreError> {
+        self.ensure_session(session_id, false)?;
+        read_uploaded_file_by_id(&self.root.join(PASTE_ARTIFACTS_DIR), artifact_id)
+    }
+
+    fn bind_uploaded_file(
+        &self,
+        session_id: SessionId,
+        reference: &UploadedFileRef,
+        source_entry_id: &str,
+    ) -> Result<UploadedFileRef, StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        let dir = self.root.join(PASTE_ARTIFACTS_DIR);
+        match bind_uploaded_file(&dir, reference, source_entry_id) {
+            Err(StoreError::ArtifactAlreadyCommitted) => {
+                let (stored, _) = read_uploaded_file_by_id(&dir, &reference.artifact_id)?;
+                let previous_source = stored
+                    .source_entry_id
+                    .ok_or(StoreError::ArtifactIntegrityMismatch)?;
+                if self.uploaded_file_is_referenced(session_id, &reference.artifact_id)? {
+                    return Err(StoreError::ArtifactAlreadyCommitted);
+                }
+                clear_uploaded_file_binding(&dir, &reference.artifact_id, &previous_source)?;
+                bind_uploaded_file(&dir, reference, source_entry_id)
+            }
+            result => result,
+        }
+    }
+
+    fn pin_uploaded_file(
+        &self,
+        session_id: SessionId,
+        reference: &UploadedFileRef,
+        owner_id: &str,
+    ) -> Result<(), StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        pin_uploaded_file(&self.root.join(PASTE_ARTIFACTS_DIR), reference, owner_id)
+    }
+
+    fn release_uploaded_file_pin(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+        owner_id: &str,
+    ) -> Result<(), StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        release_uploaded_file_pin(&self.root.join(PASTE_ARTIFACTS_DIR), artifact_id, owner_id)
+    }
+
+    fn finalize_uploaded_file_binding(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+        source_entry_id: &str,
+    ) -> Result<(), StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        finalize_uploaded_file_binding(
+            &self.root.join(PASTE_ARTIFACTS_DIR),
+            artifact_id,
+            source_entry_id,
+        )
+    }
+
+    fn reconcile_uploaded_file_pins(
+        &self,
+        session_id: SessionId,
+        live_owner_ids: &[String],
+    ) -> Result<u64, StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        reconcile_uploaded_file_pins(&self.root.join(PASTE_ARTIFACTS_DIR), live_owner_ids)
+    }
+
+    fn delete_uploaded_file(
+        &self,
+        session_id: SessionId,
+        artifact_id: &str,
+    ) -> Result<bool, StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        delete_uploaded_file(&self.root.join(PASTE_ARTIFACTS_DIR), artifact_id)
+    }
+
+    fn delete_uncommitted_uploaded_files(&self, session_id: SessionId) -> Result<u64, StoreError> {
+        self.ensure_session(session_id, false)?;
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("Worker Session append lock was poisoned"))?;
+        let dir = self.root.join(PASTE_ARTIFACTS_DIR);
+        let mut removed = delete_uncommitted_uploaded_files(&dir)?;
+        for reference in list_uploaded_file_refs(&dir)? {
+            let Some(source_entry_id) = reference.source_entry_id.as_deref() else {
+                continue;
+            };
+            if self.uploaded_file_is_referenced(session_id, &reference.artifact_id)? {
+                finalize_uploaded_file_binding(&dir, &reference.artifact_id, source_entry_id)?;
+                continue;
+            }
+            if uploaded_file_has_pending_owner(&dir, &reference.artifact_id)? {
+                continue;
+            }
+            clear_uploaded_file_binding(&dir, &reference.artifact_id, source_entry_id)?;
+            if delete_uploaded_file(&dir, &reference.artifact_id)? {
+                removed = removed
+                    .checked_add(1)
+                    .ok_or(StoreError::ArtifactQuotaExceeded)?;
+            }
+        }
+        Ok(removed)
+    }
+
     fn append_trace(
         &self,
         session_id: SessionId,
@@ -662,6 +898,212 @@ mod tests {
         }
     }
 
+    #[test]
+    fn canonical_combined_store_supports_attachment_uploads() {
+        let root = tempfile::tempdir().unwrap();
+        let session = WorkerSessionStore::new(root.path().join("session")).unwrap();
+        let store = crate::CombinedStore::new(session.clone(), ());
+        let session_id = new_session_id();
+        store
+            .create_segment(session_id, new_segment_id(), &[])
+            .unwrap();
+        let context = crate::UploadedFileUploadContext {
+            upload_id: "upload-1".into(),
+            principal_id: "account-1".into(),
+            workspace_id: "workspace-1".into(),
+            runtime_id: "runtime-1".into(),
+            worker_id: "worker-1".into(),
+        };
+        let file = store
+            .write_uploaded_file_with_context(
+                session_id,
+                "image.png",
+                "image/png",
+                b"\x89PNG\r\n\x1a\n",
+                &context,
+                crate::UploadedFileLimits::default(),
+            )
+            .expect("canonical Runtime store must support attachment upload");
+        assert_eq!(
+            store.read_uploaded_file(session_id, &file).unwrap(),
+            b"\x89PNG\r\n\x1a\n"
+        );
+        assert_eq!(
+            session.read_uploaded_file(session_id, &file).unwrap(),
+            b"\x89PNG\r\n\x1a\n"
+        );
+        assert_eq!(
+            store
+                .write_uploaded_file_with_context(
+                    session_id,
+                    "image.png",
+                    "image/png",
+                    b"\x89PNG\r\n\x1a\n",
+                    &context,
+                    crate::UploadedFileLimits::default(),
+                )
+                .unwrap(),
+            file
+        );
+        assert!(
+            store
+                .write_uploaded_file_with_context(
+                    session_id,
+                    "image.png",
+                    "image/png",
+                    b"\x89PNG\r\n\x1a\n",
+                    &crate::UploadedFileUploadContext {
+                        principal_id: "different-account".into(),
+                        ..context
+                    },
+                    crate::UploadedFileLimits::default(),
+                )
+                .is_err(),
+            "CombinedStore must not discard authenticated upload context"
+        );
+        let reopened = WorkerSessionStore::new(root.path().join("session")).unwrap();
+        assert_eq!(
+            reopened
+                .read_uploaded_file_by_id(session_id, &file.artifact_id)
+                .unwrap()
+                .0,
+            file
+        );
+        assert!(matches!(
+            store.read_uploaded_file(new_session_id(), &file),
+            Err(StoreError::Corrupt { .. })
+        ));
+        assert!(
+            store
+                .delete_uploaded_file(session_id, &file.artifact_id)
+                .unwrap()
+        );
+    }
+
+    #[test]
+    fn canonical_attachment_pins_binding_and_cleanup_survive_reopen() {
+        let root = tempfile::tempdir().unwrap();
+        let store = crate::CombinedStore::new(WorkerSessionStore::new(root.path()).unwrap(), ());
+        let id = new_session_id();
+        let limits = UploadedFileLimits::default();
+        store.create_segment(id, new_segment_id(), &[]).unwrap();
+        let file = store
+            .write_uploaded_file(id, "pending.txt", "text/plain", b"pending", limits)
+            .unwrap();
+        store.pin_uploaded_file(id, &file, "submit-1").unwrap();
+        assert!(store.delete_uploaded_file(id, &file.artifact_id).is_err());
+        drop(store);
+        let store = crate::CombinedStore::new(WorkerSessionStore::new(root.path()).unwrap(), ());
+        assert_eq!(
+            store
+                .reconcile_uploaded_file_pins(id, &["submit-1".into()])
+                .unwrap(),
+            0
+        );
+        assert_eq!(store.delete_uncommitted_uploaded_files(id).unwrap(), 0);
+        let bound = store.bind_uploaded_file(id, &file, "entry-1").unwrap();
+        store
+            .create_segment(
+                id,
+                new_segment_id(),
+                &[LogEntry::InputSegmentsCheckpoint {
+                    ts: 1,
+                    user_segments: vec![vec![protocol::Segment::UploadedFile {
+                        file: bound.clone(),
+                    }]],
+                }],
+            )
+            .unwrap();
+        store
+            .finalize_uploaded_file_binding(id, &file.artifact_id, "entry-1")
+            .unwrap();
+        assert_eq!(store.delete_uncommitted_uploaded_files(id).unwrap(), 0);
+        assert_eq!(store.read_uploaded_file(id, &bound).unwrap(), b"pending");
+        assert!(
+            store
+                .bind_uploaded_file(id, &file, "different-entry")
+                .is_err()
+        );
+        assert!(store.delete_uploaded_file(id, &file.artifact_id).is_err());
+
+        let cancelled = store
+            .write_uploaded_file(id, "cancelled.txt", "text/plain", b"cancelled", limits)
+            .unwrap();
+        store.pin_uploaded_file(id, &cancelled, "submit-2").unwrap();
+        store
+            .release_uploaded_file_pin(id, &cancelled.artifact_id, "submit-2")
+            .unwrap();
+        assert_eq!(store.delete_uncommitted_uploaded_files(id).unwrap(), 1);
+        let orphan = store
+            .write_uploaded_file(id, "orphan.txt", "text/plain", b"orphan", limits)
+            .unwrap();
+        store.pin_uploaded_file(id, &orphan, "lost-owner").unwrap();
+        assert_eq!(store.reconcile_uploaded_file_pins(id, &[]).unwrap(), 1);
+        assert_eq!(store.delete_uncommitted_uploaded_files(id).unwrap(), 1);
+
+        let retry = store
+            .write_uploaded_file(id, "retry.txt", "text/plain", b"retry", limits)
+            .unwrap();
+        store
+            .bind_uploaded_file(id, &retry, "failed-entry")
+            .unwrap();
+        store.bind_uploaded_file(id, &retry, "retry-entry").unwrap();
+        assert_eq!(store.delete_uncommitted_uploaded_files(id).unwrap(), 1);
+        assert_eq!(store.read_uploaded_file(id, &bound).unwrap(), b"pending");
+    }
+
+    #[test]
+    fn canonical_attachment_limits_integrity_and_session_fences_are_preserved() {
+        let root = tempfile::tempdir().unwrap();
+        let store = crate::CombinedStore::new(WorkerSessionStore::new(root.path()).unwrap(), ());
+        let id = new_session_id();
+        let limits = UploadedFileLimits {
+            max_file_bytes: 8,
+            max_session_bytes: 8,
+        };
+        store.create_segment(id, new_segment_id(), &[]).unwrap();
+        assert!(matches!(
+            store.write_uploaded_file(id, "../escape", "text/plain", b"x", limits),
+            Err(StoreError::InvalidUploadedFileName)
+        ));
+        assert!(matches!(
+            store.write_uploaded_file(id, "image.png", "image/png", b"invalid", limits),
+            Err(StoreError::ArtifactIntegrityMismatch)
+        ));
+        assert!(matches!(
+            store.write_uploaded_file(id, "large.txt", "text/plain", b"123456789", limits),
+            Err(StoreError::ArtifactTooLarge)
+        ));
+        let file = store
+            .write_uploaded_file(id, "notes.txt", "text/plain", b"1234", limits)
+            .unwrap();
+        let mut forged = file.clone();
+        forged.sha256 = "invalid".into();
+        assert!(matches!(
+            store.read_uploaded_file(id, &forged),
+            Err(StoreError::ArtifactIntegrityMismatch)
+        ));
+        let other = new_session_id();
+        assert!(
+            store
+                .write_uploaded_file(other, "other.txt", "text/plain", b"x", limits)
+                .is_err()
+        );
+        assert!(store.pin_uploaded_file(other, &file, "owner").is_err());
+        assert!(
+            store
+                .delete_uploaded_file(other, &file.artifact_id)
+                .is_err()
+        );
+        store
+            .write_paste_artifact(id, "paste-entry", "1234", PasteArtifactLimits::default())
+            .unwrap();
+        assert!(matches!(
+            store.write_uploaded_file(id, "overflow.txt", "text/plain", b"x", limits),
+            Err(StoreError::ArtifactQuotaExceeded)
+        ));
+    }
+
     #[test]
     fn canonical_layout_and_single_session_invariant() {
         let root = tempfile::tempdir().unwrap();

crates/tui/src/app.rs

@@ -759,12 +759,6 @@ impl App {
             }
             return None;
         }
-        if self.worker_status != WorkerStatus::Idle {
-            self.push_error(
-                "Submit requires an idle Worker; use Alt-n Notify for work already in progress.",
-            );
-            return None;
-        }
         self.record_input_history(segments.clone());
         self.input.clear();
         Some(self.method_for_run(segments))
@@ -779,9 +773,7 @@ impl App {
             .iter()
             .any(|segment| matches!(segment, Segment::UploadedFile { .. }))
         {
-            self.push_error(
-                "Notify accepts text only; remove attachments or wait until the Worker is idle to Submit.",
-            );
+            self.push_error("Notify accepts text only; remove attachments or queue a Submit.");
             return None;
         }
         let message = Segment::flatten_to_text(&segments);
@@ -3522,22 +3514,16 @@ mod completion_flow_tests {
     }
 
     #[test]
-    fn running_submit_is_rejected_locally_and_preserves_the_draft() {
+    fn running_submit_is_sent_to_the_worker_and_not_queued_locally() {
         let mut app = App::new("test".into());
         app.set_worker_status(WorkerStatus::Running);
-        insert_text(&mut app, "advisory context");
+        insert_text(&mut app, "queued turn");
 
         let method = app.submit_input();
 
-        assert!(method.is_none());
+        assert!(matches!(method, Some(Method::Submit { .. })));
         assert_eq!(app.queued_input_count(), 0);
-        assert_eq!(input_text(&app), "advisory context");
-        assert!(app.blocks.iter().any(|block| matches!(
-            block,
-            Block::Alert { message, .. }
-                if message.contains("requires an idle Worker")
-                    && message.contains("Alt-n Notify")
-        )));
+        assert_eq!(input_text(&app), "");
     }
 
     #[test]
@@ -3569,14 +3555,15 @@ mod completion_flow_tests {
     }
 
     #[test]
-    fn paused_non_empty_submit_is_rejected_locally_and_preserves_the_draft() {
+    fn paused_non_empty_submit_is_sent_without_resuming() {
         let mut app = App::new("test".into());
         app.set_worker_status(WorkerStatus::Paused);
-        insert_text(&mut app, "different request");
+        insert_text(&mut app, "next turn");
 
-        assert!(app.submit_input().is_none());
-        assert_eq!(input_text(&app), "different request");
+        assert!(matches!(app.submit_input(), Some(Method::Submit { .. })));
+        assert_eq!(app.worker_status, WorkerStatus::Paused);
         assert_eq!(app.queued_input_count(), 0);
+        assert_eq!(input_text(&app), "");
     }
 
     #[test]
@@ -4634,18 +4621,13 @@ mod completion_flow_tests {
     }
 
     #[test]
-    fn input_history_records_only_an_idle_submit_after_running_rejection() {
+    fn input_history_records_running_submits_and_suppresses_consecutive_duplicates() {
         let mut app = App::new("test".into());
-        app.set_worker_status(WorkerStatus::Running);
+        app.running = true;
 
         for c in "repeat".chars() {
             app.insert_char(c);
         }
-        assert!(app.submit_input().is_none());
-        assert_eq!(app.input_history_len(), 0);
-        assert_eq!(input_text(&app), "repeat");
-
-        app.set_worker_status(WorkerStatus::Idle);
         assert!(app.submit_input().is_some());
         assert_eq!(app.input_history_len(), 1);
         assert_eq!(app.queued_input_count(), 0);

crates/tui/src/console/mod.rs

@@ -1694,8 +1694,8 @@ mod tests {
         assert!(connection.awaiting_attachment_acceptance.is_empty());
     }
 
-    #[test]
-    fn running_attachment_submit_is_rejected_without_consuming_the_draft() {
+    #[tokio::test]
+    async fn running_attachment_submit_failure_restores_draft_without_exiting_console() {
         let file = UploadedFileRef {
             artifact_id: "artifact-queued".into(),
             file_name: "queued.txt".into(),
@@ -1706,7 +1706,7 @@ mod tests {
             sha256: "a".repeat(64),
             source_entry_id: None,
         };
-        let connection = ConsoleConnection {
+        let mut connection = ConsoleConnection {
             client: Client::new(FailOnceSocket {
                 fail_next_send: true,
             }),
@@ -1721,10 +1721,13 @@ mod tests {
         let mut app = App::new("worker".into());
         app.set_worker_status(WorkerStatus::Running);
         app.input.insert_str("queued inspect");
-        assert!(
-            app.submit_input().is_none(),
-            "running Submit must be rejected before transport"
-        );
+        let method = app
+            .submit_input()
+            .expect("running Submit is sent immediately");
+
+        send_console_method(&mut app, &mut connection, &method)
+            .await
+            .unwrap();
 
         assert_eq!(app.input.plain_text(), "queued inspect");
         assert_eq!(connection.pending_attachments, vec![file]);
@@ -2044,7 +2047,7 @@ mod tests {
     }
 
     #[test]
-    fn running_enter_rejects_submit_and_preserves_the_draft() {
+    fn running_enter_sends_submit_to_worker() {
         let mut app = App::new("agent".to_string());
         app.set_worker_status(WorkerStatus::Running);
         for c in "queued".chars() {
@@ -2057,10 +2060,13 @@ mod tests {
             );
         }
 
-        assert!(handle_key(&mut app, KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)).is_none());
+        assert!(matches!(
+            handle_key(&mut app, KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)),
+            Some(Method::Submit { .. })
+        ));
 
         assert_eq!(app.queued_input_count(), 0);
-        assert_eq!(input_text(&app), "queued");
+        assert_eq!(input_text(&app), "");
     }
 
     #[test]

crates/tui/src/ui.rs

@@ -1868,7 +1868,7 @@ fn draw_status(frame: &mut Frame, app: &App, area: Rect) {
                 .add_modifier(Modifier::BOLD),
         ));
         spans.push(Span::styled(
-            " — Enter to resume, Ctrl-X to cancel, Alt-n to notify",
+            " — Enter on empty input to resume, type + Enter to queue, Ctrl-X to cancel",
             Style::default().fg(Color::DarkGray),
         ));
     } else if !app.running {

crates/worker-runtime/src/worker_backend.rs

@@ -4980,7 +4980,7 @@ mod tests {
     }
 
     #[test]
-    fn running_worker_replays_retry_but_rejects_a_new_submit() {
+    fn running_worker_replays_retry_and_queues_a_new_submit() {
         let client = MockClient::sequential(vec![MockResponse::Hang(vec![])]);
         let runtime_base = tempfile::tempdir().unwrap();
         let cwd = tempfile::tempdir().unwrap();
@@ -5024,11 +5024,15 @@ mod tests {
 
         let mut second_input = WorkerInput::user("second");
         second_input.submission_request_id = Some("request-second".into());
-        let error = runtime
-            .send_input(&detail.worker_ref, second_input)
-            .expect_err("a new Submit must be rejected while the Worker is running");
-        assert!(error.to_string().contains("requires an idle Worker"));
-        assert!(error.to_string().contains("use Notify"));
+        let second = runtime
+            .send_input(&detail.worker_ref, second_input.clone())
+            .expect("a new Submit must queue while the Worker is running");
+        assert_eq!(
+            second.submission.as_ref().map(|ack| ack.disposition),
+            Some(protocol::SubmissionDisposition::Queued)
+        );
+        let retry = runtime.send_input(&detail.worker_ref, second_input).unwrap();
+        assert_eq!(retry.submission, second.submission);
     }
 
     #[test]

crates/worker/src/controller.rs

@@ -1472,51 +1472,6 @@ where
     Ok(workdir_for_view)
 }
 
-const SUBMIT_REQUIRES_IDLE_MESSAGE: &str =
-    "Submit requires an idle Worker; use Notify for work already in progress";
-
-fn reject_submit_while_not_idle(
-    submission_request_id: String,
-    working_event_tx: &broadcast::Sender<Event>,
-) {
-    let _ = working_event_tx.send(Event::SubmissionRejected {
-        submission_request_id,
-        message: SUBMIT_REQUIRES_IDLE_MESSAGE.into(),
-    });
-}
-
-fn replay_or_reject_submit_while_not_idle<St>(
-    pending_submissions: &PendingSubmissionHandle<St>,
-    submission_request_id: String,
-    input: Vec<protocol::Segment>,
-    source_namespace: String,
-    working_event_tx: &broadcast::Sender<Event>,
-) where
-    St: session_store::Store + Clone,
-{
-    let request_id = submission_request_id.clone();
-    match pending_submissions.replay_acceptance_from_source(
-        submission_request_id,
-        input,
-        source_namespace,
-    ) {
-        Ok(Some(acceptance)) => {
-            let _ = working_event_tx.send(Event::SubmissionAccepted {
-                submission_request_id: acceptance.submission_request_id,
-                submission_id: acceptance.submission_id,
-                disposition: acceptance.disposition,
-            });
-        }
-        Ok(None) => reject_submit_while_not_idle(request_id, working_event_tx),
-        Err(error) => {
-            let _ = working_event_tx.send(Event::SubmissionRejected {
-                submission_request_id: request_id,
-                message: error.to_string(),
-            });
-        }
-    }
-}
-
 fn durably_accept_method_while_busy<St>(
     method: Method,
     pending_submissions: &PendingSubmissionHandle<St>,
@@ -1530,13 +1485,28 @@ where
             submission_request_id,
             input,
         } => {
-            replay_or_reject_submit_while_not_idle(
-                pending_submissions,
+            let request_id = submission_request_id.clone();
+            match pending_submissions.accept_from_source(
                 submission_request_id,
                 input,
                 pending_submissions.direct_client_namespace(),
-                working_event_tx,
-            );
+                session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
+                false,
+            ) {
+                Ok(acceptance) => {
+                    let _ = working_event_tx.send(Event::SubmissionAccepted {
+                        submission_request_id: acceptance.submission_request_id,
+                        submission_id: acceptance.submission_id,
+                        disposition: acceptance.disposition,
+                    });
+                }
+                Err(error) => {
+                    let _ = working_event_tx.send(Event::SubmissionRejected {
+                        submission_request_id: request_id,
+                        message: error.to_string(),
+                    });
+                }
+            }
             None
         }
         Method::SubmitTracked {
@@ -1544,14 +1514,30 @@ where
             input,
             source,
         } => {
-            let (source_namespace, _) = resolved_input_source(pending_submissions, &source);
-            replay_or_reject_submit_while_not_idle(
-                pending_submissions,
+            let request_id = submission_request_id.clone();
+            let (source_namespace, provenance) =
+                resolved_input_source(pending_submissions, &source);
+            match pending_submissions.accept_from_source(
                 submission_request_id,
                 input,
                 source_namespace,
-                working_event_tx,
-            );
+                provenance,
+                false,
+            ) {
+                Ok(acceptance) => {
+                    let _ = working_event_tx.send(Event::SubmissionAccepted {
+                        submission_request_id: acceptance.submission_request_id,
+                        submission_id: acceptance.submission_id,
+                        disposition: acceptance.disposition,
+                    });
+                }
+                Err(error) => {
+                    let _ = working_event_tx.send(Event::SubmissionRejected {
+                        submission_request_id: request_id,
+                        message: error.to_string(),
+                    });
+                }
+            }
             None
         }
         Method::Notify {
@@ -1964,22 +1950,13 @@ async fn controller_loop<C, St>(
                 submission_request_id,
                 input,
             } => {
-                if shared_state.catalog_status() != WorkerStatus::Idle {
-                    replay_or_reject_submit_while_not_idle(
-                        &pending_submissions,
-                        submission_request_id,
-                        input,
-                        pending_submissions.direct_client_namespace(),
-                        &working_event_tx,
-                    );
-                    continue;
-                }
                 let request_id = submission_request_id.clone();
-                match pending_submissions.accept_idle_from_source(
+                match pending_submissions.accept_from_source(
                     submission_request_id,
                     input,
                     pending_submissions.direct_client_namespace(),
                     session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
+                    shared_state.catalog_status() == WorkerStatus::Idle,
                 ) {
                     Ok(acceptance) => {
                         let _ = working_event_tx.send(Event::SubmissionAccepted {
@@ -2004,24 +1981,15 @@ async fn controller_loop<C, St>(
                 input,
                 source,
             } => {
+                let request_id = submission_request_id.clone();
                 let (source_namespace, provenance) =
                     resolved_input_source(&pending_submissions, &source);
-                if shared_state.catalog_status() != WorkerStatus::Idle {
-                    replay_or_reject_submit_while_not_idle(
-                        &pending_submissions,
-                        submission_request_id,
-                        input,
-                        source_namespace,
-                        &working_event_tx,
-                    );
-                    continue;
-                }
-                let request_id = submission_request_id.clone();
-                match pending_submissions.accept_idle_from_source(
+                match pending_submissions.accept_from_source(
                     submission_request_id,
                     input,
                     source_namespace,
                     provenance,
+                    shared_state.catalog_status() == WorkerStatus::Idle,
                 ) {
                     Ok(acceptance) => {
                         let _ = working_event_tx.send(Event::SubmissionAccepted {
@@ -2978,28 +2946,64 @@ where
                         submission_request_id,
                         input,
                     }) => {
-                        replay_or_reject_submit_while_not_idle(
-                            pending_submissions,
+                        let request_id = submission_request_id.clone();
+                        match pending_submissions.accept_from_source(
                             submission_request_id,
                             input,
                             pending_submissions.direct_client_namespace(),
-                            working_event_tx,
-                        );
+                            session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
+                            false,
+                        ) {
+                            Ok(acceptance) => {
+                                let _ = working_event_tx.send(Event::SubmissionAccepted {
+                                    submission_request_id: acceptance.submission_request_id,
+                                    submission_id: acceptance.submission_id,
+                                    disposition: acceptance.disposition,
+                                });
+                                let _ = working_event_tx.send(Event::PendingSubmissionsChanged {
+                                    pending: pending_submissions.snapshot(),
+                                });
+                            }
+                            Err(error) => {
+                                let _ = working_event_tx.send(Event::SubmissionRejected {
+                                    submission_request_id: request_id,
+                                    message: error.to_string(),
+                                });
+                            }
+                        }
                     }
                     Some(Method::SubmitTracked {
                         submission_request_id,
                         input,
                         source,
                     }) => {
-                        let (source_namespace, _) =
+                        let request_id = submission_request_id.clone();
+                        let (source_namespace, provenance) =
                             resolved_input_source(pending_submissions, &source);
-                        replay_or_reject_submit_while_not_idle(
-                            pending_submissions,
+                        match pending_submissions.accept_from_source(
                             submission_request_id,
                             input,
                             source_namespace,
-                            working_event_tx,
-                        );
+                            provenance,
+                            false,
+                        ) {
+                            Ok(acceptance) => {
+                                let _ = working_event_tx.send(Event::SubmissionAccepted {
+                                    submission_request_id: acceptance.submission_request_id,
+                                    submission_id: acceptance.submission_id,
+                                    disposition: acceptance.disposition,
+                                });
+                                let _ = working_event_tx.send(Event::PendingSubmissionsChanged {
+                                    pending: pending_submissions.snapshot(),
+                                });
+                            }
+                            Err(error) => {
+                                let _ = working_event_tx.send(Event::SubmissionRejected {
+                                    submission_request_id: request_id,
+                                    message: error.to_string(),
+                                });
+                            }
+                        }
                     }
                     Some(Method::Resume { command }) => {
                         if let Err(disposition) =
@@ -3560,35 +3564,44 @@ mod tests {
     }
 
     #[test]
-    fn busy_submit_is_rejected_without_queueing() {
+    fn busy_submit_is_durable_and_idempotent_before_shutdown() {
         let dir = tempfile::tempdir().unwrap();
         let pending = PendingSubmissionHandle::for_test(dir.path());
-        let (event_tx, mut event_rx) = broadcast::channel(4);
-
-        assert!(
-            durably_accept_method_while_busy(
-                Method::SubmitTracked {
-                    submission_request_id: "request-1".into(),
-                    input: vec![protocol::Segment::text("follow-up")],
-                    source: protocol::AuthenticatedInputSource::Backend {
-                        operation_id: "operation-1".into(),
-                    },
+        let (event_tx, mut event_rx) = broadcast::channel(8);
+        for method in [
+            Method::submit_text("direct-request", "next direct turn"),
+            Method::SubmitTracked {
+                submission_request_id: "tracked-request".into(),
+                input: vec![protocol::Segment::text("next tracked turn")],
+                source: protocol::AuthenticatedInputSource::Backend {
+                    operation_id: "operation-1".into(),
                 },
-                &pending,
-                &event_tx,
-            )
-            .is_none()
+            },
+        ] {
+            for _ in 0..2 {
+                assert!(
+                    durably_accept_method_while_busy(method.clone(), &pending, &event_tx).is_none()
+                );
+                assert!(matches!(
+                    event_rx.try_recv().unwrap(),
+                    Event::SubmissionAccepted {
+                        disposition: protocol::SubmissionDisposition::Queued,
+                        ..
+                    }
+                ));
+            }
+        }
+        assert_eq!(pending.snapshot().submissions.len(), 2);
+        let entries = pending.persisted_entries_for_test();
+        assert_eq!(
+            entries
+                .iter()
+                .filter(|entry| matches!(entry,
+                    LogEntry::Extension { domain, .. } if domain == "worker.pending_activations.v1"
+                ))
+                .count(),
+            2
         );
-        assert!(matches!(
-            event_rx.try_recv().unwrap(),
-            Event::SubmissionRejected {
-                submission_request_id,
-                message,
-            } if submission_request_id == "request-1"
-                && message == SUBMIT_REQUIRES_IDLE_MESSAGE
-        ));
-        assert!(pending.snapshot().submissions.is_empty());
-        assert!(pending.persisted_entries_for_test().is_empty());
     }
 
     #[test]

crates/worker/src/feature/builtin/manage_worker.rs

@@ -850,7 +850,7 @@ impl WorkerOperation {
                 "Spawn a Backend/Runtime Worker session in one existing Workspace Workdir. The model-facing input remains singular and the shared lifecycle wraps it as the canonical attachment collection under the stable `workdir` alias. The Workdir id is authority; filesystem paths and Runtime URLs are not accepted. `initial_submit` carries the normal typed user submission. After the Orchestrator has committed a Ticket to `inprogress`, set `ticket_id` with a Flow segment in `initial_submit` to atomically assign the new Coder Worker; the operation id is derived from the durable tool call rather than model input."
             }
             Self::SendInput => {
-                "Start a fresh turn by sending user input to a known Idle Worker. Running or Paused targets reject Submit; use WorkerNotify for advisory information during work already in progress."
+                "Send user input to a known Runtime Worker. Idle Workers can start a fresh turn; Running or Paused Workers durably queue it for a later turn. Use WorkerNotify for advisory information during work already in progress. Direct SubWorkers require Idle."
             }
             Self::Notify => {
                 "Send an advisory notification to a known Runtime Worker so information can be incorporated into work already in progress without creating a queued Submit."

crates/worker/src/prompt/catalog.rs

@@ -914,7 +914,8 @@ mod tests {
                 "missing canonical tool {name}"
             );
         }
-        assert!(orchestration.contains("Idle-only Submit"));
+        assert!(orchestration.contains("durably queue it for a later turn"));
+        assert!(orchestration.contains("Direct SubWorkers still require Idle"));
         assert!(orchestration.contains("use `WorkerNotify` instead"));
         assert!(orchestration.contains("do not create a queued Submit"));
         for alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] {

crates/worker/src/worker.rs

@@ -1562,60 +1562,7 @@ where
         )
     }
 
-    pub(crate) fn accept_idle_from_source(
-        &self,
-        submission_request_id: String,
-        input: Vec<Segment>,
-        source_namespace: String,
-        provenance: WorkerHistoryProvenance,
-    ) -> Result<SubmissionAcceptance, PendingSubmissionError> {
-        self.accept_from_source(
-            submission_request_id,
-            input,
-            source_namespace,
-            provenance,
-            true,
-        )
-    }
-
-    pub(crate) fn replay_acceptance_from_source(
-        &self,
-        submission_request_id: String,
-        input: Vec<Segment>,
-        source_namespace: String,
-    ) -> Result<Option<SubmissionAcceptance>, PendingSubmissionError> {
-        if submission_request_id.trim().is_empty() {
-            return Err(PendingSubmissionError::EmptyRequestId);
-        }
-        if submission_request_id.len() > MAX_ACTIVATION_REQUEST_ID_BYTES {
-            return Err(PendingSubmissionError::RequestIdLimit);
-        }
-        if input.is_empty() {
-            return Err(PendingSubmissionError::EmptyInput);
-        }
-        let payload_digest = submission_payload_digest(&input);
-        let current = self
-            .state
-            .lock()
-            .expect("pending activation state poisoned");
-        let Some(receipt) = current.receipts.iter().find(|receipt| {
-            receipt.submission_request_id == submission_request_id
-                && receipt.source_namespace == source_namespace
-        }) else {
-            return Ok(None);
-        };
-        if receipt.payload_digest != payload_digest {
-            return Err(PendingSubmissionError::IdempotencyConflict);
-        }
-        Ok(Some(SubmissionAcceptance {
-            submission_request_id,
-            submission_id: receipt.submission_id.clone(),
-            disposition: receipt.disposition,
-            activation: None,
-        }))
-    }
-
-    fn accept_from_source(
+    pub(crate) fn accept_from_source(
         &self,
         submission_request_id: String,
         input: Vec<Segment>,

crates/worker/tests/controller_test.rs

@@ -318,6 +318,95 @@ async fn spawn_controller(worker: Worker<MockClient, TestStore>) -> WorkerHandle
     handle
 }
 
+#[tokio::test]
+async fn canonical_worker_handle_uploads_and_submits_attachment() {
+    let root = tempfile::tempdir().unwrap();
+    let manifest = WorkerManifest::from_toml(MANIFEST_TOML).unwrap();
+    let session = session_store::WorkerSessionStore::new(root.path().join("session")).unwrap();
+    let store = CombinedStore::new(
+        session.clone(),
+        session_store::WorkerAggregateStore::new(root.path(), manifest.worker.name.clone())
+            .unwrap(),
+    );
+    let client = MockClient::new(simple_text_events());
+    let engine = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
+        client.clone(),
+    );
+    let worker = Worker::new(
+        manifest,
+        engine,
+        store,
+        WorkerWorkspaceContext::local_filesystem(None),
+        WorkerFilesystemAuthority::local(root.path().to_owned(), root.path().to_owned()),
+        manifest::Scope::writable(root.path()).unwrap(),
+    )
+    .await
+    .unwrap();
+    let (handle, shutdown_rx) = WorkerController::spawn(
+        worker,
+        &root.path().join("runtime"),
+        &root.path().join("bash-output"),
+    )
+    .await
+    .unwrap();
+    let mut rx = handle.subscribe();
+    let file = handle
+        .upload_file_with_context(
+            "notes.txt",
+            "text/plain",
+            b"attached context",
+            &session_store::UploadedFileUploadContext {
+                upload_id: "upload-1".into(),
+                principal_id: "account-1".into(),
+                workspace_id: "workspace-1".into(),
+                runtime_id: "runtime-1".into(),
+                worker_id: "worker-1".into(),
+            },
+        )
+        .unwrap();
+    handle
+        .send(Method::Submit {
+            submission_request_id: "attachment-submit".into(),
+            input: vec![protocol::Segment::UploadedFile { file: file.clone() }],
+        })
+        .await
+        .unwrap();
+    assert!(
+        drain_until(&mut rx, std::time::Duration::from_secs(2), |event| {
+            matches!(
+                event,
+                Event::RunEnd {
+                    result: protocol::RunResult::Finished
+                }
+            )
+        })
+        .await
+    );
+    let requests = client.captured_requests();
+    assert_eq!(requests.len(), 1);
+    assert!(requests[0].items.iter().any(|item| {
+        item.as_text()
+            .is_some_and(|text| text.contains("notes.txt") && text.contains(&file.artifact_id))
+    }));
+    assert_eq!(handle.delete_uncommitted_uploaded_files().unwrap(), 0);
+    use session_store::Store;
+    let (bound, content) = session
+        .read_uploaded_file_by_id(session.session_id().unwrap().unwrap(), &file.artifact_id)
+        .unwrap();
+    assert!(bound.source_entry_id.is_some());
+    assert_eq!(content, b"attached context");
+    handle
+        .send(Method::Shutdown {
+            command: worker_command(&handle),
+        })
+        .await
+        .unwrap();
+    tokio::time::timeout(std::time::Duration::from_secs(2), shutdown_rx)
+        .await
+        .unwrap()
+        .unwrap();
+}
+
 #[tokio::test]
 async fn controller_grants_read_scope_for_exact_bash_output_directory() {
     let worker = make_worker(MockClient::new(simple_text_events())).await;
@@ -1446,7 +1535,7 @@ async fn events_are_broadcast() {
 }
 
 #[tokio::test]
-async fn submit_requires_an_idle_worker() {
+async fn submit_while_running_or_paused_is_durably_queued() {
     async fn wait_for_rejection(
         rx: &mut tokio::sync::broadcast::Receiver<Event>,
         expected_request_id: &str,
@@ -1492,7 +1581,13 @@ async fn submit_requires_an_idle_worker() {
         LlmEvent::text_block_start(0),
         LlmEvent::text_delta(0, "slow..."),
     ];
-    let client = MockClient::sequential(vec![MockResponse::Hang(events)]);
+    let client = MockClient::sequential(vec![
+        MockResponse::Hang(events),
+        MockResponse::Complete(simple_text_events()),
+        MockResponse::Complete(simple_text_events()),
+        MockResponse::Complete(simple_text_events()),
+    ]);
+    let captured = client.clone();
     let worker = make_worker(client).await;
     let handle = spawn_controller(worker).await;
     let mut rx = handle.subscribe();
@@ -1517,9 +1612,27 @@ async fn submit_requires_an_idle_worker() {
         .await
         .unwrap();
 
-    let running_rejection = wait_for_rejection(&mut rx, "request-running").await;
-    assert!(running_rejection.contains("requires an idle Worker"));
-    assert!(running_rejection.contains("use Notify"));
+    assert_eq!(
+        wait_for_acceptance(&mut rx, "request-running").await,
+        protocol::SubmissionDisposition::Queued
+    );
+    handle
+        .send(Method::submit_text("request-running", "second"))
+        .await
+        .unwrap();
+    assert_eq!(
+        wait_for_acceptance(&mut rx, "request-running").await,
+        protocol::SubmissionDisposition::Queued
+    );
+    handle
+        .send(Method::submit_text("request-running", "conflict"))
+        .await
+        .unwrap();
+    assert!(
+        wait_for_rejection(&mut rx, "request-running")
+            .await
+            .contains("different payload")
+    );
 
     handle.send(Method::ListPendingSubmissions).await.unwrap();
     let pending = tokio::time::timeout(std::time::Duration::from_secs(1), async {
@@ -1531,7 +1644,7 @@ async fn submit_requires_an_idle_worker() {
     })
     .await
     .expect("pending snapshot");
-    assert!(pending.submissions.is_empty());
+    assert_eq!(pending.submissions.len(), 1);
 
     handle
         .send(Method::Pause {
@@ -1545,10 +1658,86 @@ async fn submit_requires_an_idle_worker() {
         .await
         .unwrap();
 
-    let paused_rejection = wait_for_rejection(&mut rx, "request-paused").await;
-    assert!(paused_rejection.contains("requires an idle Worker"));
-    assert!(paused_rejection.contains("use Notify"));
+    assert_eq!(
+        wait_for_acceptance(&mut rx, "request-paused").await,
+        protocol::SubmissionDisposition::Queued
+    );
     assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused);
+    handle.send(Method::ListPendingSubmissions).await.unwrap();
+    let snapshot = tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        loop {
+            if let Ok(Event::PendingSubmissionsChanged { pending }) = rx.recv().await {
+                break pending;
+            }
+        }
+    })
+    .await
+    .expect("paused pending snapshot");
+    assert_eq!(snapshot.submissions.len(), 2);
+    assert_eq!(snapshot.head_id, pending.head_id);
+    handle
+        .send(Method::ContinuePending {
+            expected_revision: snapshot.revision,
+            expected_head_id: snapshot.head_id.unwrap(),
+        })
+        .await
+        .unwrap();
+    assert!(
+        drain_until(&mut rx, std::time::Duration::from_secs(1), |event| {
+            matches!(event, Event::Error { message, .. } if message.contains("Resume or Cancel"))
+        })
+        .await
+    );
+    assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused);
+
+    // Resume completes the original run, then normal completion drains each
+    // queued input exactly once and in acceptance order.
+    handle
+        .send(Method::Resume {
+            command: worker_command(&handle),
+        })
+        .await
+        .unwrap();
+    for _ in 0..3 {
+        assert!(
+            drain_until(&mut rx, std::time::Duration::from_secs(2), |event| {
+                matches!(
+                    event,
+                    Event::RunEnd {
+                        result: protocol::RunResult::Finished
+                    }
+                )
+            })
+            .await
+        );
+    }
+    wait_for_status(&handle, WorkerStatus::Idle).await;
+    let requests = captured.captured_requests();
+    assert_eq!(requests.len(), 4);
+    let last_inputs: Vec<_> = requests[1..]
+        .iter()
+        .map(|request| {
+            request
+                .items
+                .iter()
+                .rev()
+                .find_map(|item| match item {
+                    Item::Message {
+                        role: agen::Role::User,
+                        content,
+                        ..
+                    } => Some(
+                        content
+                            .iter()
+                            .map(|part| part.as_text())
+                            .collect::<String>(),
+                    ),
+                    _ => None,
+                })
+                .unwrap()
+        })
+        .collect();
+    assert_eq!(last_inputs, vec!["first", "second", "third"]);
 }
 
 #[tokio::test]
@@ -2879,12 +3068,12 @@ async fn pause_then_resume_preserves_notifications_and_history_consistency() {
     assert!(!has_tool_call, "no orphan tool_call in history");
 }
 
-/// Paused with an orphan `tool_use` in history must reject a fresh Submit.
-/// After explicit Cancel returns the Worker to Idle, a new Submit must produce
+/// Paused with an orphan `tool_use` in history must queue a fresh Submit.
+/// After explicit Cancel and ContinuePending, the queued Submit must produce
 /// a wire-valid next LLM request: the orphan is closed with a synthetic
 /// `tool_result`, a system note is inserted, and the new user input is appended.
 #[tokio::test]
-async fn paused_submit_is_rejected_before_cancel_and_fresh_run() {
+async fn paused_submit_waits_for_cancel_and_continue_pending() {
     // Response 1: emit a tool_use block (complete with stop) targeting
     // our hanging tool. The Engine commits the ToolCall to history,
     // then parks inside `execute_tools` waiting on the tool — which is
@@ -2966,20 +3155,20 @@ async fn paused_submit_is_rejected_before_cancel_and_fresh_run() {
         drain_until(&mut rx, std::time::Duration::from_secs(2), |event| {
             matches!(
                 event,
-                Event::SubmissionRejected {
+                Event::SubmissionAccepted {
                     submission_request_id,
-                    message,
+                    disposition: protocol::SubmissionDisposition::Queued,
+                    ..
                 } if submission_request_id == &paused_request_id
-                    && message.contains("requires an idle Worker")
             )
         })
         .await,
-        "Paused Submit must be rejected"
+        "Paused Submit must be queued"
     );
     assert_eq!(
         client_for_assert.captured_requests().len(),
         1,
-        "rejected Paused Submit must not start another LLM request"
+        "queued Paused Submit must not start another LLM request"
     );
 
     handle
@@ -2990,13 +3179,25 @@ async fn paused_submit_is_rejected_before_cancel_and_fresh_run() {
         .unwrap();
     wait_for_status(&handle, WorkerStatus::Idle).await;
 
-    // Once explicitly returned to Idle, fresh user input runs interrupt prep,
-    // which closes the orphan and inserts a system note before the user message.
+    // Cancel must not start the queued turn. Explicit ContinuePending runs
+    // interrupt prep, closing the orphan before committing the queued input.
+    assert_eq!(client_for_assert.captured_requests().len(), 1);
+    handle.send(Method::ListPendingSubmissions).await.unwrap();
+    let pending = tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        loop {
+            if let Ok(Event::PendingSubmissionsChanged { pending }) = rx.recv().await {
+                break pending;
+            }
+        }
+    })
+    .await
+    .expect("pending input retained after Cancel");
+    assert_eq!(pending.submissions.len(), 1);
     handle
-        .send(Method::submit_text(
-            protocol::new_submission_request_id(),
-            "new request",
-        ))
+        .send(Method::ContinuePending {
+            expected_revision: pending.revision,
+            expected_head_id: pending.head_id.unwrap(),
+        })
         .await
         .unwrap();
     assert!(
@@ -3007,7 +3208,7 @@ async fn paused_submit_is_rejected_before_cancel_and_fresh_run() {
             }
         ))
         .await,
-        "expected RunEnd::Finished after Cancel→Submit"
+        "expected RunEnd::Finished after Cancel→ContinuePending"
     );
 
     // The second LLM request carries the closure chain. Walk its items