From 5b0a6691f8df2a32119b7f23299f6f43fbc8eee1 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 5 Sep 2026 23:16:19 +0900 Subject: [PATCH] fix: fence pending queue controls --- crates/protocol/src/lib.rs | 12 ++- crates/tui/src/app.rs | 43 ++++++++++ crates/tui/src/console/mod.rs | 70 ++++++++++++++- crates/tui/src/ui.rs | 5 +- crates/worker/src/controller.rs | 86 ++++++++++++------- crates/worker/src/worker.rs | 72 ++++++++++++++-- crates/worker/tests/controller_test.rs | 31 ++++++- web/workspace/src/lib/generated/protocol.ts | 4 +- .../src/lib/workspace/console/model.test.ts | 7 +- .../console/worker-console.ui.test.ts | 13 +++ .../workers/[workerId]/console/+page.svelte | 72 ++++++++++++---- 11 files changed, 347 insertions(+), 68 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 7171461c..e3330ba0 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -73,12 +73,18 @@ pub enum Method { /// are immutable and therefore cannot be cancelled here. CancelPendingSubmission { submission_id: String, + expected_revision: u64, }, /// Remove every queued submission while preserving the active run. - ClearPendingSubmissions, + ClearPendingSubmissions { + expected_revision: u64, + }, /// Activate the next queued submission while the Worker is idle. This is an /// explicit recovery operation and never resumes a paused run implicitly. - ContinuePending, + ContinuePending { + expected_revision: u64, + expected_head_id: String, + }, Resume, Cancel, /// Stop the in-flight turn and transition to `Paused`. @@ -555,6 +561,8 @@ pub struct PendingSubmissionsSnapshot { #[serde(default)] pub notification_count: u32, #[serde(default)] + pub head_id: Option, + #[serde(default)] pub submissions: Vec, } diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 2fecd888..d23430c2 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -754,6 +754,28 @@ impl App { Some(self.method_for_run(segments)) } + pub fn submit_notify_input(&mut self) -> Option { + let segments = self.input.submit_segments(); + if segments_are_blank(&segments) { + return None; + } + if segments + .iter() + .any(|segment| matches!(segment, Segment::UploadedFile { .. })) + { + self.push_error("Notify accepts text only; remove attachments or queue a Submit."); + return None; + } + let message = Segment::flatten_to_text(&segments); + self.record_input_history(segments); + self.input.clear(); + Some(Method::Notify { + notification_request_id: protocol::new_submission_request_id(), + message, + auto_run: true, + }) + } + pub fn restore_unsent_run(&mut self, method: &Method) { let Method::Submit { input, .. } = method else { return; @@ -890,6 +912,26 @@ impl App { } } + pub fn continue_pending_method(&self) -> Option { + Some(Method::ContinuePending { + expected_revision: self.pending_submissions.revision, + expected_head_id: self.pending_submissions.head_id.clone()?, + }) + } + + pub fn clear_pending_method(&self) -> Method { + Method::ClearPendingSubmissions { + expected_revision: self.pending_submissions.revision, + } + } + + pub fn cancel_pending_method(&self, submission_id: String) -> Method { + Method::CancelPendingSubmission { + submission_id, + expected_revision: self.pending_submissions.revision, + } + } + pub fn next_queued_input_preview(&self) -> Option<&str> { self.pending_submissions .submissions @@ -3410,6 +3452,7 @@ mod completion_flow_tests { pending: protocol::PendingSubmissionsSnapshot { revision: 3, notification_count: 0, + head_id: Some("submission-1".into()), submissions: vec![protocol::PendingSubmissionSummary { submission_id: "submission-1".into(), accepted_at_ms: 7, diff --git a/crates/tui/src/console/mod.rs b/crates/tui/src/console/mod.rs index 3f0d6ddd..a7ff703e 100644 --- a/crates/tui/src/console/mod.rs +++ b/crates/tui/src/console/mod.rs @@ -1150,13 +1150,27 @@ fn handle_key(app: &mut App, key: KeyEvent) -> Option { app.clear_command_input(); Some(None) } + KeyCode::Char(c) + if c.eq_ignore_ascii_case(&'d') && alt && !ctrl && !app.is_command_mode() => + { + Some( + app.next_queued_input_preview() + .map(str::to_owned) + .map(|submission_id| app.cancel_pending_method(submission_id)), + ) + } + KeyCode::Char(c) + if c.eq_ignore_ascii_case(&'n') && alt && !ctrl && !app.is_command_mode() => + { + Some(app.submit_notify_input()) + } KeyCode::Char(c) if c.eq_ignore_ascii_case(&'q') && alt && !ctrl && !app.is_command_mode() => { - Some(Some(Method::ContinuePending)) + Some(app.continue_pending_method()) } KeyCode::Char(c) if c.eq_ignore_ascii_case(&'c') && alt && !ctrl => { - Some(Some(Method::ClearPendingSubmissions)) + Some(Some(app.clear_pending_method())) } KeyCode::Char('c') if ctrl => Some(handle_pause_or_quit(app)), KeyCode::Char('x') if ctrl => Some(handle_cancel_or_shutdown(app)), @@ -1976,6 +1990,29 @@ mod tests { assert_eq!(input_text(&app), ""); } + #[test] + fn running_alt_n_sends_explicit_notify_without_implicit_submit_conversion() { + let mut app = App::new("test".into()); + app.set_worker_status(WorkerStatus::Running); + for character in "progress".chars() { + app.insert_char(character); + } + + let method = handle_key( + &mut app, + KeyEvent::new(KeyCode::Char('n'), KeyModifiers::ALT), + ); + assert!(matches!( + method, + Some(Method::Notify { + ref message, + auto_run: true, + .. + }) if message == "progress" + )); + assert_eq!(input_text(&app), ""); + } + #[test] fn pending_queue_shortcuts_send_worker_operations() { let mut app = App::new("test".into()); @@ -1983,6 +2020,7 @@ mod tests { pending: protocol::PendingSubmissionsSnapshot { revision: 2, notification_count: 0, + head_id: Some("submission-1".into()), submissions: vec![protocol::PendingSubmissionSummary { submission_id: "submission-1".into(), accepted_at_ms: 1, @@ -1996,14 +2034,37 @@ mod tests { &mut app, KeyEvent::new(KeyCode::Char('q'), KeyModifiers::ALT), ); - assert!(matches!(continue_next, Some(Method::ContinuePending))); + assert!(matches!( + continue_next, + Some(Method::ContinuePending { + expected_revision: 2, + ref expected_head_id, + }) if expected_head_id == "submission-1" + )); assert_eq!(app.queued_input_count(), 1); + let cancel = handle_key( + &mut app, + KeyEvent::new(KeyCode::Char('d'), KeyModifiers::ALT), + ); + assert!(matches!( + cancel, + Some(Method::CancelPendingSubmission { + expected_revision: 2, + ref submission_id, + }) if submission_id == "submission-1" + )); + let clear = handle_key( &mut app, KeyEvent::new(KeyCode::Char('c'), KeyModifiers::ALT), ); - assert!(matches!(clear, Some(Method::ClearPendingSubmissions))); + assert!(matches!( + clear, + Some(Method::ClearPendingSubmissions { + expected_revision: 2 + }) + )); assert_eq!(app.queued_input_count(), 1); } @@ -2014,6 +2075,7 @@ mod tests { pending: protocol::PendingSubmissionsSnapshot { revision: 2, notification_count: 0, + head_id: Some("submission-1".into()), submissions: vec![protocol::PendingSubmissionSummary { submission_id: "submission-1".into(), accepted_at_ms: 1, diff --git a/crates/tui/src/ui.rs b/crates/tui/src/ui.rs index 577fafe9..58b790f1 100644 --- a/crates/tui/src/ui.rs +++ b/crates/tui/src/ui.rs @@ -1880,7 +1880,7 @@ fn actionbar_left_item(app: &App, now: Instant) -> Option<(String, Style)> { } if app.queued_input_count() > 0 { return Some(( - "Alt-q continue queued Alt-c clear queued".to_string(), + "Alt-n notify Alt-q continue Alt-d cancel queued Alt-c clear queued".to_string(), Style::default().fg(Color::DarkGray), )); } @@ -2144,6 +2144,7 @@ mod tests { pending: protocol::PendingSubmissionsSnapshot { revision: 1, notification_count: 0, + head_id: Some(id.into()), submissions: vec![protocol::PendingSubmissionSummary { submission_id: id.into(), accepted_at_ms: 1, @@ -2303,7 +2304,7 @@ mod tests { set_pending_submission(&mut app, "submission-1"); assert_eq!( actionbar_left_item(&app, now).map(|(text, _)| text), - Some("Alt-q continue queued Alt-c clear queued".into()) + Some("Alt-n notify Alt-q continue Alt-d cancel queued Alt-c clear queued".into()) ); app.enter_command_mode(); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 2794b35c..4834abd0 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -232,8 +232,9 @@ enum PendingRun { fn prepare_pending_run( pending_submissions: &crate::worker::PendingSubmissionHandle, notify_buffer: &NotifyBuffer, + fence: Option<(u64, &str)>, ) -> Result, crate::worker::PendingSubmissionError> { - Ok(match pending_submissions.prepare_next_activation()? { + Ok(match pending_submissions.prepare_next_activation(fence)? { Some(crate::worker::PendingActivation::Submission(submission)) => { Some(PendingRun::Submit(submission)) } @@ -1420,7 +1421,7 @@ async fn controller_loop( } if !shutdown && may_drain_pending && new_status == WorkerStatus::Idle { - match prepare_pending_run(&pending_submissions, ¬ify_buffer) { + match prepare_pending_run(&pending_submissions, ¬ify_buffer, None) { Ok(Some(next)) => { pending = Some(next); new_status = WorkerStatus::Running; @@ -1505,17 +1506,18 @@ async fn controller_loop( if auto_run { match pending_submissions.accept_notification(notification_request_id, message) { - Ok(true) => match prepare_pending_run(&pending_submissions, ¬ify_buffer) - { - Ok(Some(next)) => pending = Some(next), - Ok(None) => {} - Err(error) => { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::Internal, - message: error.to_string(), - }); + Ok(true) => { + match prepare_pending_run(&pending_submissions, ¬ify_buffer, None) { + Ok(Some(next)) => pending = Some(next), + Ok(None) => {} + Err(error) => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::Internal, + message: error.to_string(), + }); + } } - }, + } Ok(false) => {} Err(error) => { let _ = working_event_tx.send(Event::Error { @@ -1534,8 +1536,24 @@ async fn controller_loop( pending: pending_submissions.snapshot(), }); } - Method::CancelPendingSubmission { submission_id } => { - match pending_submissions.cancel(&submission_id) { + Method::CancelPendingSubmission { + submission_id, + expected_revision, + } => match pending_submissions.cancel(&submission_id, expected_revision) { + Ok(pending_snapshot) => { + let _ = working_event_tx.send(Event::PendingSubmissionsChanged { + pending: pending_snapshot, + }); + } + Err(error) => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: error.to_string(), + }); + } + }, + Method::ClearPendingSubmissions { expected_revision } => { + match pending_submissions.clear(expected_revision) { Ok(pending_snapshot) => { let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending: pending_snapshot, @@ -1549,21 +1567,22 @@ async fn controller_loop( } } } - Method::ClearPendingSubmissions => match pending_submissions.clear() { - Ok(pending_snapshot) => { - let _ = working_event_tx.send(Event::PendingSubmissionsChanged { - pending: pending_snapshot, - }); - } - Err(error) => { + Method::ContinuePending { + expected_revision, + expected_head_id, + } => { + if shared_state.get_status() != WorkerStatus::Idle { let _ = working_event_tx.send(Event::Error { - code: ErrorCode::Internal, - message: error.to_string(), + code: ErrorCode::InvalidRequest, + message: "ContinuePending requires an idle Worker; Resume or Cancel a paused run first".into(), }); + continue; } - }, - Method::ContinuePending => { - match prepare_pending_run(&pending_submissions, ¬ify_buffer) { + match prepare_pending_run( + &pending_submissions, + ¬ify_buffer, + Some((expected_revision, &expected_head_id)), + ) { Ok(Some(next)) => pending = Some(next), Ok(None) => { let _ = working_event_tx.send(Event::Error { @@ -2077,7 +2096,7 @@ where } } } - Some(Method::Resume | Method::ContinuePending) => { + Some(Method::Resume | Method::ContinuePending { .. }) => { let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn".into(), @@ -2088,8 +2107,11 @@ where pending: pending_submissions.snapshot(), }); } - Some(Method::CancelPendingSubmission { submission_id }) => { - match pending_submissions.cancel(&submission_id) { + Some(Method::CancelPendingSubmission { + submission_id, + expected_revision, + }) => { + match pending_submissions.cancel(&submission_id, expected_revision) { Ok(pending) => { let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending }); } @@ -2101,14 +2123,14 @@ where } } } - Some(Method::ClearPendingSubmissions) => { - match pending_submissions.clear() { + Some(Method::ClearPendingSubmissions { expected_revision }) => { + match pending_submissions.clear(expected_revision) { Ok(pending) => { let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending }); } Err(error) => { let _ = working_event_tx.send(Event::Error { - code: ErrorCode::Internal, + code: ErrorCode::InvalidRequest, message: error.to_string(), }); } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index cc93bb3a..3bfb9230 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -130,9 +130,20 @@ pub(crate) struct PendingActivationState { impl PendingActivationState { pub(crate) fn snapshot(&self) -> protocol::PendingSubmissionsSnapshot { + let head_id = match (self.pending.front(), self.pending_notifications.front()) { + (Some(submission), Some(notification)) + if notification.activation_sequence < submission.activation_sequence => + { + Some(notification.notification_request_id.clone()) + } + (Some(submission), _) => Some(submission.submission_id.clone()), + (None, Some(notification)) => Some(notification.notification_request_id.clone()), + (None, None) => None, + }; protocol::PendingSubmissionsSnapshot { revision: self.revision, notification_count: u32::try_from(self.pending_notifications.len()).unwrap_or(u32::MAX), + head_id, submissions: self .pending .iter() @@ -1125,6 +1136,13 @@ pub(crate) enum PendingSubmissionError { ByteLimit, #[error("pending submission artifact references exceed {MAX_PENDING_ARTIFACT_REFS}")] ArtifactLimit, + #[error("pending queue revision conflict: expected {expected}, current {current}")] + RevisionConflict { expected: u64, current: u64 }, + #[error("pending queue head conflict: expected {expected}, current {current:?}")] + HeadConflict { + expected: String, + current: Option, + }, #[error("pending submission not found: {0}")] NotFound(String), #[error("pending submission state persistence failed: {0}")] @@ -1141,6 +1159,29 @@ impl PendingSubmissionHandle where St: Store + Clone, { + fn validate_fence( + state: &PendingActivationState, + expected_revision: u64, + expected_head_id: Option<&str>, + ) -> Result<(), PendingSubmissionError> { + if state.revision != expected_revision { + return Err(PendingSubmissionError::RevisionConflict { + expected: expected_revision, + current: state.revision, + }); + } + if let Some(expected) = expected_head_id { + let current = state.snapshot().head_id; + if current.as_deref() != Some(expected) { + return Err(PendingSubmissionError::HeadConflict { + expected: expected.to_owned(), + current, + }); + } + } + Ok(()) + } + fn persist_locked(&self, state: &PendingActivationState) -> Result<(), PendingSubmissionError> { self.writer.append_entry_locked(LogEntry::Extension { ts: segment_log::now_millis(), @@ -1354,6 +1395,7 @@ where pub(crate) fn prepare_next_activation( &self, + fence: Option<(u64, &str)>, ) -> Result, PendingSubmissionError> { let _append_guard = self .writer @@ -1365,6 +1407,9 @@ where .state .lock() .expect("pending activation state poisoned"); + if let Some((expected_revision, expected_head_id)) = fence { + Self::validate_fence(&state, expected_revision, Some(expected_head_id))?; + } if state.activating.is_some() || state.activating_notification.is_some() { return Ok(None); } @@ -1495,6 +1540,7 @@ where pub(crate) fn cancel( &self, submission_id: &str, + expected_revision: u64, ) -> Result { let _append_guard = self .writer @@ -1506,6 +1552,7 @@ where .state .lock() .expect("pending activation state poisoned"); + Self::validate_fence(&state, expected_revision, None)?; let original = state.clone(); let Some(index) = state .pending @@ -1525,6 +1572,7 @@ where pub(crate) fn clear( &self, + expected_revision: u64, ) -> Result { let _append_guard = self .writer @@ -1536,6 +1584,7 @@ where .state .lock() .expect("pending activation state poisoned"); + Self::validate_fence(&state, expected_revision, None)?; let original = state.clone(); state.pending.clear(); state.pending_notifications.clear(); @@ -9189,10 +9238,23 @@ mod build_summary_prompt_tests { assert_eq!(restored.pending.len(), 1); assert_eq!(restored.pending[0].submission_id, accepted.submission_id); - let snapshot = handle.cancel(&accepted.submission_id).unwrap(); + let fence = handle.snapshot(); + assert!(matches!( + handle.cancel(&accepted.submission_id, fence.revision.saturating_sub(1)), + Err(PendingSubmissionError::RevisionConflict { .. }) + )); + assert!(matches!( + handle.prepare_next_activation(Some((fence.revision, "wrong-head"))), + Err(PendingSubmissionError::HeadConflict { .. }) + )); + assert_eq!(handle.snapshot(), fence); + + let snapshot = handle + .cancel(&accepted.submission_id, handle.snapshot().revision) + .unwrap(); assert!(snapshot.submissions.is_empty()); assert!(matches!( - handle.cancel(&accepted.submission_id), + handle.cancel(&accepted.submission_id, handle.snapshot().revision), Err(PendingSubmissionError::NotFound(_)) )); @@ -9210,7 +9272,7 @@ mod build_summary_prompt_tests { Err(PendingSubmissionError::CountLimit) )); assert_eq!(handle.snapshot().submissions.len(), MAX_PENDING_SUBMISSIONS); - let cleared = handle.clear().unwrap(); + let cleared = handle.clear(handle.snapshot().revision).unwrap(); assert!(cleared.submissions.is_empty()); assert_eq!(cleared.notification_count, 0); } @@ -9237,7 +9299,7 @@ mod build_summary_prompt_tests { .accept("request-1".into(), vec![Segment::text("submit")], false) .unwrap(); - let first = handle.prepare_next_activation().unwrap().unwrap(); + let first = handle.prepare_next_activation(None).unwrap().unwrap(); assert!(matches!( first, PendingActivation::Notification(PendingNotification { ref message, .. }) @@ -9249,7 +9311,7 @@ mod build_summary_prompt_tests { assert!(committed_state.pending_notifications.is_empty()); assert!(committed_state.activating_notification.is_none()); handle.finish_notification_activation("notification-1"); - let second = handle.prepare_next_activation().unwrap().unwrap(); + let second = handle.prepare_next_activation(None).unwrap().unwrap(); assert!(matches!(second, PendingActivation::Submission(_))); } diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 6faa2fdd..28fc96e2 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -1322,7 +1322,7 @@ async fn submit_while_running_is_durably_queued() { let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2); let mut accepted = None; - let mut pending_count = None; + let mut pending_snapshot = None; while tokio::time::Instant::now() < deadline { match tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv()).await { Ok(Ok(Event::SubmissionAccepted { @@ -1333,21 +1333,44 @@ async fn submit_while_running_is_durably_queued() { Ok(Ok(Event::PendingSubmissionsChanged { pending })) if pending.submissions.len() == 1 => { - pending_count = Some(1) + pending_snapshot = Some(pending) } Ok(Ok(Event::Error { code, message })) if code == worker::ErrorCode::AlreadyRunning => { panic!("Submit was busy-rejected: {message}") } _ => {} } - if accepted.is_some() && pending_count.is_some() { + if accepted.is_some() && pending_snapshot.is_some() { break; } } assert_eq!(accepted, Some(protocol::SubmissionDisposition::Queued)); - assert_eq!(pending_count, Some(1)); + let pending_snapshot = pending_snapshot.expect("pending snapshot"); + assert_eq!(pending_snapshot.submissions.len(), 1); handle.send(Method::Pause).await.unwrap(); + wait_for_status(&handle, WorkerStatus::Paused).await; + handle + .send(Method::ContinuePending { + expected_revision: pending_snapshot.revision, + expected_head_id: pending_snapshot.head_id.expect("pending head"), + }) + .await + .unwrap(); + let rejection = tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if let Ok(Event::Error { code, message }) = rx.recv().await + && code == worker::ErrorCode::InvalidRequest + && message.contains("requires an idle Worker") + { + break message; + } + } + }) + .await + .expect("paused ContinuePending rejection"); + assert!(rejection.contains("Resume or Cancel")); + assert_eq!(handle.shared_state.get_status(), WorkerStatus::Paused); } #[tokio::test] diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 1e744d20..6055d3e7 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -105,7 +105,7 @@ timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array, }; +export type PendingSubmissionsSnapshot = { revision: number, notification_count: number, head_id: string | null, submissions: Array, }; export type SubmissionDisposition = "started" | "queued"; @@ -231,7 +231,7 @@ export type SubscriptionFramePayload = { "frame": "request", "message": Subscrip export type SubscriptionFrame = { protocol_version: number, } & ({ "frame": "request", "message": SubscriptionRequest } | { "frame": "response", "message": SubscriptionResponse } | { "frame": "event", "message": SubscriptionEvent } | { "frame": "worker_protocol", "message": SubscriptionWorkerProtocolMethod }); -export type Method = { "method": "submit", "params": { submission_request_id: string, input: Array, } } | { "method": "notify", "params": { notification_request_id: string, message: string, auto_run?: boolean, } } | { "method": "worker_event", "params": WorkerEvent } | { "method": "list_pending_submissions" } | { "method": "cancel_pending_submission", "params": { submission_id: string, } } | { "method": "clear_pending_submissions" } | { "method": "continue_pending" } | { "method": "resume" } | { "method": "cancel" } | { "method": "pause" } | { "method": "compact" } | { "method": "list_rewind_targets" } | { "method": "rewind_to", "params": { target: RewindTargetId, expected_head_entries: number, } } | { "method": "shutdown" } | { "method": "list_completions", "params": { kind: CompletionKind, prefix: string, } } | { "method": "list_workers" } | { "method": "restore_worker", "params": { name: string, } } | { "method": "register_peer", "params": { name: string, } }; +export type Method = { "method": "submit", "params": { submission_request_id: string, input: Array, } } | { "method": "notify", "params": { notification_request_id: string, message: string, auto_run?: boolean, } } | { "method": "worker_event", "params": WorkerEvent } | { "method": "list_pending_submissions" } | { "method": "cancel_pending_submission", "params": { submission_id: string, expected_revision: number, } } | { "method": "clear_pending_submissions", "params": { expected_revision: number, } } | { "method": "continue_pending", "params": { expected_revision: number, expected_head_id: string, } } | { "method": "resume" } | { "method": "cancel" } | { "method": "pause" } | { "method": "compact" } | { "method": "list_rewind_targets" } | { "method": "rewind_to", "params": { target: RewindTargetId, expected_head_entries: number, } } | { "method": "shutdown" } | { "method": "list_completions", "params": { kind: CompletionKind, prefix: string, } } | { "method": "list_workers" } | { "method": "restore_worker", "params": { name: string, } } | { "method": "register_peer", "params": { name: string, } }; export type Event = { "event": "submission_accepted", "data": { submission_request_id: string, submission_id: string, disposition: SubmissionDisposition, } } | { "event": "submission_rejected", "data": { submission_request_id: string, message: string, } } | { "event": "pending_submissions_changed", "data": { pending: PendingSubmissionsSnapshot, } } | { "event": "user_message", "data": { segments: Array, } } | { "event": "system_item", "data": { item: unknown, } } | { "event": "invoke_start", "data": { kind: InvokeKind, } } | { "event": "turn_start", "data": { turn: number, } } | { "event": "turn_end", "data": { turn: number, result: TurnResult, } } | { "event": "llm_call_start", "data": { llm_call: number, } } | { "event": "llm_call_end", "data": { llm_call: number, } } | { "event": "llm_retry", "data": { llm_call: number, /** diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index da4f70d5..60b98633 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -2150,7 +2150,12 @@ Deno.test("snapshot restores TaskStore state from system history", () => { const event = snapshotEvent("/repo"); if (event.event !== "snapshot") throw new Error("snapshot fixture expected"); event.data.session = { - pending_submissions: { revision: 0, notification_count: 0, submissions: [] }, + pending_submissions: { + revision: 0, + notification_count: 0, + head_id: null, + submissions: [], + }, entries: [{ entry_id: "task-reminder-1", timestamp: 1, 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 b031f1d0..92d9531e 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 @@ -1077,7 +1077,20 @@ Deno.test("Web Console uses Notify while running and exposes durable pending con 'method: "cancel_pending_submission"', 'method: "clear_pending_submissions"', 'method: "continue_pending"', + "handleQueueSubmit", + "handleNotifySubmit", + ">Queue Submit", + ">Notify", ]) { assert(consolePage.includes(token), `missing durable pending control token: ${token}`); } + + const userCase = consolePage.slice( + consolePage.indexOf('case "user":'), + consolePage.indexOf('case "compact":'), + ); + assert( + !userCase.includes("workerRunning"), + "ordinary text must remain Submit instead of being implicitly converted to Notify", + ); }); diff --git a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte index 6c50e2ae..60bf188d 100644 --- a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte @@ -161,6 +161,7 @@ let pendingSubmissions = $state({ revision: 0, notification_count: 0, + head_id: null, submissions: [], }); let pendingSubmissionItems = $derived(pendingSubmissions.submissions ?? []); @@ -574,16 +575,6 @@ ): ProtocolMethod { switch (request.kind) { case "user": - if (workerRunning) { - return { - method: "notify", - params: { - notification_request_id: crypto.randomUUID(), - message: request.content, - auto_run: true, - }, - }; - } return { method: "submit", params: { @@ -672,6 +663,14 @@ void submitDraft(composerInputElement?.snapshot() ?? draft); } + function handleQueueSubmit() { + void submitDraft(composerInputElement?.snapshot() ?? draft); + } + + function handleNotifySubmit() { + void submitDraft(composerInputElement?.snapshot() ?? draft, "notify"); + } + function attachmentPath(): string { return `/api/w/${encodeURIComponent(workspaceId)}/runtimes/${encodeURIComponent(runtimeId)}/workers/${encodeURIComponent(workerId)}`; } @@ -771,7 +770,15 @@ if (event.dataTransfer?.files) addAttachmentFiles(event.dataTransfer.files); } - async function submitDraft(value: ComposerDraftSnapshot) { + async function submitDraft( + value: ComposerDraftSnapshot, + delivery: "submit" | "notify" = "submit", + ) { + if (delivery === "notify" && attachments.length > 0) { + composerNotice = null; + sendError = "Notify accepts text only; remove attachments or queue a Submit."; + return; + } const incompleteAttachment = attachments.find((attachment) => attachment.state !== "uploaded" || !attachment.reference ); @@ -805,10 +812,19 @@ return; } + let request: WorkerConsoleInputRequest = command.request; + if (delivery === "notify") { + if (request.kind !== "user") { + composerNotice = null; + sendError = "Notify accepts ordinary text, not a Composer command."; + return; + } + request = { kind: "notify", content: request.content }; + } sending = true; sendError = null; try { - const method = composerRequestToProtocolMethod(command.request); + const method = composerRequestToProtocolMethod(request); sendProtocolMethod(method); composerInputElement?.recordHistory(value); composerInputElement?.clear(); @@ -1772,7 +1788,10 @@ sendControl( { method: "cancel_pending_submission", - params: { submission_id: submission.submission_id }, + params: { + submission_id: submission.submission_id, + expected_revision: pendingSubmissions.revision, + }, }, "Pending submission cancellation", )} @@ -1782,10 +1801,16 @@ @@ -1793,7 +1818,10 @@ type="button" onclick={() => sendControl( - { method: "clear_pending_submissions" }, + { + method: "clear_pending_submissions", + params: { expected_revision: pendingSubmissions.revision }, + }, "Pending submissions clear", )} >Clear all @@ -1932,6 +1960,18 @@
+ {#if workerRunning} + + + {/if} {#if composerNotice} {composerNotice} {/if}