4 Commits
32 changed files with 1201 additions and 657 deletions
+21 -39
View File
@@ -349,13 +349,11 @@ pub enum Method {
Compact {
command: WorkerCommandEnvelope,
},
/// Ask the Worker to list valid rewind targets from its authoritative session log.
ListRewindTargets,
/// Truncate the current session back to the selected rewind target and
/// return the selected user input to the client composer.
/// Rewind to just before a user input identified by its public Session entry ID.
/// Validate the target against the current segment at execution time.
/// Later appended history is discarded too; tool side effects are not undone.
RewindTo {
target: RewindTargetId,
expected_head_entries: usize,
entry_id: String,
},
Shutdown {
command: WorkerCommandEnvelope,
@@ -1421,12 +1419,6 @@ pub enum Event {
context: Option<CompletionContext>,
entries: Vec<CompletionEntry>,
},
/// Reply to `Method::ListRewindTargets`. Clients should only open a picker
/// in response to their own pending request; the event may be broadcast.
RewindTargets {
head_entries: usize,
targets: Vec<RewindTarget>,
},
/// A rewind has truncated the authoritative session. `session` is the
/// retained canonical snapshot clients should use to reseed display state.
RewindApplied {
@@ -1569,33 +1561,6 @@ pub struct CompletionEntry {
pub invocation: Option<FeatureInvocationDescriptor>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
pub struct RewindTargetId {
#[cfg_attr(feature = "typescript", ts(type = "string"))]
#[cfg_attr(feature = "json-schema", schemars(with = "String"))]
pub segment_id: uuid::Uuid,
pub user_input_entry_index: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
pub struct RewindTarget {
pub id: RewindTargetId,
pub expected_head_entries: usize,
pub truncate_entries: usize,
pub turn_index: usize,
pub timestamp_ms: Option<u64>,
pub preview: String,
pub eligible: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub disabled_reason: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub warning: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
@@ -1998,6 +1963,23 @@ pub enum Permission {
mod tests {
use super::*;
#[test]
fn rewind_uses_public_entry_id_without_candidate_or_head_token() {
let method = Method::RewindTo {
entry_id: "user-entry".into(),
};
let wire = serde_json::json!({"method": "rewind_to", "params": {"entry_id": "user-entry"}});
assert_eq!(serde_json::to_value(&method).unwrap(), wire);
assert!(
matches!(serde_json::from_value::<Method>(wire).unwrap(), Method::RewindTo { entry_id } if entry_id == "user-entry")
);
assert!(
serde_json::from_value::<Method>(serde_json::json!({"method": "list_rewind_targets"}))
.is_err()
);
assert!(serde_json::from_value::<Method>(serde_json::json!({"method": "rewind_to", "params": {"target": {"segment_id": uuid::Uuid::nil(), "user_input_entry_index": 1}, "expected_head_entries": 3}})).is_err());
}
#[test]
fn scope_rule_defaults_to_resolved_symlink_policy() {
let rule: ScopeRule = serde_json::from_value(serde_json::json!({
+8 -10
View File
@@ -15,14 +15,14 @@ use crate::{
InvocationClientAdapter, InvocationCompletion, InvocationValue, InvokeKind, MemoryWorkerEvent,
Method, PasteArtifactAvailability, PasteArtifactMediaType, PasteArtifactRef,
PendingSubmissionSummary, PendingSubmissionsSnapshot, Permission, ReasoningConfig,
RewindSummary, RewindTarget, RewindTargetId, RunFailureKind, RunResult, RunResumeSource,
RunYieldReason, ScopeRule, Segment, SessionContentPart, SessionConversationTurn,
SessionEntryProvenance, SessionHistoryLineageBoundary, SessionHistoryPage, SessionMessageRole,
SessionSnapshot, SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment,
SubmissionDisposition, SymlinkPolicy, ToolResultDisposition, TurnResult,
UploadedFileAvailability, UploadedFileRef, WorkerBusyState, WorkerCommandAcknowledgement,
WorkerCommandDisposition, WorkerCommandEnvelope, WorkerCommandKind, WorkerEvent,
WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, WorkerStatus,
RewindSummary, RunFailureKind, RunResult, RunResumeSource, RunYieldReason, ScopeRule, Segment,
SessionContentPart, SessionConversationTurn, SessionEntryProvenance,
SessionHistoryLineageBoundary, SessionHistoryPage, SessionMessageRole, SessionSnapshot,
SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment, SubmissionDisposition,
SymlinkPolicy, ToolResultDisposition, TurnResult, UploadedFileAvailability, UploadedFileRef,
WorkerBusyState, WorkerCommandAcknowledgement, WorkerCommandDisposition, WorkerCommandEnvelope,
WorkerCommandKind, WorkerEvent, WorkerMaintenanceState, WorkerRunState, WorkerState,
WorkerStateSnapshot, WorkerStatus,
subscription::{
EventSubscriptionSelector, MAX_CORRELATION_ID_BYTES, MAX_REJECTION_MESSAGE_BYTES,
MAX_RESOURCE_ID_BYTES, MAX_SUBSCRIPTION_COLLECTION_ITEMS,
@@ -111,8 +111,6 @@ pub fn generated_protocol_types() -> String {
push_decl::<UploadedFileRef>(&cfg, &mut output);
push_decl::<ScopeRule>(&cfg, &mut output);
push_decl::<CompletionEntry>(&cfg, &mut output);
push_decl::<RewindTargetId>(&cfg, &mut output);
push_decl::<RewindTarget>(&cfg, &mut output);
push_decl::<RewindSummary>(&cfg, &mut output);
push_decl::<InFlightBlock>(&cfg, &mut output);
push_decl::<InFlightCompaction>(&cfg, &mut output);
-1
View File
@@ -401,7 +401,6 @@ pub enum WorkerInputKind {
UserIfIdle,
Notify,
Compact,
ListRewindTargets,
RegisterPeer,
}
+44 -59
View File
@@ -5,7 +5,7 @@ use std::time::{Duration, Instant};
use protocol::{
AlertLevel, AlertSource, CompletionEntry, CompletionKind, ErrorCode, Event, InFlightBlock,
InFlightSnapshot, InFlightToolCallState, InternalWorkerRef, InternalWorkerSnapshot, Method,
RewindTarget, RunResult, Segment, WorkerCommandEnvelope, WorkerStateSnapshot, WorkerStatus,
RunResult, Segment, WorkerCommandEnvelope, WorkerStateSnapshot, WorkerStatus,
};
use crate::block::{
@@ -90,10 +90,15 @@ pub struct RewindPickerScroll {
pub tail_top_offset: usize,
}
#[derive(Debug, Clone)]
pub struct RewindEntry {
pub entry_id: String,
pub preview: String,
}
#[derive(Debug, Clone)]
pub struct RewindPickerState {
pub head_entries: usize,
pub targets: Vec<RewindTarget>,
pub targets: Vec<RewindEntry>,
pub selected: usize,
pub scroll: RewindPickerScroll,
/// True after Enter submitted an authoritative `RewindTo` and before the
@@ -104,10 +109,9 @@ pub struct RewindPickerState {
}
impl RewindPickerState {
pub fn new(head_entries: usize, targets: Vec<RewindTarget>) -> Self {
let selected = targets.iter().position(|t| t.eligible).unwrap_or(0);
pub fn new(targets: Vec<RewindEntry>) -> Self {
let selected = 0;
Self {
head_entries,
targets,
selected,
scroll: RewindPickerScroll::default(),
@@ -115,7 +119,7 @@ impl RewindPickerState {
}
}
pub fn selected_target(&self) -> Option<&RewindTarget> {
pub fn selected_target(&self) -> Option<&RewindEntry> {
self.targets.get(self.selected)
}
}
@@ -334,7 +338,7 @@ pub struct App {
completion_target: String,
/// Dedicated main-view rewind picker state.
pub rewind_picker: Option<RewindPickerState>,
rewind_request_pending: bool,
rewind_entries: Vec<RewindEntry>,
/// After a successful rewind restore, ignore any queued live-update events
/// until the authoritative Worker status/snapshot catches up. This prevents
/// old stream tail events that were already in transit from re-polluting the
@@ -419,7 +423,7 @@ impl App {
pending_feature_edit: None,
completion_target: protocol::new_submission_request_id(),
rewind_picker: None,
rewind_request_pending: false,
rewind_entries: Vec::new(),
rewind_refresh_fence: false,
greeting: None,
task_store: TaskStore::new(),
@@ -1529,7 +1533,19 @@ impl App {
Event::PendingSubmissionsChanged { pending } => {
self.pending_submissions = pending;
}
Event::UserMessage { segments, .. } => {
Event::UserMessage { segments, entry_id } => {
if let Some(entry_id) = entry_id {
if !self
.rewind_entries
.iter()
.any(|entry| entry.entry_id == entry_id)
{
self.rewind_entries.push(RewindEntry {
entry_id,
preview: Segment::flatten_to_text(&segments),
});
}
}
self.turn_index += 1;
self.blocks.push(Block::TurnHeader {
turn: self.turn_index,
@@ -2038,15 +2054,6 @@ impl App {
state.selected = 0;
}
}
Event::RewindTargets {
head_entries,
targets,
} => {
if self.rewind_request_pending {
self.rewind_request_pending = false;
self.rewind_picker = Some(RewindPickerState::new(head_entries, targets));
}
}
Event::RewindApplied {
session,
input,
@@ -2467,14 +2474,14 @@ impl App {
return None;
}
self.completion = None;
self.rewind_picker = None;
self.rewind_request_pending = true;
Some(Method::ListRewindTargets)
self.rewind_picker = Some(RewindPickerState::new(
self.rewind_entries.iter().rev().cloned().collect(),
));
None
}
pub fn close_rewind_picker(&mut self) {
self.rewind_picker = None;
self.rewind_request_pending = false;
}
pub fn cancel_rewind_picker(&mut self) {
@@ -2539,17 +2546,8 @@ impl App {
);
return None;
}
let (target_id, expected_head_entries) = match picker.selected_target() {
Some(target) if target.eligible => (target.id.clone(), target.expected_head_entries),
Some(target) => {
self.push_command_diagnostic(
target
.disabled_reason
.clone()
.unwrap_or_else(|| "rewind target is disabled".into()),
);
return None;
}
let entry_id = match picker.selected_target() {
Some(target) => target.entry_id.clone(),
None => {
self.push_command_diagnostic("no rewind target is available");
return None;
@@ -2558,10 +2556,7 @@ impl App {
if let Some(picker) = self.rewind_picker.as_mut() {
picker.applying = true;
}
Some(Method::RewindTo {
target: target_id,
expected_head_entries,
})
Some(Method::RewindTo { entry_id })
}
fn command_environment(&self) -> CommandEnvironment {
@@ -2597,10 +2592,8 @@ impl App {
command: self.next_command_envelope(),
});
}
if let Some(Method::ListRewindTargets) = method.as_ref() {
self.completion = None;
self.rewind_picker = None;
self.rewind_request_pending = true;
if result.open_rewind_picker {
return self.request_rewind_picker();
}
method
}
@@ -2913,10 +2906,15 @@ impl App {
}
self.assistant_streaming = false;
self.rewind_entries.clear();
for entry in &session.entries {
use protocol::{SessionContentPart, SessionMessageRole, SessionSnapshotEntryData};
match &entry.data {
SessionSnapshotEntryData::UserInput { segments } => {
self.rewind_entries.push(RewindEntry {
entry_id: entry.entry_id.clone(),
preview: Segment::flatten_to_text(segments),
});
self.turn_index += 1;
self.blocks.push(Block::TurnHeader {
turn: self.turn_index,
@@ -3364,23 +3362,10 @@ mod rewind_refresh_tests {
fn app_with_rewind_picker() -> App {
let mut app = App::new("test".into());
app.connected = true;
app.rewind_picker = Some(RewindPickerState::new(
5,
vec![RewindTarget {
id: protocol::RewindTargetId {
segment_id: uuid::Uuid::nil(),
user_input_entry_index: 1,
},
expected_head_entries: 5,
truncate_entries: 2,
turn_index: 1,
timestamp_ms: None,
preview: "rewind target".into(),
eligible: true,
disabled_reason: None,
warning: None,
}],
));
app.rewind_picker = Some(RewindPickerState::new(vec![RewindEntry {
entry_id: "user-1".into(),
preview: "test input".into(),
}]));
app
}
+11 -3
View File
@@ -52,6 +52,7 @@ pub struct CommandEnvironment {
#[derive(Debug, Clone)]
pub struct CommandExecution {
pub method: Option<Method>,
pub open_rewind_picker: bool,
pub diagnostics: Vec<CommandDiagnostic>,
pub exit_command_mode: bool,
pub clear_input: bool,
@@ -60,6 +61,7 @@ pub struct CommandExecution {
impl CommandExecution {
pub fn diagnostic(message: impl Into<String>) -> Self {
Self {
open_rewind_picker: false,
method: None,
diagnostics: vec![CommandDiagnostic::new(message)],
exit_command_mode: false,
@@ -69,6 +71,7 @@ impl CommandExecution {
pub fn notice(message: impl Into<String>) -> Self {
Self {
open_rewind_picker: false,
method: None,
diagnostics: vec![CommandDiagnostic::new(message)],
exit_command_mode: true,
@@ -409,6 +412,7 @@ fn compact_command(invocation: CommandInvocation<'_>) -> CommandExecution {
let _ = invocation.environment;
let _ = invocation.args.raw();
CommandExecution {
open_rewind_picker: false,
method: Some(Method::Compact {
command: protocol::WorkerCommandEnvelope::new(0),
}),
@@ -423,7 +427,8 @@ fn rewind_command(invocation: CommandInvocation<'_>) -> CommandExecution {
let _ = invocation.environment;
let _ = invocation.args.raw();
CommandExecution {
method: Some(Method::ListRewindTargets),
open_rewind_picker: true,
method: None,
diagnostics: vec![CommandDiagnostic::new("rewind picker requested")],
exit_command_mode: true,
clear_input: true,
@@ -435,6 +440,7 @@ fn peer_command(invocation: CommandInvocation<'_>) -> CommandExecution {
let _ = invocation.environment;
let name = invocation.args.argv()[0].clone();
CommandExecution {
open_rewind_picker: false,
method: Some(Method::RegisterPeer { name: name.clone() }),
diagnostics: vec![CommandDiagnostic::new(format!(
"peer metadata registration requested with `{name}`"
@@ -521,7 +527,8 @@ mod tests {
let registry = CommandRegistry::builtins();
for command in ["rewind", "rollback"] {
let result = registry.dispatch(command, &env());
assert!(matches!(result.method, Some(Method::ListRewindTargets)));
assert!(result.method.is_none());
assert!(result.open_rewind_picker);
assert!(result.exit_command_mode);
assert!(result.clear_input);
assert!(result.diagnostics[0].message.contains("rewind picker"));
@@ -549,7 +556,8 @@ mod tests {
let mut paused = env();
paused.paused = true;
let result = registry.dispatch("rewind", &paused);
assert!(matches!(result.method, Some(Method::ListRewindTargets)));
assert!(result.method.is_none());
assert!(result.open_rewind_picker);
}
#[test]
+31 -69
View File
@@ -18,7 +18,7 @@ use crossterm::terminal::{EnterAlternateScreen, LeaveAlternateScreen};
use crossterm::{Command, execute};
use protocol::{Event, Method, Segment, UploadedFileRef, WorkerStatus};
#[cfg(feature = "e2e-test")]
use protocol::{Greeting, RewindSummary, RewindTarget, RewindTargetId};
use protocol::{Greeting, RewindSummary};
use ratatui::Terminal;
use ratatui::backend::CrosstermBackend;
use standalone::{StandaloneHost, StandaloneLaunchConfig};
@@ -758,10 +758,10 @@ async fn run_e2e_rewind_fixture(
});
let (_reader, mut term_rx) = TerminalEventReader::spawn()?;
let target_id = RewindTargetId {
segment_id: uuid::Uuid::from_u128(1),
user_input_entry_index: 1,
};
app.handle_worker_event(Event::UserMessage {
entry_id: Some("rewind-fixture-input".into()),
segments: vec![Segment::text("candidate rewind target")],
});
let mut rewind_submit_count = 0usize;
let mut pending_apply: Option<std::time::Instant> = None;
let apply_delay = Duration::from_millis(400);
@@ -803,45 +803,25 @@ async fn run_e2e_rewind_fixture(
.as_ref()
.map(|picker| picker.applying)
.unwrap_or(false);
if let Some(method) = handle_key(&mut app, key) {
let was_open = app.rewind_picker.is_some();
let method = handle_key(&mut app, key);
if !was_open && app.rewind_picker.is_some() {
crate::e2e_observer::emit(
"single_worker",
"rewind_picker_opened",
serde_json::json!({"targets": 1, "selected_preview": "candidate rewind target"}),
);
}
if let Some(method) = method {
match method {
Method::ListRewindTargets => {
app.handle_worker_event(Event::RewindTargets {
head_entries: 3,
targets: vec![RewindTarget {
id: target_id.clone(),
expected_head_entries: 3,
truncate_entries: 1,
turn_index: 1,
timestamp_ms: Some(1),
preview: "candidate rewind target".to_string(),
eligible: true,
disabled_reason: None,
warning: None,
}],
});
crate::e2e_observer::emit(
"single_worker",
"rewind_picker_opened",
serde_json::json!({
"targets": 1,
"selected_preview": "candidate rewind target",
}),
);
}
Method::RewindTo {
target,
expected_head_entries,
} => {
Method::RewindTo { entry_id } => {
rewind_submit_count += 1;
pending_apply = Some(std::time::Instant::now());
crate::e2e_observer::emit(
"single_worker",
"rewind_submit_sent",
serde_json::json!({
"segment_id": target.segment_id.to_string(),
"user_input_entry_index": target.user_input_entry_index,
"expected_head_entries": expected_head_entries,
"entry_id": entry_id,
"submit_count": rewind_submit_count,
}),
);
@@ -1725,10 +1705,7 @@ mod tests {
use super::*;
use crate::text_selection::{HistoryViewport, SelectionRow};
use async_trait::async_trait;
use protocol::{
Event, RewindTarget, RewindTargetId, Segment, UploadedFileAvailability, UploadedFileRef,
WorkerStatus,
};
use protocol::{Event, Segment, UploadedFileAvailability, UploadedFileRef, WorkerStatus};
#[test]
fn standalone_console_starts_with_in_process_connection_ready() {
@@ -3119,14 +3096,17 @@ mod tests {
&mut app,
KeyEvent::new(KeyCode::Char('r'), KeyModifiers::CONTROL),
);
assert!(matches!(idle, Some(Method::ListRewindTargets)));
assert!(idle.is_none());
assert!(app.rewind_picker.is_some());
app.close_rewind_picker();
app.set_worker_status(WorkerStatus::Paused);
let paused = handle_key(
&mut app,
KeyEvent::new(KeyCode::Char('r'), KeyModifiers::CONTROL),
);
assert!(matches!(paused, Some(Method::ListRewindTargets)));
assert!(paused.is_none());
assert!(app.rewind_picker.is_some());
}
#[test]
@@ -3148,21 +3128,13 @@ mod tests {
fn rewind_picker_close_returns_to_history_view() {
let mut app = App::new("agent".to_string());
app.connected = true;
app.handle_worker_event(Event::RewindTargets {
head_entries: 1,
targets: vec![],
});
assert!(app.rewind_picker.is_none());
let method = handle_key(
&mut app,
KeyEvent::new(KeyCode::Char('r'), KeyModifiers::CONTROL),
);
assert!(matches!(method, Some(Method::ListRewindTargets)));
app.handle_worker_event(Event::RewindTargets {
head_entries: 1,
targets: vec![],
});
assert!(method.is_none());
assert!(app.rewind_picker.is_some());
let method = handle_key(&mut app, KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE));
@@ -3244,13 +3216,13 @@ mod tests {
#[test]
fn rewind_apply_rejects_non_empty_composer_and_paused_status() {
let mut app = App::new("agent".to_string());
app.rewind_picker = Some(crate::app::RewindPickerState::new(1, vec![rewind_target()]));
app.rewind_picker = Some(crate::app::RewindPickerState::new(vec![rewind_target()]));
type_keys(&mut app, "draft");
assert!(app.submit_rewind_picker().is_none());
assert!(has_alert(&app, "composer is not empty"));
let mut app = App::new("agent".to_string());
app.rewind_picker = Some(crate::app::RewindPickerState::new(1, vec![rewind_target()]));
app.rewind_picker = Some(crate::app::RewindPickerState::new(vec![rewind_target()]));
app.set_worker_status(WorkerStatus::Paused);
assert!(app.submit_rewind_picker().is_none());
assert!(has_alert(
@@ -3265,7 +3237,7 @@ mod tests {
app.scroll.top_offset = 3;
app.scroll.turn_starts = vec![0, 5, 9];
app.scroll.total_lines = 42;
app.rewind_picker = Some(crate::app::RewindPickerState::new(1, vec![rewind_target()]));
app.rewind_picker = Some(crate::app::RewindPickerState::new(vec![rewind_target()]));
let original_top_offset = app.scroll.top_offset;
let original_turn_starts = app.scroll.turn_starts.clone();
let original_total_lines = app.scroll.total_lines;
@@ -3282,20 +3254,10 @@ mod tests {
assert_eq!(app.scroll.total_lines, original_total_lines);
}
fn rewind_target() -> RewindTarget {
RewindTarget {
id: RewindTargetId {
segment_id: uuid::Uuid::nil(),
user_input_entry_index: 0,
},
expected_head_entries: 1,
truncate_entries: 0,
turn_index: 1,
timestamp_ms: Some(1),
preview: "retry this".into(),
eligible: true,
disabled_reason: None,
warning: None,
fn rewind_target() -> crate::app::RewindEntry {
crate::app::RewindEntry {
entry_id: "user-1".into(),
preview: "test input".into(),
}
}
+2 -27
View File
@@ -728,7 +728,7 @@ fn draw_rewind_picker(
.fg(Color::Yellow)
.add_modifier(Modifier::BOLD),
),
Span::raw(format!(" head={} ", picker.head_entries)),
Span::raw(" "),
];
header.extend(action_spans);
logical.push(Line::from(header));
@@ -751,38 +751,13 @@ fn draw_rewind_picker(
Style::default()
.bg(Color::DarkGray)
.add_modifier(Modifier::BOLD)
} else if target.eligible {
Style::default()
} else {
Style::default().fg(Color::DarkGray)
Style::default()
};
let ts = target
.timestamp_ms
.map(|ts| format!("{}", ts))
.unwrap_or_else(|| "-".into());
logical.push(Line::from(vec![
Span::styled(marker.to_owned(), base_style),
Span::styled(
format!(
" turn {} idx {} ts {} ",
target.turn_index, target.id.user_input_entry_index, ts
),
base_style,
),
Span::styled(target.preview.clone(), base_style),
]));
if let Some(warning) = target.warning.as_ref() {
logical.push(Line::from(Span::styled(
format!(" warning: {warning}"),
Style::default().fg(Color::Yellow),
)));
}
if let Some(reason) = target.disabled_reason.as_ref() {
logical.push(Line::from(Span::styled(
format!(" disabled: {reason}"),
Style::default().fg(Color::Red),
)));
}
}
}
+1 -2
View File
@@ -12,13 +12,12 @@ pub enum WorkerInputKind {
UserIfIdle,
Notify,
Compact,
ListRewindTargets,
RegisterPeer,
}
impl WorkerInputKind {
pub fn is_empty_content_allowed(&self) -> bool {
matches!(self, Self::Compact | Self::ListRewindTargets)
matches!(self, Self::Compact)
}
}
+10 -10
View File
@@ -6752,16 +6752,16 @@ fn input_protocol_event(input: &WorkerInput) -> Option<protocol::Event> {
// Submit is projected only after the Worker commits UserInput. Queued
// payloads must never become model- or client-visible history early.
WorkerInputKind::User | WorkerInputKind::UserIfIdle | WorkerInputKind::Notify => None,
WorkerInputKind::Compact
| WorkerInputKind::ListRewindTargets
| WorkerInputKind::RegisterPeer => Some(protocol::Event::SystemItem {
entry_id: None,
item: serde_json::json!({
"kind": "embedded_worker_command_input",
"command": input.kind,
"content": input.content.clone(),
}),
}),
WorkerInputKind::Compact | WorkerInputKind::RegisterPeer => {
Some(protocol::Event::SystemItem {
entry_id: None,
item: serde_json::json!({
"kind": "embedded_worker_command_input",
"command": input.kind,
"content": input.content.clone(),
}),
})
}
}
}
+1 -2
View File
@@ -4255,7 +4255,6 @@ where
unreachable!("Notify input is dispatched before ordinary input mapping")
}
WorkerInputKind::Compact => unreachable!("compact input is dispatched above"),
WorkerInputKind::ListRewindTargets => (Method::ListRewindTargets, None),
WorkerInputKind::RegisterPeer => (
Method::RegisterPeer {
name: input.content.trim().to_string(),
@@ -7884,7 +7883,7 @@ mod tests {
));
let old_send = backend.run_on_adapter_runtime(async move {
old_handle
.send(Method::ListRewindTargets)
.send(Method::ListWorkers)
.await
.map_err(|error| error.to_string())
});
+11 -55
View File
@@ -32,8 +32,8 @@ use protocol::{
AlertLevel, AlertSource, CommandEvent as ProtocolCommandEvent,
CommandSnapshot as ProtocolCommandSnapshot, CommandStatus as ProtocolCommandStatus,
CommandStream as ProtocolCommandStream, CommandStreamSlice as ProtocolCommandStreamSlice,
ErrorCode, Event, Method, RewindTargetId, RunResult, TurnResult, UploadedFileRef,
WorkerBusyState, WorkerCommandAcknowledgement, WorkerCommandDisposition, WorkerCommandEnvelope,
ErrorCode, Event, Method, RunResult, TurnResult, UploadedFileRef, WorkerBusyState,
WorkerCommandAcknowledgement, WorkerCommandDisposition, WorkerCommandEnvelope,
WorkerCommandKind, WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStatus,
};
use workdir::{
@@ -2971,32 +2971,9 @@ async fn controller_loop<C, St>(
}
}
Method::ListRewindTargets => match shared_state.catalog_status() {
WorkerStatus::Idle | WorkerStatus::Paused => {
emit_rewind_targets(&worker, &working_event_tx)
}
WorkerStatus::Running | WorkerStatus::Stopped => {
let _ = working_event_tx.send(Event::Error {
code: ErrorCode::AlreadyRunning,
message: "Worker is already executing a turn; rewind can only run while idle or paused"
.into(),
});
}
},
Method::RewindTo {
target,
expected_head_entries,
} => match shared_state.catalog_status() {
Method::RewindTo { entry_id } => match shared_state.catalog_status() {
WorkerStatus::Idle => {
if apply_rewind(
&mut worker,
&working_event_tx,
target,
expected_head_entries,
)
.await
{
if apply_rewind(&mut worker, &working_event_tx, entry_id).await {
worker.clear_in_flight_events();
publish_context_usage(
&shared_state,
@@ -3018,8 +2995,9 @@ async fn controller_loop<C, St>(
WorkerStatus::Running | WorkerStatus::Stopped => {
let _ = working_event_tx.send(Event::Error {
code: ErrorCode::AlreadyRunning,
message: "Worker is already executing a turn; rewind can only run while idle or paused"
.into(),
message:
"Worker is already executing a turn; rewind can only run while idle"
.into(),
});
}
},
@@ -3668,10 +3646,10 @@ where
);
}
}
Some(Method::ListRewindTargets | Method::RewindTo { .. }) => {
Some(Method::RewindTo { .. }) => {
let _ = working_event_tx.send(Event::Error {
code: ErrorCode::AlreadyRunning,
message: "Worker is already executing a turn; rewind/compact can only run while idle or paused"
message: "Worker is already executing a turn; rewind/compact can only run while idle"
.into(),
});
}
@@ -3749,38 +3727,16 @@ where
}
}
fn emit_rewind_targets<C, St>(worker: &Worker<C, St>, working_event_tx: &broadcast::Sender<Event>)
where
C: LlmClient + 'static,
St: Store,
{
match worker.list_rewind_targets() {
Ok((head_entries, targets)) => {
let _ = working_event_tx.send(Event::RewindTargets {
head_entries,
targets,
});
}
Err(err) => {
let _ = working_event_tx.send(Event::Error {
code: ErrorCode::Internal,
message: err.to_string(),
});
}
}
}
async fn apply_rewind<C, St>(
worker: &mut Worker<C, St>,
working_event_tx: &broadcast::Sender<Event>,
target: RewindTargetId,
expected_head_entries: usize,
entry_id: String,
) -> bool
where
C: LlmClient + 'static,
St: Store,
{
match worker.rewind_to(target, expected_head_entries).await {
match worker.rewind_to(entry_id).await {
Ok(applied) => {
let session =
session_store::public_snapshot::project_current_session_snapshot(&applied.entries);
+179 -140
View File
@@ -531,8 +531,8 @@ use crate::skill::{SkillActivationResponse, SkillClientError};
use async_trait::async_trait;
use protocol::{
AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, CompactionPhase,
CompactionTrigger, ErrorCode, Event, InFlightCompaction, RewindSummary, RewindTarget,
RewindTargetId, RunFailureKind, RunResumeSource, RunYieldReason, Segment,
CompactionTrigger, ErrorCode, Event, InFlightCompaction, RewindSummary, RunFailureKind,
RunResumeSource, RunYieldReason, Segment,
};
use tokio::net::UnixStream;
use tokio::sync::broadcast;
@@ -3714,22 +3714,15 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
&self.store
}
/// List user-submitted turns in newest-first order for the manual rewind picker.
pub fn list_rewind_targets(&self) -> Result<(usize, Vec<RewindTarget>), RewindError> {
/// Rewind the current segment to just before the specified public user input.
pub async fn rewind_to(&mut self, entry_id: String) -> Result<RewindAppliedState, RewindError> {
// Reject invalid targets before invoking rewrite hooks or quiescing tasks.
// Resolve again under the append lock after the asynchronous barrier.
let loc = self.segment_state.location();
let entries = self.store.read_all(loc.session_id, loc.segment_id)?;
Ok((
entries.len(),
build_rewind_targets(loc.segment_id, &entries),
))
}
/// Truncate the current segment to just before a previously listed user input.
pub async fn rewind_to(
&mut self,
target: RewindTargetId,
expected_head_entries: usize,
) -> Result<RewindAppliedState, RewindError> {
resolve_rewind_input(
&self.store.read_all(loc.session_id, loc.segment_id)?,
&entry_id,
)?;
let _rewrite_guard = self
.prepare_session_rewrite(SessionRewriteKind::Rewind)
.await
@@ -3742,29 +3735,9 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.lock()
.expect("segment append lock poisoned");
let loc = self.segment_state.location();
if target.segment_id != loc.segment_id {
return Err(RewindError::Invalid(
"rewind target belongs to a different segment".into(),
));
}
let entries = self.store.read_all(loc.session_id, loc.segment_id)?;
if entries.len() != expected_head_entries {
return Err(RewindError::Invalid(format!(
"session head changed since picker opened (expected {expected_head_entries}, current {})",
entries.len()
)));
}
let input = match entries.get(target.user_input_entry_index) {
Some(LogEntry::AnnotatedUserInput { segments, .. }) => segments.clone(),
_ => {
return Err(RewindError::Invalid(
"rewind target is no longer a user message".into(),
));
}
};
let truncate_entries = rewind_truncate_entries(&entries, target.user_input_entry_index);
let (entry_index, input) = resolve_rewind_input(&entries, &entry_id)?;
let truncate_entries = rewind_truncate_entries(&entries, entry_index);
let retained = entries[..truncate_entries].to_vec();
let tool_side_effect_warning = suffix_has_tool_side_effects(&entries[truncate_entries..]);
let state = segment_log::collect_state(&retained);
@@ -8994,37 +8967,21 @@ pub struct RewindAppliedState {
pub summary: RewindSummary,
}
fn build_rewind_targets(segment_id: uuid::Uuid, entries: &[LogEntry]) -> Vec<RewindTarget> {
let head_entries = entries.len();
let mut turn_index = 0usize;
let mut targets = Vec::new();
for (entry_index, entry) in entries.iter().enumerate() {
let (segments, ts) = match entry {
LogEntry::AnnotatedUserInput { segments, ts, .. } => (segments, ts),
_ => continue,
};
turn_index += 1;
let truncate_entries = rewind_truncate_entries(entries, entry_index);
let tool_warning = suffix_has_tool_side_effects(&entries[truncate_entries..]);
targets.push(RewindTarget {
id: RewindTargetId {
segment_id,
user_input_entry_index: entry_index,
},
expected_head_entries: head_entries,
truncate_entries,
turn_index,
timestamp_ms: Some(*ts),
preview: preview_segments(segments),
eligible: true,
disabled_reason: None,
warning: tool_warning.then(|| {
"history suffix will be discarded; tool side effects are not undone".into()
}),
});
}
targets.reverse();
targets
fn resolve_rewind_input(
entries: &[LogEntry],
entry_id: &str,
) -> Result<(usize, Vec<Segment>), RewindError> {
// Resolve only actual user submissions in this segment, never seeded
// ancestor history or a display position inferred from matching text.
entries.iter().enumerate().find_map(|(index, entry)| {
let LogEntry::AnnotatedUserInput { history, segments, .. } = entry else { return None; };
let user = history.iter().find(|entry| matches!(&entry.item,
session_store::LoggedItem::Message { role: session_store::LoggedRole::User, .. }
))?;
(user.metadata.entry_id.0 == entry_id).then(|| (index, segments.clone()))
}).ok_or_else(|| RewindError::Invalid(
"rewind entry is not a user input in the current segment; it may have been removed or belong to another history".into()
))
}
fn rewind_truncate_entries(entries: &[LogEntry], user_input_entry_index: usize) -> usize {
@@ -9052,6 +9009,7 @@ fn logged_item_is_tool_call(item: &session_store::LoggedItem) -> bool {
matches!(item, session_store::LoggedItem::ToolCall { .. })
}
#[cfg(test)]
fn preview_segments(segments: &[Segment]) -> String {
let mut preview = String::new();
for segment in segments {
@@ -12305,48 +12263,139 @@ mod build_summary_prompt_tests {
);
}
fn rewind_test_inputs<C: LlmClient + 'static, St: Store>(
worker: &Worker<C, St>,
) -> Vec<(String, usize)> {
let loc = worker.segment_state.location();
let entries = worker
.store
.read_all(loc.session_id, loc.segment_id)
.unwrap();
let public = session_store::public_snapshot::project_current_session_snapshot(&entries);
public
.entries
.iter()
.filter_map(|entry| {
if !matches!(
entry.data,
protocol::SessionSnapshotEntryData::UserInput { .. }
) {
return None;
}
let index = entries.iter().position(|log| {
matches!(log, LogEntry::AnnotatedUserInput { history, .. }
if history.iter().any(|h| h.metadata.entry_id.0 == entry.entry_id))
})?;
Some((
entry.entry_id.clone(),
rewind_truncate_entries(&entries, index),
))
})
.rev()
.collect()
}
#[tokio::test]
async fn rewind_target_listing_is_newest_first_and_warns_on_tool_suffix() {
let (_dir, worker) = rewind_test_worker().await;
append_user_turn(&worker, 10, "first message");
append_user_turn(&worker, 20, "second message");
async fn rewind_resolves_duplicate_text_by_public_entry_identity() {
let (_dir, mut worker) = rewind_test_worker().await;
append_user_turn(&worker, 10, "same input");
append_user_turn(&worker, 20, "same input");
let inputs = rewind_test_inputs(&worker);
assert_ne!(inputs[0].0, inputs[1].0);
worker.rewind_to(inputs[0].0.clone()).await.unwrap();
assert_eq!(rewind_test_inputs(&worker), vec![inputs[1].clone()]);
}
#[tokio::test]
async fn rewind_rejects_unknown_foreign_and_non_user_ids_without_changing_log() {
let (_dir, mut worker) = rewind_test_worker().await;
let (_other_dir, other) = rewind_test_worker().await;
append_user_turn(&worker, 10, "same input");
append_user_turn(&other, 10, "same input");
let assistant =
crate::session_history::test_logged_history_entry(session_store::LoggedItem::Message {
role: session_store::LoggedRole::Assistant,
content: vec![session_store::LoggedContentPart::Text {
text: "answer".into(),
}],
});
let assistant_id = assistant.metadata.entry_id.0.clone();
append_test_entry(
&worker,
LogEntry::AnnotatedToolResult {
ts: 30,
entry: crate::session_history::test_logged_history_entry(
session_store::LoggedItem::ToolResult {
call_id: "call-1".into(),
summary: "wrote a file".into(),
content: None,
attachments: Vec::new(),
disposition: Default::default(),
is_error: false,
},
),
LogEntry::AnnotatedAssistantItem {
ts: 12,
entry: assistant,
},
);
let (head_entries, targets) = worker.list_rewind_targets().unwrap();
let loc = worker.segment_state.location();
let before = worker
.store
.read_all(loc.session_id, loc.segment_id)
.unwrap();
for id in [
String::new(),
"missing-entry".into(),
rewind_test_inputs(&other)[0].0.clone(),
assistant_id,
] {
assert!(
worker
.rewind_to(id)
.await
.unwrap_err()
.to_string()
.contains("not a user input")
);
assert_eq!(
serde_json::to_value(
worker
.store
.read_all(loc.session_id, loc.segment_id)
.unwrap()
)
.unwrap(),
serde_json::to_value(&before).unwrap()
);
}
}
assert_eq!(
head_entries,
worker
.store
.read_all(loc.session_id, loc.segment_id)
.unwrap()
.len()
#[tokio::test]
async fn rewind_rejects_user_input_from_seeded_ancestor_history() {
let (_dir, mut worker) = rewind_test_worker().await;
let input = crate::session_history::test_logged_history_entry(
session_store::LoggedItem::from(agen::Item::user_message("ancestor input")),
);
assert_eq!(targets.len(), 2);
assert_eq!(targets[0].preview, "second message");
assert_eq!(targets[1].preview, "first message");
let id = input.metadata.entry_id.0.clone();
let loc = worker.segment_state.location();
let mut entries = worker
.store
.read_all(loc.session_id, loc.segment_id)
.unwrap();
let LogEntry::AnnotatedSegmentStart { history, .. } = &mut entries[0] else {
panic!("segment start");
};
history.push(input);
worker
.store
.create_segment(loc.session_id, loc.segment_id, &entries)
.unwrap();
assert!(
targets[0]
.warning
.as_ref()
.unwrap()
.contains("tool side effects")
worker
.rewind_to(id)
.await
.unwrap_err()
.to_string()
.contains("current segment")
);
assert_eq!(
serde_json::to_value(
worker
.store
.read_all(loc.session_id, loc.segment_id)
.unwrap()
)
.unwrap(),
serde_json::to_value(entries).unwrap()
);
}
@@ -12371,11 +12420,11 @@ mod build_summary_prompt_tests {
),
},
);
let (head_entries, targets) = worker.list_rewind_targets().unwrap();
let expected_truncate_entries = targets[0].truncate_entries;
let target = targets[0].id.clone();
let targets = rewind_test_inputs(&worker);
let expected_truncate_entries = targets[0].1;
let target = targets[0].0.clone();
let applied = worker.rewind_to(target, head_entries).await.unwrap();
let applied = worker.rewind_to(target).await.unwrap();
assert_eq!(preview_segments(&applied.input), "second message");
assert_eq!(
@@ -12405,10 +12454,10 @@ mod build_summary_prompt_tests {
.pending_submission_handle()
.accept_notification("notification-1".into(), "keep me".into())
.unwrap();
let (head_entries, targets) = worker.list_rewind_targets().unwrap();
let targets = rewind_test_inputs(&worker);
worker
.rewind_to(targets.last().unwrap().id.clone(), head_entries)
.rewind_to(targets.last().unwrap().0.clone())
.await
.unwrap();
@@ -12465,16 +12514,11 @@ mod build_summary_prompt_tests {
},
);
let (head_entries, targets) = worker.list_rewind_targets().unwrap();
let targets = rewind_test_inputs(&worker);
assert_eq!(targets.len(), 2);
assert_eq!(targets[0].preview, "second message");
assert!(targets[0].truncate_entries > 0);
assert!(targets[0].warning.is_some());
assert!(targets[0].1 > 0);
let applied = worker
.rewind_to(targets[0].id.clone(), head_entries)
.await
.unwrap();
let applied = worker.rewind_to(targets[0].0.clone()).await.unwrap();
assert_eq!(applied.summary.truncated_to_entries, 5);
assert!(matches!(
applied.entries.first(),
@@ -12539,19 +12583,20 @@ mod build_summary_prompt_tests {
}
#[tokio::test]
async fn rewind_apply_rejects_stale_head() {
async fn rewind_by_entry_id_includes_history_appended_after_selection() {
let (_dir, mut worker) = rewind_test_worker().await;
append_user_turn(&worker, 10, "first message");
let (head_entries, targets) = worker.list_rewind_targets().unwrap();
let targets = rewind_test_inputs(&worker);
append_user_turn(&worker, 20, "newer message");
let err = worker
.rewind_to(targets[0].id.clone(), head_entries)
.await
.unwrap_err()
.to_string();
assert!(err.contains("session head changed"));
let applied = worker.rewind_to(targets[0].0.clone()).await.unwrap();
assert_eq!(preview_segments(&applied.input), "first message");
assert!(rewind_test_inputs(&worker).is_empty());
let err = worker.rewind_to(targets[0].0.clone()).await.unwrap_err();
assert!(
err.to_string()
.contains("not a user input in the current segment")
);
}
#[tokio::test]
@@ -15421,7 +15466,7 @@ permission = "write"
assert!(pending.pending_notifications.is_empty());
assert!(pending.activating_notifications.is_empty());
}
let (head, targets) = original.list_rewind_targets().unwrap();
let targets = rewind_test_inputs(&original);
let target = targets.last().unwrap();
let old = store
.read_all(original.session_id(), original.segment_id())
@@ -15431,7 +15476,7 @@ permission = "write"
store
.invocation_failure
.store(replacement_fault, Ordering::SeqCst);
let applied = original.rewind_to(target.id.clone(), head).await;
let applied = original.rewind_to(target.0.clone()).await;
let disk = store
.read_all(original.session_id(), original.segment_id())
.unwrap();
@@ -15456,15 +15501,9 @@ permission = "write"
replacement_fault
);
let applied = applied.unwrap();
assert_eq!(
applied.summary.truncated_to_entries,
target.truncate_entries
);
assert_eq!(
applied.summary.discarded_entries,
head - target.truncate_entries
);
assert_eq!(applied.entries.len(), target.truncate_entries);
assert_eq!(applied.summary.truncated_to_entries, target.1);
assert_eq!(applied.summary.discarded_entries, old.len() - target.1);
assert_eq!(applied.entries.len(), target.1);
assert_eq!(original.segment_state.entries_written(), disk.len());
assert_eq!(
serde_json::to_value(original.sink.subscribe_with_snapshot().0)
@@ -15474,10 +15513,10 @@ permission = "write"
} else {
assert!(applied.is_err());
}
assert_eq!(disk.len(), target.truncate_entries + 1);
assert_eq!(disk.len(), target.1 + 1);
assert_eq!(
serde_json::to_value(&disk[..target.truncate_entries]).unwrap(),
serde_json::to_value(&old[..target.truncate_entries]).unwrap()
serde_json::to_value(&disk[..target.1]).unwrap(),
serde_json::to_value(&old[..target.1]).unwrap()
);
let LogEntry::Extension {
domain, payload, ..
+1 -1
View File
@@ -2730,7 +2730,7 @@ async fn shutdown_closes_method_admission_before_terminal_confirmation() {
loop {
tokio::select! {
biased;
result = handle.send(Method::ListRewindTargets) => {
result = handle.send(Method::ListWorkers) => {
if result.is_err() {
break;
}
-4
View File
@@ -705,7 +705,6 @@ pub enum WorkerInputKind {
UserIfIdle,
Notify,
Compact,
ListRewindTargets,
RegisterPeer,
}
@@ -3271,7 +3270,6 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
WorkerInputKind::UserIfIdle => EmbeddedWorkerInputKind::UserIfIdle,
WorkerInputKind::Notify => EmbeddedWorkerInputKind::Notify,
WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact,
WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets,
WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer,
},
content: request.content,
@@ -5310,7 +5308,6 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
WorkerInputKind::UserIfIdle => EmbeddedWorkerInputKind::UserIfIdle,
WorkerInputKind::Notify => EmbeddedWorkerInputKind::Notify,
WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact,
WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets,
WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer,
},
content: request.content,
@@ -6748,7 +6745,6 @@ mod tests {
EmbeddedWorkerInputKind::UserIfIdle,
EmbeddedWorkerInputKind::Notify,
EmbeddedWorkerInputKind::Compact,
EmbeddedWorkerInputKind::ListRewindTargets,
EmbeddedWorkerInputKind::RegisterPeer,
] {
let input = EmbeddedWorkerInput {
-2
View File
@@ -6930,7 +6930,6 @@ fn worker_input_request_from_api(
"user" => WorkerInputKind::User,
"notify" => WorkerInputKind::Notify,
"compact" => WorkerInputKind::Compact,
"list_rewind_targets" => WorkerInputKind::ListRewindTargets,
"register_peer" => WorkerInputKind::RegisterPeer,
value => {
return Err(Error::InvalidInput(format!(
@@ -29168,7 +29167,6 @@ pub(crate) fn authorize_browser_worker_method(
| protocol::Method::Cancel { .. }
| protocol::Method::Pause { .. }
| protocol::Method::Compact { .. }
| protocol::Method::ListRewindTargets
| protocol::Method::RewindTo { .. }
| protocol::Method::Shutdown { .. }
| protocol::Method::ListCompletions { .. }
@@ -0,0 +1,310 @@
# develop 非同期・排他制御監査(2026-10-10)
## 結論
優先すべきは Mutex の一律削減ではなく、**制御系を止める backpressure、deadline の抜け、async executor 上の同期 I/O、キャンセル時の所有権喪失**の解消である。その後に、読み取りの過剰排他と独立した初期化・取得処理の直列化を修正する。
ソース上で根拠を確認した指摘を16件に整理した。P1は5件、P2は10件、P3は1件。条件付きのリスクを含み、以下は本番での発生件数や実測性能の報告ではない。
## 対象と検証方法
- Repository: `main`
- 対象: 取得時の最新 `develop`、`a9a67fd6adaf31e842faddb29d175d0f0b8bc2de`
- HEAD と `git ls-remote origin refs/heads/develop` の一致、および調査開始時の clean 状態を確認。
- 主な調査領域: workspace-server、worker-runtime、worker、workdir、agen、mcp、session-store、standalone、feature-storage、tools。client、TUI、Webの関連経路も確認。
- Mutex/RwLock、同期 filesystem/SQLite、awaitを含むループ、queue/semaphore、timeout/cancellationを探索し、該当関数と呼び出し側を読む静的監査。テスト専用コードのロックを製品問題には数えていない。
- 再現テスト・負荷試験は実施していない。すべての非同期関数を網羅した証明でもない。行番号は上記コミットに固定。
- `cargo check`: 成功。
- state-contract 命名検査: **既存状態で失敗**(unapproved 30行、stale exceptions 9件)。主に frozen repository-key migration と関連テスト・既存レポート。今回の非同期指摘とは別件であり変更していない。
- 実装ソース変更なし。本レポートのみ追加。commit/pushは行っていない。
### 優先度の意味
- **P1**: 条件成立時に進行不能、停止・キャンセル不能、広い executor 阻害につながるため優先修正。
- **P2**: 負荷、低速I/O、競合、キャンセル等で顕在化する応答性・所有権・スケーラビリティの問題。
- **P3**: 正しさを壊さないが、独立処理の不要な待ちを増やす改善。
## 一覧
| ID | 優先度 | 指摘 |
|---|---|---|
| A01 | P1 | Worker protocol の双方向 bounded queue が循環待ちを作る |
| A02 | P1 | Backend の async DB API が同期 Mutex・SQLite を直接実行する |
| A03 | P1 | MCP の request/shutdown timeout が stdin 待ちを覆わない |
| A04 | P1 | Codex refresh の body 待ちが auth Mutex を無期限に保持し得る |
| A05 | P1 | Agen の一部 policy gate が Cancel/Pause を処理しない |
| A06 | P2 | External Workdir の output 待ちが cancel 用の枠まで消費する |
| A07 | P2 | Workdir session cleanup のキャンセルで未完了レコードを失う |
| A08 | P2 | LocalWorkdirSession の async API が同期 filesystem/search を直接実行する |
| A09 | P2 | Bash output pump が無制限に drain し timeout/cancel に戻らない |
| A10 | P2 | Workdir 全体の排他 try_lock が並列 Read/Search を busy エラーにする |
| A11 | P2 | Workspace API cache の全体ロック中に初期化し、Runtime attach も直列 |
| A12 | P2 | FeatureStorage の管理 Mutex 内で fsync・DB open/migration/backup を行う |
| A13 | P2 | 大きな paste の永続化が Worker の async 実行を同期的に止める |
| A14 | P2 | PDF parser の blocking task が tool キャンセル後も残る |
| A15 | P2 | Backend TUI が同期 poll と操作完了待ちを入力ループに置いている |
| A16 | P3 | 独立した画面データ取得を直列化している |
## 詳細
### A01 — Worker protocol の双方向 backpressure による循環待ち
**根拠**
- `crates/workspace-server/src/server.rs:29214–29234`: WebSocket の受信 branch 内で method forwarding を await。その間 `connection.events` を消費しない。
- `crates/workspace-server/src/server/worker_operations.rs:592–607`: Worker ごとの session Mutex を保持して `methods.send(method).await`。
- `crates/workspace-server/src/server.rs:29324–29337, 29366–29397`: methods容量256、events容量512。bridgeは同じ select loop でmethod受信とevent送信を処理し、event送信の容量待ち中にmethodを消費できない。
**発生条件と影響**
bridgeが満杯のevent queueへの送信で停止し、browser側がmethod queueを満杯にして次の送信を待つと、`browser→method容量→bridge→event容量→browser` の循環待ちになる。session Mutexまで保持されるため同じWorkerの別接続からの送信やcleanupも巻き込む。これは単なる遅延ではなく、双方が生存したまま自力で進めなくなる構造。
Standaloneにも類似の制御阻害がある。`crates/standalone/src/host.rs:592–612, 674–678` はevent送信完了までmethod受信を止める。`crates/client/src/transport/in_process.rs:7, 28–29, 73–74` の256件channelが満杯になると、接続を維持した遅いconsumerのためにCancel/Pause/Shutdownのdispatchも止まる。この類似箇所単体の優先度はP2相当。
**改善**: 順序を保つevent writerとmethod/control受信を独立して進行させる。bufferはboundedのまま、lag時の切断・再同期方針と両pumpの共通終了処理を明示する。session/lifecycle排他を無期限のtransport容量待ちに使わない。ただし排他を単に除去してattach/stopとの整合性を崩してはいけない。
**回帰検証**: 双方のqueueをbarrierで満杯にし、Cancelまたは接続終了で規定時間内に収束すること。同じWorkerの別接続も巻き添えで停止しないこと。
### A02 — async と宣言した DB API の中身が同期 Mutex + SQLite
**根拠**
- `crates/workspace-server/src/store.rs:2108–2109, 2156–2172`: 共有 `Arc<Mutex<Connection>>` を同期取得し、そのままSQLを実行。
- 同 `3941–3951, 3954–4011`: `async fn get_workspace` / `update_workspace_display_name` が同期 query/IMMEDIATE transaction を直接実行。
- 同 `12315–12321`: SQLite busy timeoutは5秒。
- `crates/workspace-server/src/worker_projection.rs:167–172`: Tokio taskからも同期DB mutationへ進む。
- `crates/workspace-server/src/server.rs:19146–19161, 19201–19232`: Subjektiv のasync handlerも `open_subjektiv_store` と同期store処理を直呼び。
- `crates/subjektiv/src/lib.rs:624–671`: store openで `FeatureDatabase` を開き、scope確認にもtransactionを使用する。
**影響**: DB競合・fsync・低速storageがTokio executor threadを占有する。同じconnection Mutexを待つリクエストが増えると、複数のexecutor threadが同じDB待ちに使われる。`.await`という表記だけではyieldせず、外からのTokio timeoutも実行中の同期SQLを中断できない。
Subjektivは短命なstoreのopenを行ごとに繰り返す経路もある。他に強参照がない場合、FeatureStorageのweak cacheが失効し、DB初期化経路まで再実行し得る(A12参照)。
**改善**: connection取得からtransaction完了までを、boundedなblocking実行または専用DB threadへ移す。asyncメソッドだけでなくprojection等の同期API callerも対象にする。connection共有の排他とcommit→publication順序は維持する。`std::Mutex`を`tokio::Mutex`へ替えるだけではSQLのblockingは消えない。読み取りpoolはその後、整合性要件と負荷測定を踏まえて検討する。
**回帰検証**: 別connectionでwrite lockを保持し、更新APIが待つ間もtimer/health/control taskが進むこと。queue上限・overload時の応答・transaction順序も検証。
### A03 — MCP の timeout が最も詰まりやすい stdin 書き込みを覆っていない
**根拠**
- `crates/mcp/src/stdio.rs:905–910`: requestのwriteを終えた後に `request_timeout` を開始。
- 同 `1509–1524`: stdin Mutex取得、`write_all`、`flush`にdeadlineがない。
- 同 `1388–1396`: stdout readerからの応答書き込みも同じstdinを使う。
- 同 `1015–1036`: shutdownもstdin Mutex取得後にprocess待ちtimeoutを開始。
- caller: `crates/worker/src/feature/mcp.rs:1085–1094` のtool実行 → call_tool → request。
**影響**: 初期化後にMCP serverがstdinを読まなくなりpipeが埋まると、requestは設定timeoutに到達しない。reader側の返信がstdinを保持して止まった場合、shutdownもterminate/kill fallbackへ進めない。
**改善**: lock取得・frame書き込み・response読み取りを一つのdeadlineで覆う。shutdownはstdin取得失敗でもprocessを止められる経路を持つ。JSON-lineの途中でwriteをキャンセルした場合は接続を再利用しないか、完全frameを所有するwriter taskで扱う。frame混線防止のMutex自体は必要。
**回帰検証**: initialize後にstdinを読まないmock serverとpipe容量超過のrequestを使い、requestとshutdown双方が期限内に終了すること。途中write後の不完全frameを次requestへ継承しないこと。
### A04 — Codex auth Mutex を保持したまま deadline のない response body を待つ
**根拠**
- `crates/agen/src/providers/codex/mod.rs:78, 98–139`: default HTTP clientを使い、auth state Mutexをload・refresh・persist全体で保持。
- `crates/agen/src/providers/codex/refresh.rs:46–63`: 30秒timeoutは `send()` まで。その後の `json().await` / `text().await` は対象外。
- `crates/agen/src/llm_client/transport.rs:545, 648–650`: dynamic header解決はLLM HTTP送信timeoutより前。
**影響**: refresh先がheaderだけ返しbodyを完了しないと、同じproviderを共有する認証要求がすべてMutex待ちになる。明示的キャンセルは別として、自動timeout/retryでは復旧しない。
**改善**: response body消費まで含むrefresh全体のdeadlineとbody size上限を置く。token rotationのsingle-flight性は残す。ロックを外して同じrefresh tokenを同時利用する修正は不適切。
**回帰検証**: header送信後にbodyを停止するHTTP stubで、refreshと後続callerが有限時間で戻り、次回要求を処理できること。
### A05 — Agen の fallback pre-tool / turn-end policy gate が Cancel/Pause を処理しない
**根拠**
- `crates/agen/src/engine.rs:1988–1997`: `execute_tools` のpre-tool interceptorを直接await。
- 同 `2089–2090`: その後provider taskを開始するまでに再度のCancel/Pause確認がない。
- 同 `2952–2960`: assistant-turn-end interceptorも直接await。
- 対照: 同 `1534–1571` のearly admissionはCancel/Pauseを `select!` で処理。
- `AfterResponse` とearly dispatch非対応時のfallbackで到達する(同 `2717–2737`)。
**影響**: policy futureがpendingの間、Cancel/Pauseはqueueに残る。policyが後からContinueを返すと、要求済みキャンセルを処理する前に新しいproviderを開始し得る。turn-end待ちではearly-start済みtoolの制御も遅れる。Worker標準hookには30秒のchain期限があるが、promptなキャンセルとは異なる。
**改善**: early admissionと同じ制御signal監視を適用し、provider所有権の開始直前にも確認する。turn-end中断では開始済みbatchを明示的にterminalizeする。policyや永続化hookを無条件に並列化する話ではない。
**回帰検証**: AfterResponse/fallback双方でpre-toolをpendingにしCancel/Pauseを送り、toolが開始されないこと。turn-end待ちも同様にearly-start済みtoolが回収されること。
### A06 — External Workdir の output 待ちが cancel の admission まで塞ぐ
**根拠**
- `crates/workspace-server/src/server.rs:938–948, 969–982`: 全operationが同じsemaphoreを取得。timeout開始はpermit取得・enqueueの後。
- 同 `1091–1098`: waiting command outputは610秒のtimeout。
- 同 `1455–1463`: command cancelも同じoperation経路。
- 同 `23444–23452, 23532–23538`: connection-wide admissionとpending上限は16。
- `crates/workdir/src/local.rs:1930–1952`: wait=trueのoutputはcommand終了を待つ。
**影響**: 16件のoutput待ちで枠が埋まると、そのcommandを終わらせたいcancel自身が枠待ちになる。cancelの通常30秒timeoutはadmission待ちを制限しない。closeもcommand cancelを経由するため影響する。永久deadlockとは限らないが、長い待ちを自己増幅する。
**改善**: admissionとpending上限の両方でcontrol用capacityを確保し、permit/enqueue待ちを含む絶対deadlineを使う。単にcancelを無制限に通してprovider上限を破らない。
**回帰検証**: output待ちでdata枠を埋めてもcancel/closeが規定時間内に受理されること。provider全体の同時実行上限は維持されること。
### A07 — session cleanup の remove-before-await がキャンセル安全でない
**根拠**
- `crates/workspace-server/src/server.rs:515–568`: session/commandをregistryから取り出して削除。
- 同 `587–609, 618–638`: local Vecを順に `session.close().await` し、失敗項目はloop終了後だけ復元。
- 同 `17524–17539`: attachment単独cleanupも同型。
- 同 `1488–1518`: external sessionは明示的にprovider commandを終了させる必要がある。
**影響**: cleanup futureがpending中にdrop/abortされると、未処理項目・close中の項目・復元前の失敗項目を失う。retryがregistryから発見できず、external commandが動作したまま未追跡になる可能性がある。通常のHTTP切断だけで必ずhandlerがキャンセルされる、とまでは主張しない。
**改善**: unresolved項目をregistry内に保持するか、drop時に同期復元できるRAII所有権を用いる。あるいはrequest futureから独立したsupervised cleanupへ所有権を移す。成功確認後にだけ恒久削除する。
**回帰検証**: closeをbarrierで停止してcleanup futureをdropし、retryが全未完了項目を発見・closeできること。直列loopを並列化する前にこの所有権契約を直す。
### A08 — LocalWorkdirSession の async interface が同期 filesystem/search を直呼び
**根拠**
- `crates/workdir/src/local.rs:1471–1517, 1724–1828`: checkout search / read / write / edit / list / glob / grepが `fs_operation::run_*` を直接実行。
- `crates/worker-runtime/src/http_server.rs:880–924`: Runtime HTTP handlerがsession dispatcherを直接awaitし、ここにはblocking executorへの移動がない。
**影響**: 大きなtreeのgrep、低速mount、file I/Oがexecutorを占有する。同じthread上のtimer・control・別requestはその間進めない。async signatureもFuturesUnorderedも、poll内の同期処理を非同期化しない。
**改善**: provider内部でownedなrequest、session/resource guardをblocking側へ移し、同時実行数を制限する。scope検証・観測条件・mutationの順序を維持する。outer futureが消えても実I/Oが残るため、resource所有権をwaiterだけに持たせない。
**回帰検証**: 制御可能な遅いfilesystem adapterで処理を止め、同一executor上のtimerと別sessionのcontrolが進むこと。closeとblocking I/Oのresource lifetimeも検証。
### A09 — Bash output pump は select 内でも協調的に戻らない
**根拠**
- `crates/workdir/src/local.rs:2256–2298`: timer/cancel/child waitとoutput tickをselect。
- 同 `2371–2400`: output tick内の `publish_available_output` が同期readを、EOFまたは短いreadまで繰り返す。chunkは有限だが、一回の呼び出しで処理する総量は無制限。
**影響**: stdoutがread/telemetry処理より速く増え続けるとdrainが戻らず、stderr・timeout・cancelをpollしない。有限でも大量backlogで長い遅延になる。`select!`で囲っただけではbranch内の長い同期loopをプリエンプトできない。
**改善**: 1回のpumpにbyte/chunk budgetを設け、制御loopへ戻る。disk readはblocking境界または専用pumpへ移す。stdout/stderrの公平性と最終drainの完了条件を分ける。
**回帰検証**: 常にfull chunkを返す制御可能なreaderで、1 tickが有限量で終了すること。大量出力中もtimeout/cancelが規定範囲で処理されること。
### A10 — Workdir 共通 try_lock が独立した読み取りを失敗させる
**根拠**
- `crates/workdir/src/local.rs:650`: session内の `checkout_lock: StdMutex<()>`。
- 同 `1477–1513, 1526–1544, 1563–1567`: search/observe/execute全体で排他。
- 同 `1734–1751`:通常readも同じ排他 `try_lock`。競合時は `filesystem provider busy` を返す。
**影響**: 同じsessionの異なるファイルのReadやread-only searchまで排他になる。複数executor threadから重なると待つのではなく一方が失敗する。単一threadでは同期I/Oのため事実上直列に見えるが、A08を修正して並列性を出すだけではbusyエラーが増える。
**改善**: 何を守る排他かを分ける。観測済み対象のread-only処理を共有許可し、mutationのvalidator確認+更新は必要な単位で排他。same-target競合、親directory変更、scope/checkout lifecycleは維持する。単にMutexを消すのではなく、read gateとmutation gateを設計する。
**回帰検証**: 同じsessionで別ファイルを同時readして両方成功すること。writeとの競合では古い観測に基づくmutationを引き続き拒否すること。
### A11 — Workspace API cache の全体ロックで初期化し、Runtime attach も直列
**根拠**
- `crates/workspace-server/src/server.rs:2224–2238`: `apis` 全体Mutexを保持して `WorkspaceApi::new(...).await`。
- 同 `3516–3541`: config/storage初期化とRuntime projection attachを含む。
- 同 `3525–3529`: Runtimeをfor loopで逐次attach。
- `crates/workspace-server/src/worker_projection.rs:131–146`: 全Runtime共通のattach_gateを保持し、初回snapshotを最大2秒待つ。
**影響**: 一つのWorkspaceの初期化が、別Workspaceのcache hitや作成・削除準備まで待たせる。snapshotが来ないRuntimeがN件なら、初期化のsnapshot待ちが概ね2秒×Nに積み上がり得る。cached routerへの通常dispatchはapi cacheを通らない早期returnがあるため、「全APIが止まる」とはしない。
**改善**: map lockをlookup/publicationだけに短縮し、Workspace単位のsingle-flightを外で行う。publish時には削除/lifecycle状態を再確認。Runtime attachは同一Runtimeの置換だけを直列化し、異なるRuntimeはbounded concurrentにする。callerをjoin_allにするだけでは全体attach_gateで再直列化される点に注意。projection commit順序のgateは維持する。
**回帰検証**: Workspace Aの初期化を止めてもBのcache hitが戻ること。複数Runtimeの初期snapshot待ちが重なること。同一Runtime置換で旧consumerのeventが混入しないこと。
### A12 — FeatureStorage の管理 Mutex に disk/SQLite lifecycle 処理を含めすぎる
**根拠**
- `crates/feature-storage/src/lib.rs:258–290`: manager state Mutexを保持したままdirectory作成・fsync、SQLite open/configure/migrate。
- 同 `719–742`: configured/resolved ancestorへのdirectory sync。
- 同 `750–795`: WAL設定・busy retry・IMMEDIATE migration。
- 同 `293–334`: scope backupもmanager lockを全処理中保持。
**影響**: 一つのDB open/migration/backupが、同じmanagerの別Featureのcache hitやopen/lifecycle操作まで塞ぐ。接続処理の遅さをregistry全体の待ちへ広げている。BackendはWorkspaceApiごとにmanagerを構築しているため、現行Backendのこのlock単体が全Workspaceをまたぐとは主張しない。backupの本番利用頻度は未測定。
**改善**: scope/Feature単位のinitializing状態とsingle-flightを使い、global管理lock外でI/Oする。open中のdelete/shutdownを防ぐreservation・fenceを設け、publish前に確認する。backupのmaintenance exclusionも対象scope単位に限定する。
**回帰検証**: Feature Aのopenを止めても既存Feature Bの取得が戻ること。open途中のdelete/shutdownでconnectionやdirectoryを再生しないこと。
### A13 — paste 永続化の同期 lock/fsync が async Worker を止める
**根拠**
- `crates/worker/src/worker.rs:4793–4813`: async runの中で `materialize_large_pastes` を同期呼び出し。
- 同 `5449–5454`: Storeのpaste artifact書き込みへ進む。
- `crates/session-store/src/fs_store.rs:425–435`: append Mutexを保持して保存。
- `crates/session-store/src/paste_artifact.rs:83–91, 136–149`: blocking filesystem lock、aggregate使用量scan、write/fsync/rename/directory fsync。
- `crates/standalone/src/host.rs:681–690`:Standaloneの実StoreはこのFsStore。
**影響**: 遅いstorageやaggregate lock競合で、Workerをpollするexecutorが停止する。特にcurrent-thread runtimeではtimer/controlまで止まる。複数threadでもexecutor枯渇要因。
**改善**: ownedな保存jobをbounded blocking実行または専用persistence serviceへ移し、完了ack後にinputをpublishする。aggregate quota、append順序、durable commit-before-publicationは正しい要件であり削除しない。
**回帰検証**: aggregate lockを別threadで保持し、保存待ち中にtimer/controlが進むこと。ack前にはinput commitを公開しないこと。
### A14 — PDF extraction はキャンセル後も blocking task が走り続ける
**根拠**
- `crates/tools/src/web.rs:825–834`: `spawn_blocking(pdf_extract...)` のJoinHandleをその場でawait。
- `crates/agen/src/tool.rs:474–491, 593–604`: force-closeはouter async taskをabortし、default cooperative cancellationは何もしない。
**影響**: parser開始後にouter futureをdropしてもblocking closureは止まらない。toolがOutcomeUnknownで終わってもCPU/memory消費は続く。繰り返しキャンセルで処理が残り得る。入力byte上限とHTTP timeoutはparser実行時間の上限ではない。
**改善**: 実blocking処理が保持するconcurrency permitとprovider側のtask所有権を用意する。厳密な時間・停止保証が必要ならkill可能なprocessか協調キャンセル可能なparserを使う。JoinHandleにtimeoutを付けるだけでは実処理を停止できない。
**回帰検証**: parserをbarrierで停止してouter toolをキャンセルし、残存処理が同時実行枠を保持していること、無制限に後続parserを増やさないことを確認。
### A15 — Backend TUI の event loop が同期pollと操作待ちを抱える
**根拠**
- `crates/tui/src/backend_dashboard.rs:84–105`: async run_loop内で `event::poll(100ms)` / `event::read`。idle時はawaitせずloop継続。
- 同 `148–186`: reload、Intake/Orchestrator起動を入力branch内で完了までawait。
- 対照: `crates/tui/src/console/mod.rs:673–717` は専用terminal reader threadを持つ。
**影響**: idle中もasync taskが同期pollを繰り返し、pollからreturnしない。さらにHTTPをspawn_blockingへ移していても、UI loopがその完了を待つため、その間q/Esc、navigation、再描画を処理できない。I/OのoffloadとUI応答性は別問題。
**改善**: consoleと同様のevent readerを共有し、入力・結果・tickをselectする。ネットワーク操作は所有権を追跡したtaskとして走らせ、UIはpending表示と結果message処理を行う。キャンセルでmutation結果が不明な場合は失敗扱いで再送しない。
**回帰検証**: Backend応答を停止してreload/launchし、q/Escと再描画が引き続き動くこと。二重起動を許さないこと。
### A16 — 独立した画面データ取得が不要に直列
**根拠**
1. `web/workspace/src/routes/w/[workspaceId]/settings/runtimes/+page.ts:7–31`: Runtime listの完了後にsigning identity取得を開始。どちらも入力はworkspaceIdだけ。
2. `crates/tui/src/backend_dashboard.rs:62–70`: Ticket一覧とObjective一覧を一つのblocking closure内で順番に取得。
3. `crates/tui/src/backend_worker_picker.rs:96–116`: active/stopped一覧を順番に取得。include_stopped時は独立に開始できる。
**影響**: 依存関係がないのに待ち時間が各requestの和になる。実測値ではないが、二つの独立待ちの理想的critical pathは和ではなく最大値。
**改善**: WebはPromise.all、TUIは二つのtaskを同時開始してjoin、またはasync clientを直接使用。現在の部分失敗diagnosticsと表示順は維持する。単にjoinするだけで同じblocking closure内を並列にできるわけではない。
**回帰検証**: 一方のresponseを止めても他方が開始済みであることをbarrierで検証。失敗時の個別error表示も維持。
## 問題と数えなかったもの
- 短い共有state/registry更新の `std::Mutex`。awaitを跨がず短時間なら妥当であり、async Mutexへの一律置換は不要。
- `crates/agen/src/tool_server.rs:123–132`: toolをcloneしregistry lock解放後に実行する構造。
- 同一対象のmutation、SQLite transaction、durable commitとpublication順序の直列性。
- MCPのstdin frame混線防止、Codex token refreshのsingle-flightそのもの。
- `workspace_drive_host.rs:77–98` のlazy-open排他。初期化重複を防ぐ目的があり、今回の指摘はそれ自体ではない。
- Web dashboardは `web/workspace/src/lib/workspace/home/dashboard.ts:116–125, 165–206, 225` で独立requestを開始し、HOME_LIMIT=5のdetailをPromise.allで取得している。for/awaitの存在だけで直列化と判定していない。
- Workdir lifecycleや権限境界の排他は、単純な性能改善のために除去しない。
## 推奨する修正順と設計上の注意
1. **制御の進行保証**: A01/A03/A04/A05。満杯queue、partial body、policy pendingを使う決定的テストを先に用意する。
2. **executor上のblocking解消**: A02/A08/A09/A13。boundedな実行数、実処理側resource所有権、commit ackをセットで導入。無制限spawn_blockingへの置換はしない。
3. **キャンセル・cleanup所有権**: A06/A07/A14。control用capacityと実task lifetimeを明示する。
4. **排他粒度と独立処理の並列化**: A10/A11/A12/A15/A16。まず保護対象を明確にし、read-only/別Runtime/別Workspaceを分ける。
A08とA10は同時に扱う必要がある。同期I/Oをoffloadして並列に動くようにすると、現在のtry_lockによるbusy拒否が顕在化しやすくなる。またA11はcallerのjoin_allだけではattach_gateで再直列化される。
性能改善の測定は、mutex取得待ち/保持時間、DB job queue待ち、blocking job稼働数、control dispatch遅延、初期化critical pathを分けて行う。高負荷時のp95/p99とキャンセル完了時間を観測し、根拠なくmutex数の減少を成果指標にしない。
@@ -19,6 +19,7 @@ type State = {
params?: {
input?: import("../../../web/workspace/src/lib/generated/protocol.ts").Segment[];
submission_request_id?: string;
entry_id?: string;
};
}[];
events: { event: string }[];
@@ -181,6 +182,67 @@ async function attach(page: Page, name: string): Promise<void> {
buffer: Buffer.from("Synthetic fixture document"),
});
}
Deno.test("production Composer stays editable before connection and during admission without losing history focus or later edits", () =>
fixture(async (page, base, errors) => {
await page.evaluate(() => localStorage.setItem(
"yoi.composer-history.v1.workspace.console-live-review",
JSON.stringify({ version: 1, entries: [{ segments: [{ kind: "text", content: "previous input" }], preserveExactText: false }] }),
));
let release!: () => void;
const waiting = new Promise<void>((resolve) => { release = resolve; });
await page.route("**/workers/worker-a/session", async (route) => {
await waiting;
await route.continue();
});
try {
await page.reload({ waitUntil: "domcontentloaded" });
await page.locator(".cm-content[contenteditable=true]").waitFor();
await draft(page, "offline draft");
const editor = page.locator(".cm-content");
const assertFocused = async () => assert(await editor.evaluate((node) =>
node === document.activeElement && node.getAttribute("contenteditable") === "true"
));
await page.keyboard.press("Control+Enter");
assertEquals((await state(base)).methods.filter((method) => method.method === "submit"), []);
assert(await page.locator(".composer-send-button").isDisabled());
await page.keyboard.press("ArrowUp");
assertEquals(await editor.textContent(), "previous input");
await assertFocused();
await page.keyboard.press("ArrowDown");
assertEquals(await editor.textContent(), "offline draft");
await assertFocused();
await page.keyboard.type(" edited");
assertEquals(await editor.textContent(), "offline draft edited");
release();
await page.getByText("Ready. Choose a Feature", { exact: false }).waitFor();
await page.waitForFunction(() => !(document.querySelector(".composer-send-button") as HTMLButtonElement)?.disabled);
await assertFocused();
await control(base, { submit: "hold" });
await page.keyboard.press("Control+Enter");
await eventually(base, (value) => value.methods.filter((method) => method.method === "submit").length === 1);
assert(await page.locator(".composer-send-button").isDisabled());
await assertFocused();
await page.keyboard.type(" while pending");
await page.keyboard.press("Control+Enter");
assertEquals((await state(base)).methods.filter((method) => method.method === "submit").length, 1);
await control(base, { accept_pending: true });
await page.getByRole("button", { name: "Stop Worker", exact: true }).waitFor();
assertEquals(await editor.textContent(), "offline draft edited while pending");
await assertFocused();
await page.keyboard.type(" while running");
assertEquals(await editor.textContent(), "offline draft edited while pending while running");
await page.keyboard.press("ArrowUp");
assertEquals(await editor.textContent(), "offline draft edited");
await assertFocused();
await page.keyboard.press("ArrowDown");
assertEquals(await editor.textContent(), "offline draft edited while pending while running");
await assertFocused();
assertEquals(errors, []);
} finally {
release();
}
}));
Deno.test("production leading slash remains responsive after loop fix", () =>
fixture(async (page, base, errors, caseOutput) => {
await draft(page, "/");
@@ -441,3 +503,50 @@ Deno.test("production selected invocation preserves multiple positional typed ar
JSON.stringify({ status: "pass", fixture: await state(base), errors }, null, 2),
);
}));
Deno.test("production rewind selects transcript rows locally and sends only entry_id on Enter", () =>
fixture(async (page, base, errors, caseOutput) => {
await control(base, { seed_rewind_history: true });
await page.reload();
await page.getByText("Ready. Choose a Feature", { exact: false }).waitFor();
await draft(page, ":rewind");
await page.keyboard.press("Enter");
const selected = page.locator(".rewind-selected");
await selected.waitFor();
assert((await selected.innerText()).includes("Review the release notes"));
assertEquals((await state(base)).methods, [], "selection must not request a candidate list");
assertEquals(await page.locator(".rewind-targets").count(), 0);
await page.keyboard.press("ArrowUp");
assert((await selected.innerText()).includes("Earlier rewind input"));
await page.keyboard.press("ArrowDown");
assert((await selected.innerText()).includes("Review the release notes"));
assert(await page.locator(".cm-content").evaluate((node) => node === document.activeElement));
await page.screenshot({ path: join(caseOutput, "rewind-selection.png") });
await page.keyboard.press("Escape");
assertEquals(await selected.count(), 0);
assertEquals((await state(base)).methods, []);
await draft(page, ":rewind");
await page.keyboard.press("Enter");
await selected.waitFor();
await control(base, { rewind: "reject" });
await page.keyboard.press("Enter");
await page.getByText("Synthetic rewind rejection", { exact: true }).first().waitFor();
await selected.waitFor();
await control(base, { rewind: "hold" });
await page.keyboard.press("Enter");
await page.keyboard.press("Enter");
const sent = await eventually(base, (value) => value.methods.filter((method) => method.method === "rewind_to").length === 2);
assertEquals(sent.methods, [
{ method: "rewind_to", params: { entry_id: "user-1" } },
{ method: "rewind_to", params: { entry_id: "user-1" } },
]);
await control(base, { apply_rewind: true });
await page.waitForFunction(() => document.querySelector(".cm-content")?.textContent === "Review the release notes before sending.");
assertEquals(await selected.count(), 0);
assertEquals(await page.locator(".cm-content").getAttribute("contenteditable"), "true");
await page.keyboard.type(" edited");
assertEquals(await page.locator(".cm-content").textContent(), "Review the release notes before sending. edited");
assertEquals(errors, []);
await page.screenshot({ path: join(caseOutput, "rewind-applied.png") });
}, true));
@@ -144,6 +144,8 @@ const events: unknown[] = [];
let submitMode: "accept" | "reject" | "hold" = "accept";
let uploadMode: "normal" | "fail" | "hold" = "normal";
const pendingAcceptances: (() => void)[] = [];
let rewindMode: "accept" | "reject" | "hold" = "accept";
const pendingRewinds: (() => void)[] = [];
const cancelledUploads = new Set<string>();
const heldUploads = new Map<string, (response: Response) => void>();
const releaseHeldUploads = new Map<string, () => void>();
@@ -197,6 +199,12 @@ Deno.serve({ hostname: "127.0.0.1", port }, async (request) => {
const control = await request.json();
if (["accept", "reject", "hold"].includes(control.submit)) submitMode = control.submit;
if (["normal", "fail", "hold"].includes(control.upload)) uploadMode = control.upload;
if (["accept", "reject", "hold"].includes(control.rewind)) rewindMode = control.rewind;
if (control.seed_rewind_history) {
entries.unshift({ ...entries[0], entry_id: "user-older", content: [{ kind: "text", text: "Earlier rewind input" }] },
{ ...entries[1], entry_id: "assistant-older", content: [{ kind: "text", text: "Earlier reply" }] });
}
if (control.apply_rewind) { for (const apply of pendingRewinds.splice(0)) apply(); }
if (control.accept_pending) { for (const accept of pendingAcceptances.splice(0)) accept(); }
if (control.release_uploads) {
for (const release of [...releaseHeldUploads.values()]) release();
@@ -365,6 +373,21 @@ Deno.serve({ hostname: "127.0.0.1", port }, async (request) => {
methods.push(method);
const subscription_id = frame.message.subscription_id;
if (subscriptions.get(subscription_id) !== "worker_protocol") return;
const emit = (event: unknown) => {
events.push(event);
if (socket.readyState !== WebSocket.OPEN) return;
send({
protocol_version: 2,
frame: "event",
message: {
event: "event",
data: {
subscription_id,
payload: { event: "worker_protocol", data: { worker_id: workerId, event } },
},
},
});
};
if (method.method === "list_completions") {
const { kind, prefix, request_id, context } = method.params;
let candidates: unknown[] = [];
@@ -424,24 +447,24 @@ Deno.serve({ hostname: "127.0.0.1", port }, async (request) => {
},
},
});
} else if (method.method === "rewind_to") {
const index = entries.findIndex((entry) => entry.entry_id === method.params.entry_id && entry.role === "user");
if (rewindMode === "reject" || index < 0) {
emit({ event: "error", data: { code: "invalid_request", message: "Synthetic rewind rejection" } });
} else {
const apply = () => {
const input = entries[index].content.map((part) => ({ kind: "text", content: part.text }));
const discarded_entries = entries.length - index;
entries.splice(index);
emit({ event: "rewind_applied", data: { session, input, summary: {
truncated_to_entries: index, discarded_entries, tool_side_effect_warning: true,
} } });
};
if (rewindMode === "hold") pendingRewinds.push(apply); else apply();
}
} else if (method.method === "submit") {
const emit = (event: unknown) => {
events.push(event);
if (socket.readyState !== WebSocket.OPEN) return;
send({
protocol_version: 2,
frame: "event",
message: {
event: "event",
data: {
subscription_id,
payload: { event: "worker_protocol", data: { worker_id: workerId, event } },
},
},
});
};
const submission_request_id = method.params.submission_request_id;
const accept = () =>
const accept = () => {
emit({
event: "submission_accepted",
data: {
@@ -450,6 +473,20 @@ Deno.serve({ hostname: "127.0.0.1", port }, async (request) => {
disposition: "started",
},
});
// A held admission models an in-flight run; immediate fixture runs
// finish synchronously. Publish authoritative state, not only an ack.
emit({
event: "worker_state",
data: {
snapshot: {
last_command_id: 0,
state: submitMode === "hold"
? { kind: "busy", state: { kind: "run", state: "running" } }
: { kind: "idle" },
},
},
});
};
if (submitMode === "reject") {
emit({
event: "submission_rejected",
File diff suppressed because one or more lines are too long
+2 -6
View File
@@ -155,10 +155,6 @@ symlink_policy: SymlinkPolicy, };
export type CompletionEntry = { value: string, is_dir: boolean, description?: string | null, usage?: string | null, invocation?: FeatureInvocationDescriptor | null, };
export type RewindTargetId = { segment_id: string, user_input_entry_index: number, };
export type RewindTarget = { id: RewindTargetId, expected_head_entries: number, truncate_entries: number, turn_index: number, timestamp_ms: number | null, preview: string, eligible: boolean, disabled_reason: string | null, warning: string | null, };
export type RewindSummary = { truncated_to_entries: number, discarded_entries: number, tool_side_effect_warning: boolean, };
export type InFlightBlock = { "kind": "text", text: string, finished?: boolean, } | { "kind": "thinking", text: string, finished?: boolean, } | { "kind": "tool_call", id: string, name: string, args: string, state?: InFlightToolCallState, };
@@ -416,7 +412,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<Segment>, } } | { "method": "notify", "params": { notification_request_id: string, message: string, } } | { "method": "worker_event", "params": WorkerEvent } | { "method": "list_pending_submissions" } | { "method": "cancel_pending_submission", "params": { submission_id: string, } } | { "method": "clear_pending_submissions", "params": { expected_submission_ids: Array<string>, } } | { "method": "continue_pending", "params": { expected_head_id: string, } } | { "method": "resume", "params": { command: WorkerCommandEnvelope, } } | { "method": "cancel", "params": { command: WorkerCommandEnvelope, } } | { "method": "pause", "params": { command: WorkerCommandEnvelope, } } | { "method": "compact", "params": { command: WorkerCommandEnvelope, } } | { "method": "list_rewind_targets" } | { "method": "rewind_to", "params": { target: RewindTargetId, expected_head_entries: number, } } | { "method": "shutdown", "params": { command: WorkerCommandEnvelope, } } | { "method": "list_completions", "params": { kind: CompletionKind, prefix: string,
export type Method = { "method": "submit", "params": { submission_request_id: string, input: Array<Segment>, } } | { "method": "notify", "params": { notification_request_id: string, message: string, } } | { "method": "worker_event", "params": WorkerEvent } | { "method": "list_pending_submissions" } | { "method": "cancel_pending_submission", "params": { submission_id: string, } } | { "method": "clear_pending_submissions", "params": { expected_submission_ids: Array<string>, } } | { "method": "continue_pending", "params": { expected_head_id: string, } } | { "method": "resume", "params": { command: WorkerCommandEnvelope, } } | { "method": "cancel", "params": { command: WorkerCommandEnvelope, } } | { "method": "pause", "params": { command: WorkerCommandEnvelope, } } | { "method": "compact", "params": { command: WorkerCommandEnvelope, } } | { "method": "rewind_to", "params": { entry_id: string, } } | { "method": "shutdown", "params": { command: WorkerCommandEnvelope, } } | { "method": "list_completions", "params": { kind: CompletionKind, prefix: string,
/**
* Client query identity. Echoed verbatim so identical-prefix ABA
* responses cannot cross edits, target switches, or permission changes.
@@ -468,4 +464,4 @@ in_flight?: InFlightSnapshot,
* Parent-owned Internal Worker sessions visible to this client.
* Service-private Internal Workers are deliberately excluded.
*/
internal_workers?: Array<InternalWorkerSnapshot>, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, event: Event, } } | { "event": "internal_worker_removed", "data": { worker: InternalWorkerRef, } } | { "event": "segment_rotated", "data": { session: SessionSnapshot, } } | { "event": "worker_state", "data": { snapshot: WorkerStateSnapshot, } } | { "event": "command_acknowledged", "data": { acknowledgement: WorkerCommandAcknowledgement, } } | { "event": "command", "data": { event: CommandEvent, } } | { "event": "completions", "data": { kind: CompletionKind, prefix: string, request_id?: string | null, context?: CompletionContext | null, entries: Array<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { session: SessionSnapshot, input: Array<Segment>, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compaction_progress", "data": { compaction: InFlightCompaction | null, } } | { "event": "compact_start", "data": { lifecycle: CompactionLifecycle, } } | { "event": "compact_done", "data": { lifecycle: CompactionLifecycle, } } | { "event": "compact_failed", "data": { lifecycle: CompactionLifecycle, } } | { "event": "shutdown" };
internal_workers?: Array<InternalWorkerSnapshot>, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, event: Event, } } | { "event": "internal_worker_removed", "data": { worker: InternalWorkerRef, } } | { "event": "segment_rotated", "data": { session: SessionSnapshot, } } | { "event": "worker_state", "data": { snapshot: WorkerStateSnapshot, } } | { "event": "command_acknowledged", "data": { acknowledgement: WorkerCommandAcknowledgement, } } | { "event": "command", "data": { event: CommandEvent, } } | { "event": "completions", "data": { kind: CompletionKind, prefix: string, request_id?: string | null, context?: CompletionContext | null, entries: Array<CompletionEntry>, } } | { "event": "rewind_applied", "data": { session: SessionSnapshot, input: Array<Segment>, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compaction_progress", "data": { compaction: InFlightCompaction | null, } } | { "event": "compact_start", "data": { lifecycle: CompactionLifecycle, } } | { "event": "compact_done", "data": { lifecycle: CompactionLifecycle, } } | { "event": "compact_failed", "data": { lifecycle: CompactionLifecycle, } } | { "event": "shutdown" };
@@ -27,6 +27,42 @@ function deferred<T>() {
return { promise, resolve };
}
test("history, clear and restore keep editing and focus without stealing focus from elsewhere", async () => {
const ui = render(ComposerInput, { historyScope: "always-editable" });
const cm = editor(ui.container);
type(cm, "previous input");
ui.component.recordHistory(ui.component.snapshot());
ui.component.clear();
type(cm, "draft");
const editable = () => {
expect(cm.state.readOnly).toBe(false);
expect(cm.contentDOM.getAttribute("contenteditable")).toBe("true");
expect(cm.hasFocus).toBe(true);
};
await fireEvent.keyDown(cm.contentDOM, { key: "ArrowUp" });
expect(cm.state.doc.toString()).toBe("previous input");
editable();
await fireEvent.keyDown(cm.contentDOM, { key: "ArrowDown" });
expect(cm.state.doc.toString()).toBe("draft");
editable();
await fireEvent.paste(cm.contentDOM, { clipboardData: { getData: () => " pasted" } });
expect(cm.state.doc.toString()).toBe("draft pasted");
expect(undo(cm)).toBe(true);
expect(cm.state.doc.toString()).toBe("draft");
expect(redo(cm)).toBe(true);
expect(cm.state.doc.toString()).toBe("draft pasted");
editable();
ui.component.clear();
editable();
ui.component.restoreSegments([{ kind: "text", content: "restored" }]);
editable();
ui.component.replaceRange(8, 8, " edit");
expect(cm.state.doc.toString()).toBe("restored edit");
cm.contentDOM.blur();
ui.component.restoreSegments([{ kind: "text", content: "background restore" }]);
expect(cm.hasFocus).toBe(false);
});
test("commands auto-open with descriptions; ambiguous Tab does not choose the first; arrows beat history", async () => {
const oncommand = vi.fn();
const onsubmit = vi.fn();
@@ -122,7 +158,7 @@ test("file popup scrolls selection, drills into directories and Enter accepts wi
expect(onsubmit).not.toHaveBeenCalled();
});
test("cursor changes, Escape, scope changes and disable discard late file results", async () => {
test("cursor changes, Escape and scope changes discard late file results", async () => {
const old = deferred<Array<{ value: string }>>();
const resolveFileCompletions = vi.fn(() => old.promise);
const ui = render(ComposerInput, {
@@ -145,10 +181,7 @@ test("cursor changes, Escape, scope changes and disable discard late file result
await ui.findByRole("listbox");
await ui.rerender({ completionScope: "b" });
expect(ui.queryByRole("listbox")).toBeNull();
type(cm, ":rew");
await ui.findByRole("listbox");
await ui.rerender({ disabled: true });
expect(ui.queryByRole("listbox")).toBeNull();
});
test("synchronous draft and cursor ABA invalidate old Feature completion generations", async () => {
@@ -1,6 +1,5 @@
<script lang="ts">
import {
Compartment,
EditorSelection,
EditorState,
Prec,
@@ -62,7 +61,6 @@
} from "#lib/workspace/console/composer-invocation.ts";
interface Props {
disabled?: boolean;
ariaLabel?: string;
ariaKeyShortcuts?: string;
historyScope: string;
@@ -85,7 +83,6 @@
}
let {
disabled = false,
ariaLabel = "Message",
ariaKeyShortcuts = "Meta+Enter Control+Enter",
historyScope,
@@ -137,7 +134,6 @@
return retained;
},
});
const editable = new Compartment();
const completionId = $props.id();
let completionToken = $state<ComposerCompletionToken | null>(null);
@@ -151,7 +147,7 @@
const completionStart = $derived(Math.max(0, (completionSelected ?? 0) - 5));
const visibleCompletions = $derived(completionEntries.slice(completionStart, completionStart + 6));
function closeCompletion(): void {
export function closeCompletion(): void {
completionVersion++;
completionAbort?.abort();
completionAbort = null;
@@ -170,7 +166,7 @@
if (!force && context === completionContext) return;
completionContext = context;
closeCompletion();
if (disabled || !view.hasFocus || view.composing || !selection.empty) return;
if (!view.hasFocus || view.composing || !selection.empty) return;
const token = completionTokenAt(
document,
selection.head,
@@ -213,7 +209,7 @@
}
function acceptCompletion(action: "tab" | "accept", index = completionSelected): void {
if (!view || !completionToken || disabled) return;
if (!view || !completionToken) return;
const token = completionToken;
const document = view.state.doc.toString();
if (token.kind === "command" && index === null) {
@@ -256,7 +252,7 @@
}
function completionKeydown(event: KeyboardEvent): boolean {
if (disabled || event.isComposing || view?.composing || event.keyCode === 229) return false;
if (event.isComposing || view?.composing || event.keyCode === 229) return false;
if (event.ctrlKey || event.metaKey || event.altKey || event.shiftKey) return false;
if (event.key === "Escape" && completionToken) {
event.preventDefault(); closeCompletion(); return true;
@@ -299,12 +295,6 @@
completionContext = "";
});
$effect(() => {
void disabled;
closeCompletion();
completionContext = "";
});
const registerPaste = StateEffect.define<{ key: number; paste: ComposerPaste }>();
const pasteRegistry = StateField.define<ReadonlyMap<number, ComposerPaste>>({
create: () => new Map(),
@@ -503,7 +493,7 @@
segment: Segment,
options: { from?: number; to?: number; cleanup?: boolean; label?: string; uploadReservation?: boolean } = {},
): number | null {
if (!view || view.state.readOnly) return null;
if (!view) return null;
const selection = view.state.selection.main;
const from = options.from ?? selection.from;
const to = options.to ?? selection.to;
@@ -525,7 +515,7 @@
}
function chipifySelectedInvocation(strict = false): void {
if (!view || view.composing || view.state.readOnly) return;
if (!view || view.composing) return;
// Only completion-selected or explicitly edited calls become structured input.
// Descriptor discovery must never promote unrelated typed or pasted prose.
for (const selected of [...view.state.field(selectedInvocationState)].reverse()) {
@@ -553,7 +543,7 @@
let editVersion = 0;
let editorGeneration = 0;
async function editInvocation(key: number): Promise<void> {
if (!view || disabled) return;
if (!view) return;
const scope = completionScope;
const version = ++editVersion;
const document = view.state.doc.toString();
@@ -570,7 +560,7 @@
descriptor = entries.find((entry) => entry.invocation?.identity === invocation.identity)?.invocation ?? undefined;
} catch { return; }
}
if (!descriptor || !view || disabled || version !== editVersion || scope !== completionScope ||
if (!descriptor || !view || version !== editVersion || scope !== completionScope ||
generation !== editorGeneration || document !== view.state.doc.toString() ||
selection.from !== view.state.selection.main.from || selection.to !== view.state.selection.main.to) return;
atom = composerTypedAtoms(view.state.doc.toString(), view.state.field(typedRegistry))
@@ -622,7 +612,6 @@
cursorLine,
lineCount: currentView.state.doc.lines,
selectionEmpty: selection.empty,
readOnly: currentView.state.readOnly,
composing: currentView.composing,
})) return false;
@@ -679,7 +668,6 @@
}
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;
@@ -718,7 +706,6 @@
currentView: EditorView,
direction: "backward" | "forward",
): boolean {
if (currentView.state.readOnly) return false;
const selection = currentView.state.selection.main;
const atoms = [
...composerPasteAtoms(
@@ -751,20 +738,11 @@
state: EditorState.create({
extensions: editorExtensions = [
history(),
Prec.highest(EditorView.domEventHandlers({ keydown: completionKeydown })),
Prec.highest(EditorView.domEventHandlers({ keydown(event) {
onkeydown?.(event);
return event.defaultPrevented || completionKeydown(event);
} })),
Prec.highest(keymap.of([
{
key: "Mod-z",
run: (currentView) => currentView.state.readOnly,
},
{
key: "Mod-Shift-z",
run: (currentView) => currentView.state.readOnly,
},
{
key: "Mod-y",
run: (currentView) => currentView.state.readOnly,
},
{
key: "ArrowUp",
run: (currentView) => browseHistory(currentView, "older"),
@@ -791,10 +769,6 @@
: []),
pasteChips,
typedChips,
editable.of([
EditorView.editable.of(!disabled),
EditorState.readOnly.of(disabled),
]),
EditorState.allowMultipleSelections.of(false),
EditorView.lineWrapping,
EditorView.contentAttributes.of({
@@ -840,7 +814,7 @@
const target = event.target instanceof Element
? event.target.closest(".composer-typed-chip")
: null;
if (!target || currentView.state.readOnly) return false;
if (!target) return false;
let position: number;
try {
position = currentView.posAtDOM(target);
@@ -867,7 +841,6 @@
return true;
},
cut(event, currentView) {
if (disabled) return false;
const content = selectedClipboardContent(currentView.state);
if (content === null || !event.clipboardData) return false;
event.preventDefault();
@@ -881,13 +854,11 @@
return true;
},
keydown(event) {
onkeydown?.(event);
if (event.defaultPrevented) return true;
if (
shouldSubmitChatKey(event, {
mode: "mod-enter",
modKey: "auto",
enabled: !disabled,
enabled: true,
})
) {
event.preventDefault();
@@ -918,16 +889,6 @@
};
});
$effect(() => {
const isDisabled = disabled;
view?.dispatch({
effects: editable.reconfigure([
EditorView.editable.of(!isDisabled),
EditorState.readOnly.of(isDisabled),
]),
});
});
export function snapshot(): ComposerDraftSnapshot {
chipifySelectedInvocation(true);
return currentSnapshot();
@@ -957,7 +918,7 @@
}
export function replaceRange(from: number, to: number, content: string): void {
if (!view || view.state.readOnly) return;
if (!view) return;
view.dispatch({
changes: { from, to, insert: content },
selection: EditorSelection.cursor(from + content.length),
@@ -1007,9 +968,6 @@
closeCompletion();
completionContext = "";
view.setState(EditorState.create({ extensions: editorExtensions }));
view.dispatch({ effects: editable.reconfigure([
EditorView.editable.of(!disabled), EditorState.readOnly.of(disabled),
]) });
emitChange();
}
@@ -1070,7 +1028,7 @@
}
</script>
<div class="composer-input" class:disabled>
<div class="composer-input">
{#if completionToken}
<div class="completion-popup">
<div id={completionId} role="listbox" aria-label="Composer completions" aria-busy={completionBusy}>
@@ -1147,10 +1105,6 @@
}
.completion-hint { padding: var(--space-1) var(--space-2); color: var(--text-muted); }
.composer-input.disabled {
opacity: 0.56;
}
.composer-input :global(.cm-editor) {
min-height: 2.65rem;
max-height: 10rem;
@@ -7,10 +7,11 @@
type Props = {
item: ConsoleLine;
rewindSelected?: boolean;
attachmentUrl?: (attachment: SessionToolAttachment) => string | null;
};
let { item, attachmentUrl = () => null }: Props = $props();
let { item, rewindSelected = false, attachmentUrl = () => null }: Props = $props();
let detailOpen = $state(false);
let nowMs = $state(Date.now());
@@ -83,6 +84,8 @@
<li
class={`console-line ${lineClass(item)} ${toolClass(item)}`}
class:error-line={item.error}
class:rewind-selected={rewindSelected}
aria-current={rewindSelected ? "true" : undefined}
data-console-line-id={item.id}
>
{#if item.compaction}
@@ -165,6 +168,12 @@
</li>
<style>
.rewind-selected {
outline: 2px solid var(--accent);
outline-offset: -2px;
background: color-mix(in srgb, var(--accent) 12%, transparent);
}
.console-line {
min-width: 0;
padding: var(--space-1) 0;
@@ -17,7 +17,7 @@ Deno.test("compact and rewind remain explicit Composer commands", () => {
ok: true, request: { kind: "compact", content: "" }, notice: "compact requested",
});
assertEquals(buildComposerRequest(":rewind"), {
ok: true, request: { kind: "list_rewind_targets", content: "" }, notice: "rewind targets requested",
ok: true, action: "rewind",
});
});
@@ -5,7 +5,6 @@ export type WorkerConsoleInputKind =
| "user"
| "notify"
| "compact"
| "list_rewind_targets"
| "register_peer";
export type WorkerConsoleInputRequest = {
@@ -15,7 +14,7 @@ export type WorkerConsoleInputRequest = {
};
export type ComposerCommandResult =
| { ok: true; request?: WorkerConsoleInputRequest; notice?: string }
| { ok: true; request?: WorkerConsoleInputRequest; action?: "rewind"; notice?: string }
| { ok: false; message: string };
type CommandSpec = {
@@ -43,7 +42,7 @@ export const COMMANDS: Record<string, CommandSpec> = {
},
rewind: {
usage: ":rewind",
description: "Ask the Worker for rewind targets.",
description: "Select a user input in Console history with ↑/↓, then Enter to rewind.",
},
rollback: {
usage: ":rollback",
@@ -200,8 +199,7 @@ function buildColonCommand(commandLine: string): ComposerCommandResult {
}
return {
ok: true,
request: { kind: "list_rewind_targets", content: "" },
notice: "rewind targets requested",
action: "rewind",
};
case "peer":
if (argv.length !== 1) {
@@ -37,7 +37,7 @@ export class ComposerAdmissions {
const prior = this.get(target);
if (prior && prior.status !== "rejected") {
throw new Error(
"Resolve the in-flight admission before editing or sending a new draft.",
"Resolve the in-flight admission before sending a new draft.",
);
}
const unchanged = prior && prior.method.method === method.method &&
@@ -11,7 +11,6 @@ export type ComposerHistoryCursor = {
cursorLine: number;
lineCount: number;
selectionEmpty: boolean;
readOnly: boolean;
composing: boolean;
};
@@ -85,7 +84,7 @@ function sameEntry(
export function shouldBrowseComposerHistory(
cursor: ComposerHistoryCursor,
): boolean {
if (cursor.readOnly || cursor.composing || !cursor.selectionEmpty) {
if (cursor.composing || !cursor.selectionEmpty) {
return false;
}
return cursor.direction === "older"
@@ -1325,7 +1325,6 @@ export function applyProtocolEvent(
break;
case "memory_worker":
case "completions":
case "rewind_targets":
case "workers_listed":
case "worker_restored":
case "peer_registered":
@@ -913,7 +913,7 @@ Deno.test("Worker Console removes redundant chrome and uses shared alerts", asyn
!page.includes("margin-inline: calc(-1 * var(--space-6))") &&
page.includes("import { pushWorkspaceAlert }") &&
page.includes('title: "Worker control"') &&
page.includes('title: "Rewind targets"') &&
!page.includes('aria-label="Rewind targets"') &&
page.includes('pushWorkspaceAlert("error"') &&
!page.includes("controlNotice") &&
!page.includes("console-notice") &&
@@ -1050,13 +1050,13 @@ Deno.test("Worker Console paste chips preserve typed draft and target authority"
composerInput.includes(
"composerDeletionRange(selection, atoms, direction)",
) &&
composerInput.includes("EditorState.readOnly.of(isDisabled)") &&
composerInput.includes('key: "Mod-z"') &&
composerInput.includes("if (!view || view.state.readOnly) return") &&
composerInput.includes("if (currentView.state.readOnly) return false") &&
!composerInput.includes("readOnly") &&
!composerInput.includes("disabled") &&
!composerInput.includes("EditorView.editable") &&
composerInput.includes("...historyKeymap") &&
consolePage.includes("completionScope={activeComposerTargetKey}") &&
composerInput.includes("version !== completionVersion") &&
consolePage.includes("if (!composerEditable) return") &&
consolePage.includes("if (!canUploadAttachments) return") &&
composerInput.includes('chip.setAttribute("aria-label", label)') &&
composerInput.includes("preserveExactText = false") &&
consolePage.includes("const command = buildComposerSegmentsRequest(") &&
@@ -1083,20 +1083,18 @@ Deno.test("Worker Console routes image paste through the existing attachment upl
input.indexOf("function selectedClipboardContent"),
);
assert(
paste.indexOf("if (disabled || view?.state.readOnly)") <
paste.indexOf("handleComposerImagePaste(event, onpasteimages)") &&
paste.indexOf("handleComposerImagePaste(event, onpasteimages)") <
paste.indexOf("handleComposerImagePaste(event, onpasteimages)") <
paste.indexOf('getData("text/plain")') &&
consolePage.includes("onpasteimages={addPastedImages}") &&
consolePage.includes("addAttachmentFiles(files.map(namePastedImage))"),
"image paste must respect editability and take priority over text before using the attachment callback",
"image paste must take priority over text before using the attachment callback",
);
const attach = consolePage.slice(
consolePage.indexOf("function addAttachmentFiles"),
consolePage.indexOf("async function removeAttachment"),
);
assert(
attach.includes("if (!composerEditable) return") &&
attach.includes("if (!canUploadAttachments) return") &&
attach.includes("MAX_FILES_PER_SUBMISSION - attachments.length") &&
attach.includes(".slice(0, available)") &&
attach.includes("startAttachmentUpload(attachment)"),
@@ -1273,15 +1271,16 @@ Deno.test("Worker Console route resolves logical Worker authority before Runtime
'const workerRunning = $derived(workerState === "running");',
) &&
consolePage.includes(
'const composerEditable = $derived(protocolState === "open" && !sending);',
'const canSendComposerControl = $derived(protocolState === "open" && !sending && !rewindApplying);',
) &&
consolePage.includes('sendWorkerControl("cancel")') &&
consolePage.includes("lifecycleMethod(command)") &&
consolePage.includes("command_id: commandId") &&
!consolePage.includes("expected_execution_generation") &&
consolePage.includes("onsubmit={handleComposerSubmit}") &&
consolePage.includes("disabled={!composerEditable}") &&
consolePage.includes("class:stop={workerRunning}") &&
!consolePage.includes("disabled={!composerEditable}") &&
consolePage.includes("if (!canSendComposerControl) return;") &&
consolePage.includes("class:stop={workerRunning && !rewindEntryId}") &&
consolePage.includes('"Stop Worker"') &&
consolePage.includes("disabled={composerSubmitDisabled}") &&
!consolePage.includes("disabled={!inputReady || sending}") &&
@@ -44,7 +44,6 @@
Event as ProtocolEvent,
Method as ProtocolMethod,
PendingSubmissionsSnapshot,
RewindTarget,
Segment,
SessionHistoryPage,
SessionToolAttachment,
@@ -114,6 +113,7 @@
let workerError = $state<string | null>(untrack(() => data.workerError));
type ComposerInputHandle = {
snapshot(): ComposerDraftSnapshot;
closeCompletion(): void;
focus(): void;
containsTarget(target: EventTarget | null): boolean;
cursor(): number;
@@ -175,8 +175,9 @@
reportComposerError(error instanceof Error ? error.message : String(error));
}
}
let rewindTargets = $state<RewindTarget[]>([]);
let rewindHeadEntries = $state(0);
let rewindEntryId = $state<string | null>(null);
let rewindApplying = $state(false);
let rewindLoadingEarlier = false;
let protocolState = $state<"connecting" | "open" | "closed" | "error">(
"connecting",
);
@@ -304,6 +305,7 @@
const turnNavigationItems = $derived(
conversationTurnPreviewsFromLines(lines),
);
const rewindEntries = $derived(lines.filter((line) => line.kind === "user" && line.entryId));
const tasks = $derived(selectedConsoleProjection.tasks);
const diagnostics = $derived(
mergeDiagnostics(worker?.diagnostics ?? [], streamDiagnostics),
@@ -313,7 +315,8 @@
);
const workerRunning = $derived(workerState === "running");
const workerPaused = $derived(workerState === "paused");
const composerEditable = $derived(protocolState === "open" && !sending);
const canSendComposerControl = $derived(protocolState === "open" && !sending && !rewindApplying);
const canUploadAttachments = $derived(consoleTarget !== null);
const draftHasText = $derived(draft.content.trim().length > 0);
const draftHasAttachments = $derived(
attachments.length > 0 || draft.segments.some((segment) => segment.kind === "uploaded_file"),
@@ -323,7 +326,7 @@
delivery: "submit",
workerState,
protocolOpen: protocolState === "open",
sending,
sending: sending || rewindApplying,
hasText: draftHasText,
hasAttachments: draftHasAttachments,
}),
@@ -333,7 +336,7 @@
delivery: "queue",
workerState,
protocolOpen: protocolState === "open",
sending,
sending: sending || rewindApplying,
hasText: draftHasText,
hasAttachments: draftHasAttachments,
}),
@@ -343,14 +346,14 @@
delivery: "notify",
workerState,
protocolOpen: protocolState === "open",
sending,
sending: sending || rewindApplying,
hasText: draftHasText,
hasAttachments: draftHasAttachments,
}),
);
const canStopFromComposer = $derived(workerRunning && composerEditable);
const canStopFromComposer = $derived(workerRunning && canSendComposerControl);
const composerSubmitDisabled = $derived(
workerRunning ? !canStopFromComposer : !canSubmitDraft,
rewindEntryId ? !canSendComposerControl : workerRunning ? !canStopFromComposer : !canSubmitDraft,
);
async function getJson<T>(path: string): Promise<T> {
@@ -771,6 +774,8 @@
) {
handleProtocolCommandEvent(payload);
if (payload.event === "rewind_applied") {
rewindApplying = false;
rewindEntryId = null;
if (draft.segments.length === 0 && attachments.length === 0) {
composerInputElement?.restoreSegments(payload.data.input);
} else {
@@ -822,6 +827,7 @@
}
function handleComposerKeydown(event: KeyboardEvent) {
if (handleRewindKeydown(event)) return;
if (event.key === "PageUp" || event.key === "PageDown") {
event.preventDefault();
scrollConsoleByPage(event.key === "PageDown" ? 1 : -1);
@@ -932,17 +938,88 @@
sendWorkerControl(command);
}
function rewindTo(target: RewindTarget) {
sendControl(
{
method: "rewind_to",
params: {
target: target.id,
expected_head_entries: rewindHeadEntries,
},
},
`Rewind to ${target.preview || "target"}`,
);
async function startRewindSelection(): Promise<void> {
if (rewindApplying) return;
const targetKey = activeComposerTargetKey;
await selectConsoleWorkerView(null);
if (activeComposerTargetKey !== targetKey) return;
const target = rewindEntries.at(-1);
if (!target?.entryId) {
reportComposerError("No user input is loaded. Load earlier conversation to select a rewind point.");
return;
}
composerInputElement?.clear();
rewindEntryId = target.entryId;
autoFollowConsole = false;
await tick();
if (rewindEntryId !== target.entryId) return;
selectConsoleTurn(target.id);
autoFollowConsole = false;
composerInputElement?.focus();
}
async function moveRewindSelection(direction: -1 | 1): Promise<void> {
if (!rewindEntryId || rewindApplying || rewindLoadingEarlier) return;
const targetKey = activeComposerTargetKey;
const selected = rewindEntryId;
let index = rewindEntries.findIndex((entry) => entry.entryId === selected);
if (index === 0 && direction === -1 && selectedHistory.hasMore && selectedHistory.cursor && consoleTarget) {
rewindLoadingEarlier = true;
try {
await requestHistoryPage(consoleTarget, selectedHistoryKey, selectedHistory.cursor, reloadToken);
await tick();
} finally { rewindLoadingEarlier = false; }
if (activeComposerTargetKey !== targetKey || rewindEntryId !== selected) return;
index = rewindEntries.findIndex((entry) => entry.entryId === selected);
}
if (index < 0) { rewindEntryId = null; return; }
const target = rewindEntries[Math.max(0, Math.min(rewindEntries.length - 1, index + direction))];
rewindEntryId = target.entryId!;
await tick();
if (rewindEntryId !== target.entryId) return;
selectConsoleTurn(target.id);
autoFollowConsole = false;
}
function applySelectedRewind(): void {
if (!rewindEntryId || rewindApplying) return;
if (!canSendComposerControl) {
reportComposerError("Cannot send rewind while disconnected or another input is awaiting admission.");
return;
}
if (!rewindEntries.some((entry) => entry.entryId === rewindEntryId)) {
reportComposerError("The selected input is no longer in the displayed history.");
rewindEntryId = null;
return;
}
rewindApplying = true;
try {
sendProtocolMethod({ method: "rewind_to", params: { entry_id: rewindEntryId } });
} catch (error) {
rewindApplying = false;
reportComposerError(error instanceof Error ? error.message : String(error));
}
}
function handleRewindKeydown(event: KeyboardEvent): boolean {
if (!rewindEntryId || event.isComposing || event.keyCode === 229 ||
event.ctrlKey || event.metaKey || event.altKey || event.shiftKey) return false;
if (["ArrowUp", "ArrowDown", "Enter", "Escape"].includes(event.key)) {
event.preventDefault();
event.stopPropagation();
if (event.key === "Escape") rewindEntryId = null;
else if (event.key === "Enter") { if (!event.repeat) applySelectedRewind(); }
else void moveRewindSelection(event.key === "ArrowUp" ? -1 : 1);
return true;
}
// Typing resumes normal draft editing without disabling the composer.
if (event.key.length === 1 || event.key === "Backspace" || event.key === "Delete") rewindEntryId = null;
return false;
}
function handleConsoleKeydown(event: KeyboardEvent): void {
if (!event.defaultPrevented && !isEditableTarget(event.target) && handleRewindKeydown(event)) return;
handleWorkerControlShortcut(event);
}
function composerRequestToProtocolMethod(
@@ -972,8 +1049,6 @@
if (!method) throw new Error("Worker state snapshot is not available");
return method;
}
case "list_rewind_targets":
return { method: "list_rewind_targets" };
case "register_peer":
return {
method: "register_peer",
@@ -990,6 +1065,7 @@
}
function handleComposerChange(snapshot: ComposerDraftSnapshot) {
if (snapshot.document !== draft.document) rewindEntryId = null;
draft = snapshot;
composerDrafts.set(activeComposerTargetKey, cachedComposerDraft(snapshot));
}
@@ -1024,12 +1100,12 @@
activeComposerTargetKey = nextKey;
syncAdmission();
const pending = admissions.get(nextKey);
const restored = pending && pending.status !== "rejected"
? cachedComposerDraft(pending.snapshot)
: composerDrafts.get(nextKey) ?? {
segments: [],
preserveExactText: false,
};
// A pending request owns its snapshot, not subsequent edits to the draft.
const restored = composerDrafts.get(nextKey) ?? (
pending && pending.status !== "rejected"
? cachedComposerDraft(pending.snapshot)
: { segments: [], preserveExactText: false }
);
void tick().then(() => {
if (activeComposerTargetKey !== nextKey) return;
composerInputElement?.restoreSegments(
@@ -1055,12 +1131,14 @@
}
function handleComposerSubmit() {
if (rewindEntryId) { applySelectedRewind(); return; }
const value = composerSnapshotOrReport();
if (!value) return;
if (value.document.trimStart().startsWith(":")) {
handleComposerCommand();
return;
}
if (!canSendComposerControl) return;
if (workerRunning) {
sendWorkerControl("cancel");
return;
@@ -1108,7 +1186,7 @@
reference: NonNullable<ComposerAttachment["reference"]>,
): void {
if (!attachments.some((candidate) => candidate.id === attachment.id) ||
!composerInputElement || !composerEditable || attachment.uploadPath !== attachmentPath()) {
!composerInputElement || !consoleTarget || attachment.uploadPath !== attachmentPath()) {
const reservation = uploadReservations.get(attachment.id);
uploadReservations.delete(attachment.id);
if (reservation !== undefined) composerInputElement?.cancelUpload(reservation);
@@ -1164,7 +1242,7 @@
}
function addAttachmentFiles(files: Iterable<File>): void {
if (!composerEditable) return;
if (!canUploadAttachments) return;
const stagedFiles = draft.segments.filter((segment) => segment.kind === "uploaded_file").length;
const available = Math.max(0, MAX_FILES_PER_SUBMISSION - attachments.length - stagedFiles);
for (const file of Array.from(files).slice(0, available)) {
@@ -1272,6 +1350,11 @@
reportComposerError(command.message);
return;
}
if (command.action === "rewind") {
if (delivery !== "submit") { reportComposerError("Rewind is a local Console action, not queued input or a notification."); return; }
await startRewindSelection();
return;
}
if (!command.request) {
if (command.notice) pushWorkspaceAlert("info", command.notice, { title: "Console command" });
composerInputElement?.clear();
@@ -1281,7 +1364,7 @@
delivery,
workerState,
protocolOpen: protocolState === "open",
sending,
sending: sending || rewindApplying,
hasText: value.content.trim().length > 0,
hasAttachments: attachments.length > 0 ||
value.segments.some((segment) => segment.kind === "uploaded_file"),
@@ -1289,7 +1372,7 @@
// Commands are controls, not idle-only chat submissions. Keep the
// removed header controls available while the Worker is busy too.
const isCommand = delivery === "submit" && command.request.kind !== "user";
if (isCommand ? !composerEditable : !canDeliverComposerDraft(deliveryState)) {
if (isCommand ? !canSendComposerControl : !canDeliverComposerDraft(deliveryState)) {
return;
}
@@ -1590,8 +1673,8 @@
if (segment.kind === "uploaded_file") stagedAttachmentPaths.delete(segment.file.artifact_id);
}
composerInputElement?.recordHistory(record.snapshot);
// A late acceptance of a previously rejected attempt must not erase
// edits made while the recovered draft was unlocked.
// Admission never locks the editor. Do not erase edits made
// after this request's snapshot, including after a rejection.
if (JSON.stringify(draft.segments) === JSON.stringify(record.snapshot.segments)) {
composerInputElement?.clear(true);
attachments = [];
@@ -1600,8 +1683,8 @@
liveWorkerState = "running";
}
} else {
// The editor has stayed frozen with the exact typed input, Undo state
// and staged resources. Unlock it; unchanged retries reuse the IDs.
// Keep the current draft and Undo state; the pending snapshot
// remains available for unchanged retries with the same IDs.
reportComposerError(message ?? "Worker rejected the input.");
}
syncAdmission();
@@ -1619,19 +1702,11 @@
}
return;
}
if (event.event === "rewind_targets") {
rewindHeadEntries = event.data.head_entries;
rewindTargets = event.data.targets;
pushWorkspaceAlert(
"info",
event.data.targets.length === 0
? "No rewind targets are available."
: `Loaded ${event.data.targets.length} rewind target(s).`,
{ id: controlAlertId, title: "Rewind targets" },
);
return;
}
if (event.event === "error") {
if (rewindApplying) {
rewindApplying = false;
reportComposerError(event.data.message);
}
const error = new Error(event.data.message);
if (
fileCompletions.pending || featureCompletions.pending ||
@@ -1652,6 +1727,10 @@
function rejectPendingCompletion(error: Error, targetKey = activeComposerTargetKey) {
admissionDisconnected(targetKey);
rewindEntryId = null;
rewindApplying = false;
// Invalidate suggestions on disconnect without locking the draft.
composerInputElement?.closeCompletion();
fileCompletions.close(error);
featureCompletions.close(error);
for (const lane of featureArgumentCompletions.values()) lane.close(error);
@@ -1688,6 +1767,7 @@
rememberCurrent = true,
) {
if (sessionId === selectedWorkerViewSessionId) return;
rewindEntryId = null;
const generation = ++workerViewSelectionGeneration;
if (rememberCurrent) rememberConsoleWorkerViewScroll();
const target = workerViews.find((view) => view.sessionId === sessionId) ??
@@ -1794,6 +1874,8 @@
const target = consoleTarget;
const targetWorker = data.worker;
const targetWorkerError = data.workerError;
rewindEntryId = null;
rewindApplying = false;
workerViewSelectionGeneration += 1;
selectedWorkerViewSessionId = null;
consoleViewScroll.clear();
@@ -1849,7 +1931,7 @@
});
</script>
<svelte:window onkeydown={handleWorkerControlShortcut} />
<svelte:window onkeydown={handleConsoleKeydown} />
<svelte:head>
<title>Worker Console · Yoi Workspace</title>
@@ -1860,28 +1942,6 @@
</svelte:head>
<div class="console-shell worker-console-shell">
{#if rewindTargets.length > 0}
<section class="card rewind-targets" aria-label="Rewind targets">
<h3>Rewind targets</h3>
<div class="rewind-target-list">
{#each rewindTargets as target (JSON.stringify(target.id))}
<button
type="button"
class="secondary-button"
disabled={protocolState !== "open" || !target.eligible}
title={target.disabled_reason ?? target.warning ?? undefined}
onclick={() => rewindTo(target)}
>
{target.preview || `${target.turn_index}`}
{#if target.warning || target.disabled_reason}
<span>{target.warning ?? target.disabled_reason}</span>
{/if}
</button>
{/each}
</div>
</section>
{/if}
{#key selectedHistoryKey}
<svelte:boundary onerror={(error) => console.error("Console transcript rendering failed", error)}>
<div class:with-task-pane={taskPaneOpen} class="console-history">
@@ -1927,7 +1987,7 @@
{#if (consoleDisplayState.kind === "ready" || consoleDisplayState.kind === "stale") && lines.length > 0}
<ol class="console-log">
{#each lines as item (item.id)}
<ConsoleLineItem {item} {attachmentUrl} />
<ConsoleLineItem {item} {attachmentUrl} rewindSelected={rewindEntryId !== null && item.entryId === rewindEntryId} />
{/each}
</ol>
{/if}
@@ -2146,12 +2206,17 @@
accept="text/*,application/json,application/pdf,image/png,image/jpeg,image/gif,image/webp"
onchange={handleFileInput}
/>
{#if rewindEntryId || rewindApplying}
<div class="rewind-status" role="status">
{rewindApplying ? "Applying rewind…" : "Rewind · ↑↓ select input · Enter apply · Esc cancel"}
<span>Rewinds to before the selected input. Tool side effects are not undone.</span>
</div>
{/if}
<ComposerInput
bind:this={composerInputElement}
historyScope={workspaceId}
ariaLabel="Console input"
ariaKeyShortcuts="Meta+Enter Control+Enter"
disabled={!composerEditable}
onchange={handleComposerChange}
onkeydown={handleComposerKeydown}
completionScope={activeComposerTargetKey}
@@ -2212,7 +2277,7 @@
type="button"
aria-label="Attach file"
title="Attach file"
disabled={!composerEditable || attachments.length >= MAX_FILES_PER_SUBMISSION}
disabled={!canUploadAttachments || attachments.length >= MAX_FILES_PER_SUBMISSION}
onclick={() => fileInput?.click()}
>
<svg
@@ -2264,16 +2329,18 @@
{/if}
<button
class="composer-send-button"
class:stop={workerRunning}
class:stop={workerRunning && !rewindEntryId}
type="submit"
aria-label={workerRunning
aria-label={rewindEntryId ? "Rewind to selected input" : workerRunning
? "Stop Worker"
: sending
? "Sending message"
: "Send message"}
disabled={composerSubmitDisabled}
>
{#if workerRunning}
{#if rewindEntryId}
<svg class="composer-send-icon" aria-hidden="true" viewBox="0 0 24 24"><path d="M9 5L3 11L9 17M3 11H15A5 5 0 0 1 15 21" /></svg>
{:else if workerRunning}
<svg
class="composer-send-icon"
aria-hidden="true"
@@ -2321,6 +2388,9 @@
</div>
<style>
.rewind-status { font-size: var(--font-size-compact); color: var(--text-muted); }
.rewind-status span { display: block; }
.console-shell {
width: 100%;
max-width: 920px;
@@ -2405,28 +2475,7 @@
transform: translateX(10px);
}
.rewind-targets {
display: flex;
align-items: center;
gap: var(--space-3);
padding: var(--space-3);
}
.rewind-targets h3 {
margin: 0;
font-size: var(--font-size-body);
}
.rewind-target-list {
display: flex;
flex-wrap: wrap;
gap: var(--space-2);
}
.rewind-target-list span {
margin-left: var(--space-2);
color: var(--text-muted);
}
.console-history {
display: grid;
@@ -397,6 +397,40 @@ test("Feature and argument completion correlate same-prefix ABA, context and oth
expect((await screen.findByRole("option")).textContent).toContain("a b/資料");
});
test.each(["running", "paused"] as const)("Composer edits remain available while Worker is %s", async (state) => {
const { cm, view } = await liveComposer();
protocolEvent({ event: "worker_state", data: { snapshot: { last_command_id: 0, state: { kind: "busy", state: { kind: "run", state } } } } });
await settleMicrotasks();
cm.focus();
await fireEvent.paste(cm.contentDOM, { clipboardData: { getData: () => "next input" } });
expect(cm.state.doc.toString()).toBe("next input");
expect(cm.state.readOnly).toBe(false);
expect(cm.contentDOM.getAttribute("contenteditable")).toBe("true");
expect(undo(cm)).toBe(true);
expect(cm.state.doc.toString()).toBe("");
expect(cm.hasFocus).toBe(true);
expect(view.container.querySelector(".composer-input.disabled")).toBeNull();
});
test("stopped Worker with retained conversation allows drafting but not sending", async () => {
vi.stubGlobal("fetch", vi.fn(async (input: RequestInfo | URL) => Response.json(
String(input).includes("/session/history") ? emptyHistoryPage() : { availability: "retained_snapshot", snapshot: emptySession() },
)));
const data = pageData();
data.worker.state = "stopped";
const view = render(ConsolePage, { data });
await screen.findByText("Read-only retained conversation");
const cm = EditorView.findFromDOM(view.container.querySelector(".cm-editor") as HTMLElement)!;
cm.focus();
await fireEvent.paste(cm.contentDOM, { clipboardData: { getData: () => "draft for later" } });
expect(cm.state.doc.toString()).toBe("draft for later");
expect(cm.state.readOnly).toBe(false);
expect(cm.contentDOM.getAttribute("contenteditable")).toBe("true");
await fireEvent.submit(view.container.querySelector(".console-composer")!);
expect(multiplexer.sendWorkerMethod).not.toHaveBeenCalled();
expect(cm.state.doc.toString()).toBe("draft for later");
});
test("Submit rejection keeps exact typed draft, request/invocation IDs, and only acceptance clears", async () => {
const alert = vi.spyOn(alerts, "pushWorkspaceAlert");
const { cm, form } = await liveComposer();
@@ -406,13 +440,16 @@ test("Submit rejection keeps exact typed draft, request/invocation IDs, and only
expect(first.method).toBe("submit");
expect(first.params.input[0]).toMatchObject({ kind: "feature_invoke", invocation: { identity: runDescriptor.identity, arguments: [{ name: "path", value: { kind: "string", value: "a b/資料" } }] } });
const doc = cm.state.doc.toString();
await waitFor(() => expect(cm.state.readOnly).toBe(true));
expect(cm.state.readOnly).toBe(false);
await fireEvent.keyDown(cm.contentDOM, { key: "Backspace" });
await fireEvent.keyDown(cm.contentDOM, { key: "z", ctrlKey: true });
expect(cm.state.doc.toString()).not.toBe(doc);
expect(undo(cm)).toBe(true);
expect(cm.state.doc.toString()).toBe(doc);
await fireEvent.submit(form);
expect(multiplexer.sendWorkerMethod.mock.calls.filter(([method]) => method.method === "submit")).toHaveLength(1);
protocolEvent({ event: "submission_rejected", data: { submission_request_id: "other-client", message: "foreign" } });
await settleMicrotasks();
expect(cm.state.readOnly).toBe(true);
expect(cm.state.readOnly).toBe(false);
protocolEvent({ event: "submission_rejected", data: { submission_request_id: first.params.submission_request_id, message: "Feature validation failed" } });
await waitFor(() => expect(cm.state.readOnly).toBe(false));
expect(alert).toHaveBeenCalledWith("error", "Feature validation failed", expect.any(Object));
@@ -505,12 +542,18 @@ test("cached unsent typed draft survives target round-trip and remains sendable"
expect(multiplexer.sendWorkerMethod).toHaveBeenLastCalledWith({ method: "submit", params: { submission_request_id: expect.any(String), input: [expect.objectContaining({ kind: "feature_invoke" })] } });
});
test("unknown Submit survives target ABA and cannot be deleted into a new admission", async () => {
test("unknown Submit keeps its original retry while edits survive disconnect and target ABA", async () => {
const { cm, form, view } = await liveComposer();
await selectRun(cm);
await fireEvent.submit(form);
const first = multiplexer.sendWorkerMethod.mock.calls.at(-1)![0] as Extract<Method, { method: "submit" }>;
await waitFor(() => expect(cm.state.readOnly).toBe(true));
latestListener().onStatus?.("closed", { kind: "transport", message: "ack lost" });
await settleMicrotasks();
expect(cm.state.readOnly).toBe(false);
await fireEvent.keyDown(cm.contentDOM, { key: "Backspace" });
cm.dispatch({ changes: { from: 0, to: cm.state.doc.length, insert: "new draft while disconnected" }, selection: { anchor: 28 } });
await fireEvent.submit(form);
expect(multiplexer.sendWorkerMethod.mock.calls.filter(([method]) => method.method === "submit")).toHaveLength(1);
await view.rerender({ data: pageData("worker-b", "runtime-b") });
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledTimes(2));
latestListener().onFrame(subscribedFrame());
@@ -518,13 +561,15 @@ test("unknown Submit survives target ABA and cannot be deleted into a new admiss
await view.rerender({ data: pageData() });
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledTimes(3));
latestListener().onFrame(subscribedFrame());
await waitFor(() => expect(cm.dom.querySelector(".composer-typed-chip")).not.toBeNull());
expect(cm.state.readOnly).toBe(true);
const doc = cm.state.doc.toString();
await fireEvent.keyDown(cm.contentDOM, { key: "Backspace" });
expect(cm.state.doc.toString()).toBe(doc);
await waitFor(() => expect(cm.state.doc.toString()).toBe("new draft while disconnected"));
expect(cm.state.readOnly).toBe(false);
await fireEvent.submit(form);
expect(multiplexer.sendWorkerMethod.mock.calls.filter(([method]) => method.method === "submit")).toHaveLength(1);
await fireEvent.click(screen.getByRole("button", { name: "Retry admission" }));
expect(multiplexer.sendWorkerMethod).toHaveBeenLastCalledWith(first);
protocolEvent({ event: "submission_accepted", data: { submission_request_id: first.params.submission_request_id, submission_id: "accepted", disposition: "queued" } });
await settleMicrotasks();
expect(cm.state.doc.toString()).toBe("new draft while disconnected");
view.unmount();
});
@@ -902,11 +947,15 @@ test.each(["idle", "running", "paused"] as const)("Composer completion executes
method: "compact", params: { command: expect.objectContaining({ command_id: 1 }) },
}));
await waitFor(() => expect(cm.state.doc.toString()).toBe(""));
protocolEvent({ event: "user_message", data: { entry_id: "rewind-user", segments: [{ kind: "text", content: "rewind input" }] } });
await screen.findByText("rewind input");
const callsBeforeRewind = multiplexer.sendWorkerMethod.mock.calls.length;
cm.dispatch({ changes: { from: 0, insert: ":rew" }, selection: { anchor: 4 } });
const option = await screen.findByRole("option");
await fireEvent.pointerDown(option);
await fireEvent.click(option);
await waitFor(() => expect(multiplexer.sendWorkerMethod).toHaveBeenCalledWith({ method: "list_rewind_targets" }));
await waitFor(() => expect(view.container.querySelector(".rewind-selected")?.textContent).toContain("rewind input"));
expect(multiplexer.sendWorkerMethod).toHaveBeenCalledTimes(callsBeforeRewind);
expect(multiplexer.sendWorkerMethod.mock.calls.every(([method]) => method.method !== "submit" && method.method !== "cancel")).toBe(true);
});
@@ -1656,3 +1705,119 @@ test("pending groups independently follow authoritative changes, including zero
latestListener().onFrame(subscribedFrame(session));
await waitFor(() => expect(screen.queryByRole("region", { name: "Pending activations" })).toBeNull());
});
async function enterRewind(cm: EditorView, form: HTMLFormElement) {
cm.focus();
cm.dispatch({ changes: { from: 0, to: cm.state.doc.length, insert: ":rewind" }, selection: { anchor: 7 } });
await fireEvent.submit(form);
await waitFor(() => expect(document.querySelector(".rewind-selected")).not.toBeNull());
}
function rewindSession(): SessionSnapshot {
return { ...emptySession(), entries: ["first", "second"].map((id) => ({
kind: "user_input", entry_id: id, timestamp: 1, provenance: "human_input", derived_from: [],
segments: [{ kind: "text", content: "identical input" }],
})) };
}
test("rewind selects existing history by ID, follows arrows, tolerates appended history and suppresses repeated Enter", async () => {
const { cm, form, view } = await liveComposer();
latestListener().onFrame(subscribedFrame(rewindSession()));
await waitFor(() => expect(view.container.querySelectorAll(".console-line.user")).toHaveLength(2));
const rows = () => [...view.container.querySelectorAll(".console-line.user")];
await enterRewind(cm, form);
expect(multiplexer.sendWorkerMethod).not.toHaveBeenCalled();
expect(rows()[1].classList.contains("rewind-selected")).toBe(true);
await fireEvent.keyDown(cm.contentDOM, { key: "ArrowUp" });
expect(rows()[0].classList.contains("rewind-selected")).toBe(true);
expect(cm.state.doc.toString()).toBe("");
await fireEvent.keyDown(cm.contentDOM, { key: "ArrowDown" });
expect(rows()[1].classList.contains("rewind-selected")).toBe(true);
protocolEvent({ event: "user_message", data: { entry_id: "third", segments: [{ kind: "text", content: "newly appended" }] } });
await screen.findByText("newly appended");
expect(rows()[1].classList.contains("rewind-selected")).toBe(true);
await fireEvent.keyDown(cm.contentDOM, { key: "Enter" });
await fireEvent.keyDown(cm.contentDOM, { key: "Enter", repeat: true });
await fireEvent.keyDown(cm.contentDOM, { key: "Enter" });
expect(multiplexer.sendWorkerMethod).toHaveBeenCalledExactlyOnceWith({ method: "rewind_to", params: { entry_id: "second" } });
expect(cm.state.readOnly).toBe(false);
protocolEvent({ event: "rewind_applied", data: { session: { ...rewindSession(), entries: rewindSession().entries.slice(0, 1) }, input: [{ kind: "text", content: "identical input" }], summary: { truncated_to_entries: 1, discarded_entries: 2, tool_side_effect_warning: true } } });
await waitFor(() => expect(cm.state.doc.toString()).toBe("identical input"));
expect(view.container.querySelector(".rewind-selected")).toBeNull();
expect(cm.hasFocus).toBe(true);
});
test("rewind Escape and draft editing leave selection without sending; rejection keeps it available", async () => {
const alert = vi.spyOn(alerts, "pushWorkspaceAlert");
const { cm, form, view } = await liveComposer();
latestListener().onFrame(subscribedFrame(rewindSession()));
await waitFor(() => expect(view.container.querySelectorAll(".console-line.user")).toHaveLength(2));
await enterRewind(cm, form);
await fireEvent.keyDown(cm.contentDOM, { key: "Escape" });
expect(view.container.querySelector(".rewind-selected")).toBeNull();
expect(multiplexer.sendWorkerMethod).not.toHaveBeenCalled();
await enterRewind(cm, form);
await fireEvent.keyDown(cm.contentDOM, { key: "Enter" });
protocolEvent({ event: "error", data: { code: "invalid_request", message: "Target is outside current segment" } });
await waitFor(() => expect(alert).toHaveBeenCalledWith("error", "Target is outside current segment", expect.any(Object)));
expect(view.container.querySelector(".rewind-selected")).not.toBeNull();
expect((screen.getByRole("button", { name: "Rewind to selected input" }) as HTMLButtonElement).disabled).toBe(false);
await fireEvent.paste(cm.contentDOM, { clipboardData: { getData: () => "new draft" } });
expect(cm.state.doc.toString()).toBe("new draft");
expect(view.container.querySelector(".rewind-selected")).toBeNull();
expect(multiplexer.sendWorkerMethod).toHaveBeenCalledTimes(1);
});
test("rewind reply preserves draft edits made while applying and selection is cleared on target change", async () => {
const { cm, form, view } = await liveComposer();
latestListener().onFrame(subscribedFrame(rewindSession()));
await waitFor(() => expect(view.container.querySelectorAll(".console-line.user")).toHaveLength(2));
await enterRewind(cm, form);
await fireEvent.keyDown(cm.contentDOM, { key: "Enter" });
await fireEvent.paste(cm.contentDOM, { clipboardData: { getData: () => "keep this edit" } });
protocolEvent({ event: "rewind_applied", data: { session: rewindSession(), input: [{ kind: "text", content: "old input" }], summary: { truncated_to_entries: 1, discarded_entries: 1, tool_side_effect_warning: false } } });
await settleMicrotasks();
expect(cm.state.doc.toString()).toBe("keep this edit");
await enterRewind(cm, form);
latestListener().onStatus?.("closed", { kind: "transport", message: "disconnected" });
await waitFor(() => expect(view.container.querySelector(".rewind-selected")).toBeNull());
expect(cm.state.readOnly).toBe(false);
latestListener().onFrame(subscribedFrame(rewindSession()));
await enterRewind(cm, form);
await view.rerender({ data: pageData("worker-b", "runtime-b") });
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledTimes(2));
expect(view.container.querySelector(".rewind-selected")).toBeNull();
});
test("rewind Up loads the next history page and selects its nearest user input", async () => {
const fetchMock = vi.fn(async (input: RequestInfo | URL) => {
const url = String(input);
return Response.json(!url.includes("/session/history") ? { availability: "live_protocol" }
: url.includes("cursor=") ? historyPage([1], null) : historyPage([2], "older-2"));
});
vi.stubGlobal("fetch", fetchMock);
const view = render(ConsolePage, { data: pageData() });
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce());
latestListener().onFrame(subscribedFrame());
await waitFor(() => expect(view.container.querySelector(".console-line.user")?.textContent).toContain("question 2"));
const cm = EditorView.findFromDOM(view.container.querySelector(".cm-editor") as HTMLElement)!;
await enterRewind(cm, view.container.querySelector(".console-composer") as HTMLFormElement);
await fireEvent.keyDown(cm.contentDOM, { key: "ArrowUp" });
await waitFor(() => expect(view.container.querySelector(".rewind-selected")?.textContent).toContain("question 1"));
await fireEvent.keyDown(cm.contentDOM, { key: "Enter" });
expect(multiplexer.sendWorkerMethod).toHaveBeenCalledExactlyOnceWith({ method: "rewind_to", params: { entry_id: "user-1" } });
});
test("rewind does not confirm an IME Enter", async () => {
const { cm, form, view } = await liveComposer();
latestListener().onFrame(subscribedFrame(rewindSession()));
await waitFor(() => expect(view.container.querySelectorAll(".console-line.user")).toHaveLength(2));
await enterRewind(cm, form);
await fireEvent.compositionStart(cm.contentDOM);
await fireEvent.keyDown(cm.contentDOM, { key: "Enter", isComposing: true, keyCode: 229 });
expect(multiplexer.sendWorkerMethod).not.toHaveBeenCalled();
expect(cm.state.readOnly).toBe(false);
await fireEvent.compositionEnd(cm.contentDOM);
});
@@ -76,7 +76,6 @@ Deno.test("Composer history uses only the multiline input boundaries", () => {
const base = {
lineCount: 3,
selectionEmpty: true,
readOnly: false,
composing: false,
};
@@ -108,14 +107,6 @@ Deno.test("Composer history uses only the multiline input boundaries", () => {
selectionEmpty: false,
}),
);
assert(
!shouldBrowseComposerHistory({
...base,
direction: "newer",
cursorLine: 3,
readOnly: true,
}),
);
assert(
!shouldBrowseComposerHistory({
...base,