From 9f16efb8da79e2a10af14674bbc6ecfc075ef1a7 Mon Sep 17 00:00:00 2001 From: BingCHuanJ Date: Sat, 19 Sep 2026 01:17:37 +0800 Subject: [PATCH 1/4] feat(remote-connect): support targeted session rollback and edit-resend from remote control The remote control surface previously offered only resend and delete, where delete was a local hide that left the turn and file mutations intact on the desktop. Port the targeted rollback and edit-and-resend capability to OpenBitFun 1.0.1: - Project the owning turn identity onto user `ChatMessage`s (`turn_id` + `turn_index`). Targeted rollback addresses turns rather than messages, and `turn_index` carries the same optimistic-concurrency guard the desktop sends so a stale transcript cannot retire the wrong turn. - Add `RemoteCommand::RollbackSessionToTurn` and `RemoteResponse::SessionRolledBack` to the owner contract in `openbitfun-services-integrations`, routed through the session command group. - Implement `RemoteSessionRuntimeHost::rollback_session_to_turn` in core on top of `AgentRuntime::rollback_session_to_turn`, passing the workspace ownership gate (`ensure_remote_binding_runtime_ownership`) and rejecting remote workspaces. - Mobile web gains "edit & resend" and "roll back to here" on user messages, using `MobileConfirmSheet` and `MobileTextarea` to conform to `@openbitfun/ui/mobile` component boundaries. - Disable rollback mutations while a turn is active to preserve consistency with desktop `assertSessionIdleForHistoryMutation`. - Nudge the stream poller immediately upon rollback completion to pull the authoritative message snapshot from the host. - Provide full i18n support across en-US, zh-CN, and zh-TW. --- .../core/src/service_agent_runtime.rs | 82 +++++++- .../src/remote_connect.rs | 178 +++++++++++++++++- .../tests/remote_connect_contracts.rs | 94 +++++++++ .../src/components/ChatMessageActions.tsx | 39 +++- src/mobile-web/src/i18n/messages.ts | 39 ++++ src/mobile-web/src/pages/ChatPage.tsx | 156 ++++++++++++++- .../src/services/RemoteSessionManager.ts | 41 ++++ .../src/styles/components/chat.scss | 35 ++++ 8 files changed, 650 insertions(+), 14 deletions(-) diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs index 6eb2db285a..dbd4f4d209 100644 --- a/src/crates/assembly/core/src/service_agent_runtime.rs +++ b/src/crates/assembly/core/src/service_agent_runtime.rs @@ -22,11 +22,11 @@ use openbitfun_agent_runtime::sdk::{ use openbitfun_events::AgenticEvent; #[cfg(feature = "remote-connect")] use openbitfun_runtime_ports::{ - AgentDialogSteerRequest, AgentInputAttachment, AgentSubmissionSource, - AgentTurnCancellationRequest, DialogSteerOutcome, PermissionPolicyPreset, - RemoteControlStatePort, RemoteControlStateRequest, RemoteControlStateSnapshot, - RemoteSessionWorkspaceIdentity, RuntimeServiceCapability, RuntimeServicePort, - ToolPermissionConfig, + AgentDialogSteerRequest, AgentInputAttachment, AgentSessionComposerUpdate, + AgentSubmissionSource, AgentTurnCancellationRequest, DialogSteerOutcome, + PermissionPolicyPreset, RemoteControlStatePort, RemoteControlStateRequest, + RemoteControlStateSnapshot, RemoteSessionWorkspaceIdentity, RuntimeServiceCapability, + RuntimeServicePort, ToolPermissionConfig, }; use openbitfun_runtime_ports::{ AgentDialogTurnPort, AgentDialogTurnRequest, AgentLifecycleDeliveryPort, @@ -54,10 +54,11 @@ use openbitfun_services_integrations::remote_connect::{ RemoteInitialSyncRuntimeHost, RemoteInteractionRuntimeHost, RemoteModelCapabilityFact, RemoteModelCatalog, RemoteModelCatalogFacts, RemoteModelFacts, RemotePermissionMode, RemotePollRuntimeHost, RemoteRecentWorkspaceFacts, RemoteSessionMetadata, - RemoteSessionModelSelection, RemoteSessionRuntimeHost, RemoteSessionStateTracker, - RemoteSessionTrackerHost, RemoteTerminalPrewarmRequest, RemoteWorkspaceFacts, - RemoteWorkspaceFileRuntimeHost, RemoteWorkspaceKind as RemoteConnectWorkspaceKind, - RemoteWorkspaceRuntimeHost, RemoteWorkspaceUpdate, + RemoteSessionModelSelection, RemoteSessionRollbackOutcome, RemoteSessionRuntimeHost, + RemoteSessionStateTracker, RemoteSessionTrackerHost, RemoteTerminalPrewarmRequest, + RemoteWorkspaceFacts, RemoteWorkspaceFileRuntimeHost, + RemoteWorkspaceKind as RemoteConnectWorkspaceKind, RemoteWorkspaceRuntimeHost, + RemoteWorkspaceUpdate, }; use std::sync::Arc; use std::time::Duration; @@ -866,6 +867,7 @@ fn remote_chat_history_turn_from_core_turn( RemoteChatHistoryTurn { turn_id: turn.turn_id.clone(), + turn_index: turn.turn_index, user_message_id: turn.user_message.id.clone(), user_display_content: user_projection.content, user_timestamp_ms: turn.user_message.timestamp, @@ -3411,6 +3413,60 @@ impl RemoteSessionRuntimeHost for CoreRemoteSessionRuntimeHost { CoreServiceAgentRuntime::load_remote_chat_messages(session_storage_dir, session_id).await } + /// Reuse the desktop targeted-rollback transaction so a remote client + /// retires turns and restores files for real instead of hiding messages in + /// its own transcript. + async fn rollback_session_to_turn( + &self, + session_id: &str, + target_turn_id: &str, + expected_storage_turn_index: Option, + ) -> Result { + let binding = CoreServiceAgentRuntime::resolve_session_workspace_binding(session_id) + .await + .ok_or_else(|| { + format!("Session workspace binding not available for session: {session_id}") + })?; + if binding.is_remote() { + return Err("Session rollback is unavailable for remote workspaces".to_string()); + } + ensure_remote_binding_runtime_ownership(self.coordinator.as_ref(), &binding).await?; + + let outcome = self + .runtime + .rollback_session_to_turn(AgentSessionRollbackToTurnRequest { + workspace_path: binding.logical_workspace_path_string(), + workspace_id: binding.workspace_id.clone(), + workspace_hostname: Some(binding.session_identity.hostname.clone()), + session_id: session_id.to_string(), + target_turn_id: target_turn_id.to_string(), + expected_storage_turn_index, + expected_catalog_revision: None, + remote_connection_id: None, + remote_ssh_host: None, + }) + .await + .map_err(|error| error.into_message())?; + + match outcome { + AgentSessionRollbackToTurnOutcome::Completed { result } => { + Ok(RemoteSessionRollbackOutcome { + retired_turn_ids: result.retired_turn_ids, + restored_files: result.restored_files, + composer_text: match result.composer { + AgentSessionComposerUpdate::Replace { text } => Some(text), + AgentSessionComposerUpdate::Preserve + | AgentSessionComposerUpdate::Clear => None, + }, + changed: result.changed, + }) + } + AgentSessionRollbackToTurnOutcome::RecoveryRequired { reason, .. } => Err(format!( + "Session rollback requires recovery before it can continue: {reason}" + )), + } + } + async fn delete_session( &self, session_storage_dir: &std::path::Path, @@ -3991,6 +4047,14 @@ mod tests { .and_then(|source| source.split("fn remove_tracker").next()) .expect("remote session delete"); assert!(delete.contains("ensure_remote_binding_runtime_ownership")); + + let rollback = remote_session_host + .split("async fn rollback_session_to_turn") + .nth(1) + .and_then(|source| source.split("async fn delete_session").next()) + .expect("remote session rollback"); + assert!(rollback.contains("ensure_remote_binding_runtime_ownership")); + assert!(rollback.contains("binding.is_remote()")); } #[test] diff --git a/src/crates/services/services-integrations/src/remote_connect.rs b/src/crates/services/services-integrations/src/remote_connect.rs index af9e166285..4d13073245 100644 --- a/src/crates/services/services-integrations/src/remote_connect.rs +++ b/src/crates/services/services-integrations/src/remote_connect.rs @@ -1368,6 +1368,32 @@ pub fn remote_session_deleted_response(session_id: impl Into) -> RemoteR } } +/// What the host observed while rolling a session back to a turn. +/// +/// This mirrors the fields the desktop rollback result exposes that a remote +/// transcript actually needs, so this crate keeps describing the wire contract +/// without depending on the runtime port types. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RemoteSessionRollbackOutcome { + pub retired_turn_ids: Vec, + pub restored_files: Vec, + pub composer_text: Option, + pub changed: bool, +} + +pub fn remote_session_rolled_back_response( + session_id: impl Into, + outcome: RemoteSessionRollbackOutcome, +) -> RemoteResponse { + RemoteResponse::SessionRolledBack { + session_id: session_id.into(), + retired_turn_ids: outcome.retired_turn_ids, + restored_files: outcome.restored_files, + composer_text: outcome.composer_text, + changed: outcome.changed, + } +} + #[async_trait::async_trait] pub trait RemoteSessionRuntimeHost: Send + Sync { async fn workspace_by_id(&self, _workspace_id: &str) -> Result { @@ -1413,6 +1439,16 @@ pub trait RemoteSessionRuntimeHost: Send + Sync { session_storage_dir: &Path, session_id: &str, ) -> Result<(Vec, bool), String>; + /// Roll the session back to `target_turn_id`, retiring that turn and the + /// ones after it and restoring the files those turns wrote. Hosts are + /// expected to reuse the same targeted-rollback path the desktop uses rather + /// than hiding turns, and to reject sessions whose workspace they do not own. + async fn rollback_session_to_turn( + &self, + session_id: &str, + target_turn_id: &str, + expected_storage_turn_index: Option, + ) -> Result; async fn delete_session( &self, session_storage_dir: &Path, @@ -1690,6 +1726,29 @@ where Err(message) => RemoteResponse::Error { message }, } } + RemoteCommand::RollbackSessionToTurn { + session_id, + target_turn_id, + expected_storage_turn_index, + } => { + if target_turn_id.trim().is_empty() { + return RemoteResponse::Error { + message: "Rollback requires a target turn id".into(), + }; + } + + match host + .rollback_session_to_turn( + session_id, + target_turn_id.trim(), + *expected_storage_turn_index, + ) + .await + { + Ok(outcome) => remote_session_rolled_back_response(session_id.clone(), outcome), + Err(message) => RemoteResponse::Error { message }, + } + } _ => RemoteResponse::Error { message: "Unknown session command".into(), }, @@ -2202,6 +2261,11 @@ pub struct ChatMessage { pub metadata: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub turn_id: Option, + /// Storage turn index for `turn_id`, carried so a remote rollback can send + /// the same optimistic-concurrency guard the desktop sends and cannot + /// retire a different turn after the transcript moved underneath it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub turn_index: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub status: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -2235,6 +2299,7 @@ pub struct ChatMessageItem { #[derive(Debug, Clone, PartialEq)] pub struct RemoteChatHistoryTurn { pub turn_id: String, + pub turn_index: usize, pub user_message_id: String, pub user_display_content: String, pub user_timestamp_ms: u64, @@ -2300,6 +2365,7 @@ pub fn build_remote_chat_messages(turns: Vec) -> Vec) -> Vec, + }, CancelTool { tool_id: String, reason: Option, @@ -2913,6 +2995,19 @@ pub enum RemoteResponse { SessionDeleted { session_id: String, }, + SessionRolledBack { + session_id: String, + /// Turns the host retired, including the target turn itself. Reported so + /// a client can tell the user what was withdrawn; the transcript itself + /// is repaired by the next poll's authoritative `message_snapshot`. + retired_turn_ids: Vec, + restored_files: Vec, + /// Prompt text the desktop would have put back into its composer, so a + /// remote client can offer the same "edit and resend" continuation. + #[serde(skip_serializing_if = "Option::is_none")] + composer_text: Option, + changed: bool, + }, InitialSync { #[serde(default, skip_serializing_if = "Option::is_none")] workspace_id: Option, @@ -3097,7 +3192,8 @@ where | RemoteCommand::SetSessionModel { .. } | RemoteCommand::UpdateSessionTitle { .. } | RemoteCommand::GetSessionMessages { .. } - | RemoteCommand::DeleteSession { .. } => host.handle_session_command(command).await, + | RemoteCommand::DeleteSession { .. } + | RemoteCommand::RollbackSessionToTurn { .. } => host.handle_session_command(command).await, RemoteCommand::PollSession { .. } => host.handle_poll_command(command).await, @@ -4610,6 +4706,7 @@ mod tests { model_updates: Mutex>)>>, removed_trackers: Mutex>, history_error: Option, + rollback_requests: Mutex)>>, } fn fake_workspace( @@ -4797,7 +4894,8 @@ mod tests { content: "hello".to_string(), timestamp: "1".to_string(), metadata: None, - turn_id: None, + turn_id: Some("turn-1".to_string()), + turn_index: Some(0), status: None, error: None, images: None, @@ -4809,6 +4907,25 @@ mod tests { )) } + async fn rollback_session_to_turn( + &self, + session_id: &str, + target_turn_id: &str, + expected_storage_turn_index: Option, + ) -> Result { + self.rollback_requests.lock().unwrap().push(( + session_id.to_string(), + target_turn_id.to_string(), + expected_storage_turn_index, + )); + Ok(RemoteSessionRollbackOutcome { + retired_turn_ids: vec!["turn-2".to_string()], + restored_files: vec!["src/main.rs".to_string()], + composer_text: Some("previous prompt".to_string()), + changed: true, + }) + } + async fn delete_session( &self, _session_storage_dir: &Path, @@ -5092,6 +5209,59 @@ mod tests { ); } + #[tokio::test] + async fn remote_session_handler_forwards_rollback_target_and_guard() { + let host = FakeSessionHost::default(); + + let response = handle_remote_session_command( + &host, + &RemoteCommand::RollbackSessionToTurn { + session_id: "session-a".to_string(), + target_turn_id: " turn-1 ".to_string(), + expected_storage_turn_index: Some(3), + }, + ) + .await; + + assert_eq!( + response, + RemoteResponse::SessionRolledBack { + session_id: "session-a".to_string(), + retired_turn_ids: vec!["turn-2".to_string()], + restored_files: vec!["src/main.rs".to_string()], + composer_text: Some("previous prompt".to_string()), + changed: true, + } + ); + assert_eq!( + host.rollback_requests.lock().unwrap().as_slice(), + [("session-a".to_string(), "turn-1".to_string(), Some(3))] + ); + } + + #[tokio::test] + async fn remote_session_handler_rejects_rollback_without_target_turn() { + let host = FakeSessionHost::default(); + + let response = handle_remote_session_command( + &host, + &RemoteCommand::RollbackSessionToTurn { + session_id: "session-a".to_string(), + target_turn_id: " ".to_string(), + expected_storage_turn_index: None, + }, + ) + .await; + + assert_eq!( + response, + RemoteResponse::Error { + message: "Rollback requires a target turn id".to_string(), + } + ); + assert!(host.rollback_requests.lock().unwrap().is_empty()); + } + #[tokio::test] async fn remote_session_handler_propagates_history_outcome_unknown() { let host = FakeSessionHost { @@ -5132,6 +5302,7 @@ mod tests { timestamp: "1".to_string(), metadata: None, turn_id: Some(turn_id.to_string()), + turn_index: None, status: status.map(str::to_string), error: None, tools: None, @@ -5203,6 +5374,7 @@ mod tests { timestamp: "1".to_string(), metadata: None, turn_id: None, + turn_index: None, status: None, error: None, tools: None, @@ -5268,6 +5440,7 @@ mod tests { timestamp: "1".to_string(), metadata: None, turn_id: None, + turn_index: None, status: None, error: None, tools: None, @@ -5317,6 +5490,7 @@ mod tests { timestamp: "1".to_string(), metadata: None, turn_id: None, + turn_index: None, status: None, error: None, tools: None, diff --git a/src/crates/services/services-integrations/tests/remote_connect_contracts.rs b/src/crates/services/services-integrations/tests/remote_connect_contracts.rs index fb2403d108..1dae68f6ed 100644 --- a/src/crates/services/services-integrations/tests/remote_connect_contracts.rs +++ b/src/crates/services/services-integrations/tests/remote_connect_contracts.rs @@ -542,6 +542,8 @@ fn remote_chat_history_assembly_preserves_message_shape_and_item_order() { assert_eq!(messages[0].role, "user"); assert_eq!(messages[0].content, "original question"); assert_eq!(messages[0].timestamp, "1"); + assert_eq!(messages[0].turn_id.as_deref(), Some("turn-1")); + assert_eq!(messages[0].turn_index, Some(4)); assert_eq!( messages[0].images.as_ref().unwrap()[0], ChatImageAttachment { @@ -552,6 +554,7 @@ fn remote_chat_history_assembly_preserves_message_shape_and_item_order() { assert_eq!(messages[1].id, "turn-1_assistant"); assert_eq!(messages[1].role, "assistant"); + assert_eq!(messages[1].turn_index, None); assert_eq!(messages[1].content, "visible text"); assert_eq!(messages[1].timestamp, "1"); assert_eq!(messages[1].thinking.as_deref(), Some("visible thought")); @@ -664,6 +667,7 @@ fn remote_connect_cancel_and_restore_policy_preserve_runtime_decisions() { fn remote_history_contract_turn(is_in_progress: bool) -> RemoteChatHistoryTurn { RemoteChatHistoryTurn { turn_id: "turn-1".to_string(), + turn_index: 4, user_message_id: "user-1".to_string(), user_display_content: "original question".to_string(), user_timestamp_ms: 1_000, @@ -1365,6 +1369,25 @@ async fn remote_connect_command_owner_preserves_cancel_and_group_routing() { ); } +#[tokio::test] +async fn remote_connect_command_owner_routes_rollback_to_the_session_group() { + let host = RecordingCommandHost::default(); + + let response = handle_remote_command( + &host, + &RemoteCommand::RollbackSessionToTurn { + session_id: "session-1".to_string(), + target_turn_id: "turn-7".to_string(), + expected_storage_turn_index: Some(6), + }, + RemoteConnectSubmissionSource::Relay, + ) + .await; + + assert!(matches!(response, RemoteResponse::SessionCreated { .. })); + assert_eq!(host.events(), vec!["session"]); +} + #[tokio::test] async fn remote_connect_dialog_runtime_owns_restore_prewarm_and_submit_order() { let host = RecordingDialogHost::new(false, Some("D:/workspace/project")); @@ -2556,6 +2579,7 @@ fn remote_connect_message_dtos_keep_current_wire_shape() { timestamp: "1".to_string(), metadata: None, turn_id: Some("turn-1".to_string()), + turn_index: None, status: Some("done".to_string()), error: None, tools: Some(vec![RemoteToolStatus { @@ -2803,6 +2827,75 @@ fn remote_connect_response_wire_shape_lives_in_owner_contract() { assert_eq!(title_updated["title"], "Renamed session"); } +#[test] +fn remote_connect_rollback_wire_shape_lives_in_owner_contract() { + let with_guard: RemoteCommand = serde_json::from_value(serde_json::json!({ + "cmd": "rollback_session_to_turn", + "session_id": "session-1", + "target_turn_id": "turn-7", + "expected_storage_turn_index": 6 + })) + .expect("deserialize rollback command with guard"); + let without_guard: RemoteCommand = serde_json::from_value(serde_json::json!({ + "cmd": "rollback_session_to_turn", + "session_id": "session-1", + "target_turn_id": "turn-7" + })) + .expect("deserialize rollback command without guard"); + + assert!(matches!( + with_guard, + RemoteCommand::RollbackSessionToTurn { + expected_storage_turn_index: Some(6), + ref target_turn_id, + .. + } if target_turn_id == "turn-7" + )); + assert!(matches!( + without_guard, + RemoteCommand::RollbackSessionToTurn { + expected_storage_turn_index: None, + .. + } + )); + + let rolled_back = serde_json::to_value(RemoteResponse::SessionRolledBack { + session_id: "session-1".to_string(), + retired_turn_ids: vec!["turn-8".to_string(), "turn-9".to_string()], + restored_files: vec!["src/main.rs".to_string()], + composer_text: Some("original question".to_string()), + changed: true, + }) + .expect("serialize rollback response"); + + assert_eq!(rolled_back["resp"], "session_rolled_back"); + assert_eq!(rolled_back["retired_turn_ids"][1], "turn-9"); + assert_eq!(rolled_back["restored_files"][0], "src/main.rs"); + assert_eq!(rolled_back["composer_text"], "original question"); + assert_eq!(rolled_back["changed"], true); + + let without_composer = serde_json::to_value(RemoteResponse::SessionRolledBack { + session_id: "session-1".to_string(), + retired_turn_ids: Vec::new(), + restored_files: Vec::new(), + composer_text: None, + changed: false, + }) + .expect("serialize rollback response without composer text"); + assert!(without_composer.get("composer_text").is_none()); +} + +#[test] +fn remote_connect_user_message_carries_rollback_turn_identity_on_the_wire() { + let messages = build_remote_chat_messages(vec![remote_history_contract_turn(false)]); + let user = serde_json::to_value(&messages[0]).expect("serialize user message"); + let assistant = serde_json::to_value(&messages[1]).expect("serialize assistant message"); + + assert_eq!(user["turn_id"], "turn-1"); + assert_eq!(user["turn_index"], 4); + assert!(assistant.get("turn_index").is_none()); +} + fn sample_remote_model_catalog(version: u64) -> RemoteModelCatalog { RemoteModelCatalog { version, @@ -3453,6 +3546,7 @@ fn remote_connect_poll_helpers_preserve_delta_and_completion_policy() { timestamp: "2".to_string(), metadata: None, turn_id: Some("turn-1".to_string()), + turn_index: None, status: Some("done".to_string()), error: None, tools: None, diff --git a/src/mobile-web/src/components/ChatMessageActions.tsx b/src/mobile-web/src/components/ChatMessageActions.tsx index d2149471d8..fcd5b46c33 100644 --- a/src/mobile-web/src/components/ChatMessageActions.tsx +++ b/src/mobile-web/src/components/ChatMessageActions.tsx @@ -1,4 +1,4 @@ -import { Copy as LucideCopy, RotateCw as LucideRotateCw, Trash2 as LucideTrash2 } from 'lucide-react'; +import { Copy as LucideCopy, Edit3 as LucideEdit3, RotateCcw as LucideRotateCcw, RotateCw as LucideRotateCw, Trash2 as LucideTrash2 } from 'lucide-react'; import React from 'react'; import { MobileActionSheet, type MobileActionSheetItem } from '@openbitfun/ui/mobile'; import { useI18n } from '../i18n'; @@ -7,21 +7,54 @@ import type { ChatMessage } from '../services/RemoteSessionManager'; interface ChatMessageActionsProps { deleting: boolean; message: ChatMessage | null; + streaming?: boolean; onClose: () => void; onCopy: () => void; onDelete: () => void; onResend: () => void; + onEdit?: () => void; + onRollback?: () => void; } const CopyIcon = () =>