From 2006b390b420abec945952645c8bbc6347ccb7a0 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 05:16:57 +0900 Subject: [PATCH] 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 | 16 +- crates/session-store/src/worker_metadata.rs | 106 +++++ .../session-store/src/worker_session_store.rs | 446 +++++++++++++++++- crates/tui/src/app.rs | 42 +- crates/tui/src/console/mod.rs | 26 +- crates/tui/src/ui.rs | 2 +- crates/worker-runtime/src/worker_backend.rs | 16 +- crates/worker/src/controller.rs | 235 ++++----- .../src/feature/builtin/manage_worker.rs | 2 +- crates/worker/src/prompt/catalog.rs | 3 +- crates/worker/src/worker.rs | 55 +-- crates/worker/tests/controller_test.rs | 249 +++++++++- .../prompts/common/worker-orchestration.md | 2 +- .../workspace/console/ComposerInput.svelte | 10 +- .../console/composer-delivery.test.ts | 71 +++ .../workspace/console/composer-delivery.ts | 8 +- .../lib/workspace/console/composer-paste.ts | 50 +- .../console/worker-console.ui.test.ts | 119 ++++- .../workers/[workerRef]/console/+page.svelte | 220 ++++++--- web/workspace/tests/composer-paste.test.ts | 139 ++++++ 20 files changed, 1478 insertions(+), 339 deletions(-) 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;