diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index a9bcc7a5d..107d08ab2 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -219,9 +219,9 @@ impl From 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 { diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index f04a1245a..93ab6b62d 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/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 { + 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 { + 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, 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), 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 { + 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 { + self.session_store + .reconcile_uploaded_file_pins(session_id, live_owner_ids) + } + fn delete_uploaded_file( + &self, + session_id: SessionId, + artifact_id: &str, + ) -> Result { + self.session_store + .delete_uploaded_file(session_id, artifact_id) + } + fn delete_uncommitted_uploaded_files( + &self, + session_id: SessionId, + ) -> Result { + self.session_store + .delete_uncommitted_uploaded_files(session_id) + } + fn copy_committed_uploaded_files( + &self, + source_session_id: SessionId, + target_session_id: SessionId, + ) -> Result { + self.session_store + .copy_committed_uploaded_files(source_session_id, target_session_id) + } fn append_trace( &self, session_id: SessionId, diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index a30aedde2..b8a5bdf29 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/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 { + 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 { + 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 { + 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, 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), 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 { + 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 { + 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 { + 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 { + 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(); diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 0c0147387..50a94a225 100644 --- a/crates/tui/src/app.rs +++ b/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); diff --git a/crates/tui/src/console/mod.rs b/crates/tui/src/console/mod.rs index 93ca9251a..35ea1438e 100644 --- a/crates/tui/src/console/mod.rs +++ b/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] diff --git a/crates/tui/src/ui.rs b/crates/tui/src/ui.rs index 33d4b2756..39788ccdf 100644 --- a/crates/tui/src/ui.rs +++ b/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 { diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 4f62c9199..d46ee5502 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/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] diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index a48c5f59a..42200fd82 100644 --- a/crates/worker/src/controller.rs +++ b/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, -) { - let _ = working_event_tx.send(Event::SubmissionRejected { - submission_request_id, - message: SUBMIT_REQUIRES_IDLE_MESSAGE.into(), - }); -} - -fn replay_or_reject_submit_while_not_idle( - pending_submissions: &PendingSubmissionHandle, - submission_request_id: String, - input: Vec, - source_namespace: String, - working_event_tx: &broadcast::Sender, -) 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( method: Method, pending_submissions: &PendingSubmissionHandle, @@ -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( 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( 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] diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 371c18206..eab20bc41 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/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." diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 97e63b940..9d7bb468a 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/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"] { diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 666eddec1..76e2b255d 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1562,60 +1562,7 @@ where ) } - pub(crate) fn accept_idle_from_source( - &self, - submission_request_id: String, - input: Vec, - source_namespace: String, - provenance: WorkerHistoryProvenance, - ) -> Result { - 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, - source_namespace: String, - ) -> Result, 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, diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 0872affcc..1a3d5282e 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -318,6 +318,95 @@ async fn spawn_controller(worker: Worker) -> 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, 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::(), + ), + _ => 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 diff --git a/resources/prompts/common/worker-orchestration.md b/resources/prompts/common/worker-orchestration.md index 4505e85cb..50a47793f 100644 --- a/resources/prompts/common/worker-orchestration.md +++ b/resources/prompts/common/worker-orchestration.md @@ -4,7 +4,7 @@ When SubWorker-management tools are available, create direct children with `SubWorkerSpawn`, discover them with `WorkerList`, continue them with `WorkerSendInput`, and release their delegated authority with `WorkerStop`. Pass the exact `{ kind: "sub_worker", name }` subject returned by `WorkerList`; do not invent direct-only aliases. SubWorker notifications are background signals for the parent Worker to handle at a natural stopping point. Do not ignore routine follow-up, but do not interrupt the current user request unnecessarily. -`WorkerSendInput` is an Idle-only Submit that starts a fresh turn. Do not use it for a Worker that is Running or Paused. When `WorkerNotify` is available and information should be incorporated into work already in progress, use `WorkerNotify` instead; notifications are advisory and do not create a queued Submit. +For Runtime Workers, `WorkerSendInput` submits user input: Idle Workers can start a fresh turn, while Running or Paused Workers durably queue it for a later turn. Queueing does not implicitly resume a paused run; Cancel/error retains pending input until explicitly continued. Direct SubWorkers still require Idle for `WorkerSendInput`. When `WorkerNotify` is available and information should be incorporated into work already in progress, use `WorkerNotify` instead; notifications are advisory and do not create a queued Submit. The parent Worker does not need to keep a turn open or call tools solely to wait for a notification. Do not use `sleep` or polling loops just to wait for SubWorker output; if there is no useful immediate work, return control and handle the SubWorker when notified or when the user next asks. diff --git a/web/workspace/src/lib/workspace/console/ComposerInput.svelte b/web/workspace/src/lib/workspace/console/ComposerInput.svelte index f905af840..d17c407ca 100644 --- a/web/workspace/src/lib/workspace/console/ComposerInput.svelte +++ b/web/workspace/src/lib/workspace/console/ComposerInput.svelte @@ -24,6 +24,7 @@ import { onMount } from "svelte"; import type { Segment } from "$lib/generated/protocol.ts"; import { + handleComposerImagePaste, measureComposerPaste, type ComposerPasteMeasurement, } from "$lib/workspace/console/composer-paste.ts"; @@ -55,6 +56,7 @@ onchange?: (snapshot: ComposerDraftSnapshot) => void; onkeydown?: (event: KeyboardEvent) => void; onsubmit?: () => void; + onpasteimages?: (files: File[]) => void; } let { @@ -65,6 +67,7 @@ onchange, onkeydown, onsubmit, + onpasteimages, }: Props = $props(); let mountElement: HTMLDivElement; @@ -265,6 +268,7 @@ function handlePasteEvent(event: ClipboardEvent): boolean { if (disabled || view?.state.readOnly) return false; + if (handleComposerImagePaste(event, onpasteimages)) return true; const content = event.clipboardData?.getData("text/plain"); if (!content) return false; const measurement = measureComposerPaste(content); @@ -557,7 +561,7 @@ } .composer-input :global(.cm-editor) { - min-height: 5.35rem; + min-height: 2.65rem; max-height: 10rem; } @@ -566,8 +570,8 @@ } .composer-input :global(.cm-content) { - min-height: 5.35rem; - padding: 0.55rem 3.4rem 3rem 0.65rem; + min-height: 2.65rem; + padding: 0.55rem 0.65rem 0.35rem; line-height: 1.45; } diff --git a/web/workspace/src/lib/workspace/console/composer-delivery.test.ts b/web/workspace/src/lib/workspace/console/composer-delivery.test.ts index b2225e65e..ebf3bfa32 100644 --- a/web/workspace/src/lib/workspace/console/composer-delivery.test.ts +++ b/web/workspace/src/lib/workspace/console/composer-delivery.test.ts @@ -20,6 +20,77 @@ const base = { hasAttachments: false, }; +Deno.test("Queue accepts text and attachments while running or paused", () => { + for (const workerState of ["running", "paused"]) { + for ( + const payload of [ + { hasText: true, hasAttachments: false }, + { hasText: false, hasAttachments: true }, + { hasText: true, hasAttachments: true }, + ] + ) { + assertEquals( + canDeliverComposerDraft({ + ...base, + ...payload, + workerState, + delivery: "queue", + }), + true, + ); + } + } +}); + +Deno.test("Queue respects lifecycle, transport, send and empty-input fences", () => { + for (const workerState of ["idle", "stopped", "loading", "unknown"]) { + assertEquals( + canDeliverComposerDraft({ + ...base, + workerState, + delivery: "queue", + }), + false, + ); + } + for ( + const fence of [ + { protocolOpen: false }, + { sending: true }, + { hasText: false, hasAttachments: false }, + ] + ) { + const state = { + ...base, + ...fence, + workerState: "running", + delivery: "queue" as const, + }; + assertEquals(canDeliverComposerDraft(state), false); + let sent = false; + assertEquals( + sendComposerDelivery(state, "submit", () => { + sent = true; + }), + false, + ); + assertEquals(sent, false); + } +}); + +Deno.test("Queue dispatches Submit, not Notify", () => { + const sent: string[] = []; + assertEquals( + sendComposerDelivery( + { ...base, workerState: "running", delivery: "queue" }, + "submit", + (method) => sent.push(method), + ), + true, + ); + assertEquals(sent.join(","), "submit"); +}); + Deno.test("running Composer enables Notify but not Submit", () => { assertEquals( canDeliverComposerDraft({ diff --git a/web/workspace/src/lib/workspace/console/composer-delivery.ts b/web/workspace/src/lib/workspace/console/composer-delivery.ts index 7d908da02..f00f088e7 100644 --- a/web/workspace/src/lib/workspace/console/composer-delivery.ts +++ b/web/workspace/src/lib/workspace/console/composer-delivery.ts @@ -1,4 +1,4 @@ -export type ComposerDelivery = "submit" | "notify"; +export type ComposerDelivery = "submit" | "queue" | "notify"; export type ComposerDeliveryState = { delivery: ComposerDelivery; @@ -11,7 +11,8 @@ export type ComposerDeliveryState = { /** * Resolve whether the current Composer draft can use one delivery action. - * Immediate Submit is idle-only; Notify is running-only. + * Immediate Submit is idle-only; Queue submits to the busy Worker's FIFO. + * Notify is running-only and remains distinct from next-turn input. */ export function canDeliverComposerDraft(state: ComposerDeliveryState): boolean { if (!state.protocolOpen || state.sending) return false; @@ -20,6 +21,9 @@ export function canDeliverComposerDraft(state: ComposerDeliveryState): boolean { switch (state.delivery) { case "submit": return state.workerState === "idle" && hasInput; + case "queue": + return (state.workerState === "running" || state.workerState === "paused") && + hasInput; case "notify": return state.workerState === "running" && state.hasText && !state.hasAttachments; diff --git a/web/workspace/src/lib/workspace/console/composer-paste.ts b/web/workspace/src/lib/workspace/console/composer-paste.ts index ab205a872..73f7ccdd7 100644 --- a/web/workspace/src/lib/workspace/console/composer-paste.ts +++ b/web/workspace/src/lib/workspace/console/composer-paste.ts @@ -10,10 +10,58 @@ export interface ComposerPasteMeasurement { } export interface ComposerPasteEvent { - clipboardData: { getData(format: string): string } | null; + clipboardData: { + getData(format: string): string; + files?: ArrayLike; + items?: ArrayLike<{ + kind: string; + type: string; + getAsFile(): File | null; + }>; + } | null; preventDefault(): void; } +/** Keep repeated screenshots from colliding with unsubmitted image.png uploads. */ +export function namePastedImage(file: File): File { + return new File( + [file], + `pasted-${crypto.randomUUID()}-${file.name || "image"}`, + { + type: file.type, + lastModified: file.lastModified, + }, + ); +} + +/** + * Prefer image files over alternate text/HTML clipboard representations. + * Use items only as a fallback: browsers can expose the same image in both + * collections. Validation and upload remain the attachment handler's job. + */ +export function handleComposerImagePaste( + event: ComposerPasteEvent, + attachImages: ((files: File[]) => void) | undefined, + enabled = true, +): boolean { + if (!enabled || !attachImages || !event.clipboardData) return false; + const clipboard = event.clipboardData; + let images = Array.from(clipboard.files ?? []).filter((file) => + file.type.startsWith("image/") + ); + if (images.length === 0) { + images = Array.from(clipboard.items ?? []).flatMap((item) => { + if (item.kind !== "file" || !item.type.startsWith("image/")) return []; + const file = item.getAsFile(); + return file ? [file] : []; + }); + } + if (images.length === 0) return false; + event.preventDefault(); + attachImages(images); + return true; +} + /** * Count user-visible Unicode scalar values rather than UTF-16 code units. * JavaScript string iteration combines a valid surrogate pair into one value. diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 245b47400..c2ecb84e1 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -805,7 +805,6 @@ Deno.test("Worker Console composer keeps a compact bounded chip editor", async ( ) && consolePage.includes("scrollConsoleByPage") && consolePage.includes('class="composer-input-footer"') && - consolePage.includes("pointer-events: none") && consolePage.includes('class="composer-footer-slot"') && consolePage.includes('class="composer-send-button"') && consolePage.includes("pointer-events: auto") && @@ -818,6 +817,78 @@ Deno.test("Worker Console composer keeps a compact bounded chip editor", async ( ); }); +Deno.test("Worker Console composer separates the editor from its icon toolbar", async () => { + const consolePage = await Deno.readTextFile( + new URL( + "./../../../routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte", + import.meta.url, + ), + ); + const composerInput = await Deno.readTextFile( + new URL("./ComposerInput.svelte", import.meta.url), + ); + const footerStyle = consolePage.match(/\.composer-input-footer\s*\{([^}]+)\}/)?.[1]; + assert( + footerStyle?.includes("display: grid") && + !/position:\s*(absolute|fixed)/.test(footerStyle) && + !footerStyle.includes("pointer-events: none"), + "Composer toolbar must stay in normal flow below the scrollable editor", + ); + assert( + composerInput.includes("min-height: 2.65rem") && + composerInput.includes("padding: 0.55rem 0.65rem 0.35rem") && + !composerInput.includes("padding: 0.55rem 3.4rem 3rem 0.65rem"), + "Editor must not reserve an overlapping toolbar inside its scrolling content", + ); + const attachButton = consolePage.match( + //, + )?.[0]; + assert( + attachButton?.includes('aria-label="Attach file"') && + attachButton.includes('title="Attach file"') && + attachButton.includes('class="composer-attach-icon"') && + attachButton.includes('aria-hidden="true"') && + attachButton.includes('d="M12 5V19M5 12H19"') && + attachButton.includes("fileInput?.click()") && + !attachButton.includes(">Attach file"), + "Attach file must remain an accessible file-picker button with a plus icon", + ); +}); + +Deno.test("Worker Console keeps Notify beside Submit without persistent notices", async () => { + const consolePage = await Deno.readTextFile( + new URL( + "./../../../routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte", + import.meta.url, + ), + ); + const actions = consolePage.match( + /
([\s\S]*?)<\/div>/, + )?.[1]; + assert( + actions?.includes("{#if workerRunning}") && + actions.includes('class="composer-notify-button"') && + actions.includes('type="button"') && + actions.includes('aria-label="Notify Worker"') && + actions.includes('aria-hidden="true"') && + actions.includes('class="composer-notify-icon"') && + actions.includes("disabled={!canNotifyDraft}") && + actions.includes("onclick={handleNotifySubmit}") && + actions.includes('class="composer-send-button"') && + actions.indexOf('class="composer-notify-button"') < + actions.indexOf('class="composer-send-button"'), + "Running Worker Notify must be an accessible bell button beside Submit/Stop", + ); + assert( + !consolePage.includes("composerNotice") && + !consolePage.includes("composer-notice") && + !consolePage.includes('class="composer-actions"') && + !consolePage.includes("Sent through Worker protocol.") && + consolePage.includes('{#if sendError}
@@ -2265,9 +2279,19 @@ + > + + {#if completionBusy || completionError || completionEntries.length > 0}
{#if completionBusy} @@ -2286,51 +2310,78 @@
{/if}
- {/if} - + {#if workerRunning} + + {/if} + +
-
- {#if workerRunning} - - {/if} - {#if composerNotice} - {composerNotice} - {/if} - {#if sendError}

{sendError}

{/if} -
+ {#if sendError}{/if} @@ -2642,7 +2693,7 @@ display: flex; flex-wrap: wrap; gap: var(--space-2); - padding: 0 var(--space-3) 3rem; + padding: 0 var(--space-3) var(--space-2); } .composer-attachment { @@ -2672,7 +2723,9 @@ } .composer-attachment button, - .composer-attach-button { + .composer-attach-button, + .composer-queue-button, + .composer-notify-button { border: 0; background: transparent; color: var(--text-muted); @@ -2681,24 +2734,45 @@ font: inherit; } - .composer-attach-button { - padding: 0.35rem 0; + .composer-attach-button, + .composer-queue-button, + .composer-notify-button { + display: inline-grid; + flex: 0 0 auto; + width: 2.35rem; + height: 2.35rem; + place-items: center; + border-radius: 999px; + padding: 0; + } + + .composer-attach-button:hover:not(:disabled), + .composer-queue-button:hover:not(:disabled), + .composer-notify-button:hover:not(:disabled) { + background: var(--bg-subtle); + color: var(--text-strong); + } + + .composer-attach-button:disabled, + .composer-queue-button:disabled, + .composer-notify-button:disabled { + cursor: not-allowed; + opacity: 0.55; } .composer-input-footer { - position: absolute; - right: 0.7rem; - bottom: 0.7rem; - left: 1rem; display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: var(--space-2); align-items: end; min-height: 2.35rem; - pointer-events: none; + padding: 0 0.35rem 0.35rem 0.65rem; } .composer-footer-slot { + display: flex; + align-items: center; + gap: var(--space-2); min-width: 0; } @@ -2726,6 +2800,9 @@ opacity: 0.55; } + .composer-attach-icon, + .composer-queue-icon, + .composer-notify-icon, .composer-send-icon { width: 1.2rem; height: 1.2rem; @@ -2745,21 +2822,10 @@ font-size: var(--font-size-compact); } - .composer-notice { - color: var(--text-muted); - font-size: var(--font-size-compact); - } - - .composer-actions { + .composer-submit-actions { display: flex; - flex-wrap: wrap; align-items: center; - justify-content: flex-end; - gap: 0.7rem; - } - - .composer-actions .composer-notice { - margin-right: auto; + gap: var(--space-2); } @media (max-width: 960px) { diff --git a/web/workspace/tests/composer-paste.test.ts b/web/workspace/tests/composer-paste.test.ts index b59850ce3..58e03794c 100644 --- a/web/workspace/tests/composer-paste.test.ts +++ b/web/workspace/tests/composer-paste.test.ts @@ -3,11 +3,14 @@ import fixtureJson from "../../../tests/fixtures/composer-paste-policy.json" wit }; import { + type ComposerPasteEvent, type ComposerPasteMeasurement, + handleComposerImagePaste, handleComposerPaste, MAX_PLAIN_TEXT_PASTE_CHARS, MAX_PLAIN_TEXT_PASTE_LOGICAL_LINES, measureComposerPaste, + namePastedImage, } from "../src/lib/workspace/console/composer-paste.ts"; declare const Deno: { @@ -30,6 +33,142 @@ function assertEquals( } } +function imagePasteEvent(clipboardData: ComposerPasteEvent["clipboardData"]) { + let prevented = 0; + return { + event: { + clipboardData, + preventDefault: () => { + prevented += 1; + }, + }, + prevented: () => prevented, + }; +} + +Deno.test("image paste attaches original files once without reading alternate text", () => { + const image = new File(["png"], "image.png", { type: "image/png" }); + const { event, prevented } = imagePasteEvent({ + getData: () => { + throw new Error("image must take priority over text"); + }, + files: [image], + items: [{ kind: "file", type: image.type, getAsFile: () => image }], + }); + const batches: File[][] = []; + assertEquals( + handleComposerImagePaste(event, (files) => batches.push(files)), + true, + ); + assertEquals(prevented(), 1); + assertEquals(batches.length, 1); + assertEquals(batches[0].length, 1); + assertEquals(batches[0][0] === image, true); +}); + +Deno.test("image paste falls back to file items and ignores text, other files and null entries", () => { + const image = new File(["jpeg"], "photo.jpg", { type: "image/jpeg" }); + const { event, prevented } = imagePasteEvent({ + getData: () => "alternate image title", + files: [], + items: [ + { kind: "string", type: "text/plain", getAsFile: () => null }, + { + kind: "file", + type: "application/pdf", + getAsFile: () => new File([], "a.pdf"), + }, + { kind: "file", type: "image/png", getAsFile: () => null }, + { kind: "file", type: image.type, getAsFile: () => image }, + ], + }); + let files: File[] = []; + assertEquals( + handleComposerImagePaste(event, (value) => { + files = value; + }), + true, + ); + assertEquals(files.length, 1); + assertEquals(files[0] === image, true); + assertEquals(prevented(), 1); +}); + +Deno.test("multiple pasted images retain order and use attachment validation for unsupported types", () => { + const images = ["image/png", "image/webp", "image/bmp"].map((type) => + new File([type], "image", { type }) + ); + const { event } = imagePasteEvent({ + getData: () => "", + files: [ + images[0], + new File([], "note.txt", { type: "text/plain" }), + ...images.slice(1), + ], + }); + let attached: File[] = []; + assertEquals( + handleComposerImagePaste(event, (files) => { + attached = files; + }), + true, + ); + assertEquals(attached.length, images.length); + assertEquals(attached.every((file, index) => file === images[index]), true); +}); + +Deno.test("non-image or absent clipboard leaves text paste and its exact content untouched", () => { + for ( + const clipboardData of [null, { getData: () => " text\r\n" }, { + getData: () => "file name", + files: [new File([], "file.pdf", { type: "application/pdf" })], + }] + ) { + const { event, prevented } = imagePasteEvent(clipboardData); + assertEquals( + handleComposerImagePaste(event, () => { + throw new Error("unexpected attachment"); + }), + false, + ); + assertEquals(prevented(), 0); + } +}); + +Deno.test("image paste does not attach when disabled or no callback is installed", () => { + const { event, prevented } = imagePasteEvent({ + getData: () => "", + files: [new File(["png"], "image.png", { type: "image/png" })], + }); + assertEquals(handleComposerImagePaste(event, undefined), false); + assertEquals( + handleComposerImagePaste(event, () => { + throw new Error("disabled attachment"); + }, false), + false, + ); + assertEquals(prevented(), 0); +}); + +Deno.test("pasted screenshot names are unique without changing bytes or media type", async () => { + const file = new File([new Uint8Array([137, 80, 78, 71])], "image.png", { + type: "image/png", + lastModified: 1234, + }); + const first = namePastedImage(file); + const second = namePastedImage(file); + assertEquals(first.name === second.name, false); + assertEquals(first.name.endsWith("-image.png"), true); + assertEquals(first.type, file.type); + assertEquals(first.lastModified, file.lastModified); + assertEquals(Array.from(new Uint8Array(await first.arrayBuffer())), [ + 137, + 80, + 78, + 71, + ]); +}); + interface FixturePart { value: string; repeat: number;