feat(agentic,web-ui): user steering with pending queue and agent runtime refactor (#610)

Backend changes:
- Remove orphan ModelRound struct and model_round.rs module; the actual
  round lifecycle is fully tracked through execution engine state
- Simplify DialogTurn / DialogTurnState — only TurnStats and new_turn_id
  helper survive; on-disk shape lives in SessionState as before
- Refactor execution engine, round_executor, and stream_processor to align
  with the simplified runtime model
- Add user-steering support in coordinator and scheduler: mid-turn messages
  from the user can be injected without aborting the current round
- Improve agentic tool pipeline (pipeline/tool_pipeline.rs, types.rs),
  deep-review policy, and review specialist agents
- Anthropic stream handler and tool-call accumulator hardening
- MCP server manager reconnect and interaction improvements
- Snapshot service dedup / reference-safety fixes
- Various service cleanups: session manager, persistence, side_question,
  project context, search, workspace

Frontend changes:
- Add PendingQueuePanel: shows queued user messages above the chat input;
  supports inline edit, "send now" mid-turn steering, and delete
- Add UserSteeringBubble: renders a user-steering flow item as a normal
  user bubble inside the active model round
- Add PendingQueueModule: manages per-session pending message queue state
- Add modelRoundItemGrouping: groups consecutive flow items by model round
  for virtualised rendering
- Update FlowChatManager, EventHandlerModule, TextChunkModule, and
  MessageModule to handle steering events and pending queue lifecycle
- Update modernFlowChatStore and FlowChatContext for new item types
- Update AgentAPI with steering / queue endpoints
- Update flow-chat types with FlowUserSteeringItem
- Update AgentCompanionDesktopPet click-through and pixel-pet UI
- Update ChatInput / ChatInputPixelPet for queue-aware send behaviour
- Theme minor colour fixes across all presets
- i18n: add pending-queue and steering keys for en-US, zh-CN, zh-TW;
  remove stale error keys
- Remove deprecated SessionConfig option and related locale keys

Author: bobleer

src/apps/desktop/src/api/agentic_api.rs

@@ -156,6 +156,26 @@ pub struct CancelDialogTurnRequest {
     pub dialog_turn_id: String,
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SteerDialogTurnRequest {
+    pub session_id: String,
+    pub dialog_turn_id: String,
+    /// Rendered content delivered to the model. When omitted by the caller this
+    /// equals the displayed user text.
+    pub content: String,
+    /// Original user text for UI rendering (defaults to `content`).
+    #[serde(default)]
+    pub display_content: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SteerDialogTurnResponse {
+    pub success: bool,
+    pub steering_id: String,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct CancelSessionRequest {
@@ -615,6 +635,40 @@ pub async fn cancel_dialog_turn(
         })
 }
 
+#[tauri::command]
+pub async fn steer_dialog_turn(
+    scheduler: State<'_, Arc<DialogScheduler>>,
+    request: SteerDialogTurnRequest,
+) -> Result<SteerDialogTurnResponse, String> {
+    let SteerDialogTurnRequest {
+        session_id,
+        dialog_turn_id,
+        content,
+        display_content,
+    } = request;
+
+    let trimmed = content.trim();
+    if trimmed.is_empty() {
+        return Err("Steering content cannot be empty".to_string());
+    }
+
+    let outcome = scheduler
+        .submit_steering(session_id, dialog_turn_id, content, display_content)
+        .await
+        .map_err(|e| format!("Failed to steer dialog turn: {}", e))?;
+
+    let steering_id = match outcome {
+        bitfun_core::agentic::coordination::DialogSteerOutcome::Buffered {
+            steering_id, ..
+        } => steering_id,
+    };
+
+    Ok(SteerDialogTurnResponse {
+        success: true,
+        steering_id,
+    })
+}
+
 #[tauri::command]
 pub async fn cancel_session(
     coordinator: State<'_, Arc<ConversationCoordinator>>,

src/apps/desktop/src/api/btw_api.rs

@@ -48,7 +48,10 @@ pub async fn btw_cancel(
         return Err("requestId is required".to_string());
     }
 
-    state.side_question_runtime.cancel(&request.request_id).await;
+    state
+        .side_question_runtime
+        .cancel(&request.request_id)
+        .await;
     if let Some(active_turn) = state
         .side_question_runtime
         .get_btw_turn(&request.request_id)
@@ -58,7 +61,10 @@ pub async fn btw_cancel(
             .cancel_dialog_turn(&active_turn.session_id, &active_turn.turn_id)
             .await
             .map_err(|e| e.to_string())?;
-        state.side_question_runtime.remove(&request.request_id).await;
+        state
+            .side_question_runtime
+            .remove(&request.request_id)
+            .await;
     }
     Ok(())
 }
@@ -110,7 +116,10 @@ pub async fn btw_ask_stream(
     let coordinator = coordinator.inner().clone();
     tokio::spawn(async move {
         loop {
-            let Some(session) = coordinator.get_session_manager().get_session(&child_session_id) else {
+            let Some(session) = coordinator
+                .get_session_manager()
+                .get_session(&child_session_id)
+            else {
                 runtime.remove(&request_id).await;
                 break;
             };

src/apps/desktop/src/api/commands.rs

@@ -2028,8 +2028,8 @@ fn load_pet_manifest_from_bytes(bytes: &[u8]) -> Result<(serde_json::Value, Path
 fn load_pet_package_source(source_path: &Path) -> Result<PetPackageSource, String> {
     if source_path.is_dir() {
         let pet_json_path = source_path.join("pet.json");
-        let pet_json = std::fs::read(&pet_json_path)
-            .map_err(|e| format!("Failed to read pet.json: {}", e))?;
+        let pet_json =
+            std::fs::read(&pet_json_path).map_err(|e| format!("Failed to read pet.json: {}", e))?;
         let (_, spritesheet_name) = load_pet_manifest_from_bytes(&pet_json)?;
         let spritesheet_path = source_path.join(&spritesheet_name);
         let spritesheet = std::fs::read(&spritesheet_path)
@@ -2043,8 +2043,8 @@ fn load_pet_package_source(source_path: &Path) -> Result<PetPackageSource, Strin
 
     let file = std::fs::File::open(source_path)
         .map_err(|e| format!("Failed to open pet zip package: {}", e))?;
-    let mut archive = zip::ZipArchive::new(file)
-        .map_err(|e| format!("Failed to read pet zip package: {}", e))?;
+    let mut archive =
+        zip::ZipArchive::new(file).map_err(|e| format!("Failed to read pet zip package: {}", e))?;
 
     let mut manifest_index = None;
     for index in 0..archive.len() {
@@ -2098,15 +2098,22 @@ fn companion_user_packages_dir(state: &AppState) -> PathBuf {
         .join("agent-companions")
 }
 
-fn pet_package_dto_from_dir(dir: &Path, source: &str) -> Result<AgentCompanionPetPackageDto, String> {
+fn pet_package_dto_from_dir(
+    dir: &Path,
+    source: &str,
+) -> Result<AgentCompanionPetPackageDto, String> {
     let pet_json_path = dir.join("pet.json");
     let pet_json = std::fs::read(&pet_json_path)
         .map_err(|e| format!("Failed to read {}: {}", pet_json_path.display(), e))?;
     let (manifest, spritesheet_rel_path) = load_pet_manifest_from_bytes(&pet_json)?;
     let raw_id = manifest
         .get("id")
         .and_then(|value| value.as_str())
-        .unwrap_or_else(|| dir.file_name().and_then(|name| name.to_str()).unwrap_or("pet"));
+        .unwrap_or_else(|| {
+            dir.file_name()
+                .and_then(|name| name.to_str())
+                .unwrap_or("pet")
+        });
     let display_name = manifest
         .get("displayName")
         .and_then(|value| value.as_str())
@@ -2121,7 +2128,10 @@ fn pet_package_dto_from_dir(dir: &Path, source: &str) -> Result<AgentCompanionPe
         .filter(|value| !value.is_empty());
     let spritesheet_path = dir.join(&spritesheet_rel_path);
     if !spritesheet_path.is_file() {
-        return Err(format!("Spritesheet not found: {}", spritesheet_path.display()));
+        return Err(format!(
+            "Spritesheet not found: {}",
+            spritesheet_path.display()
+        ));
     }
     let spritesheet_file_name = spritesheet_rel_path
         .file_name()
@@ -2154,7 +2164,11 @@ fn scan_pet_package_dirs(root: &Path, source: &str) -> Vec<AgentCompanionPetPack
             Err(err) => warn!("Skipping invalid Agent companion pet package: {}", err),
         }
     }
-    pets.sort_by(|a, b| a.display_name.to_lowercase().cmp(&b.display_name.to_lowercase()));
+    pets.sort_by(|a, b| {
+        a.display_name
+            .to_lowercase()
+            .cmp(&b.display_name.to_lowercase())
+    });
     pets
 }
 
@@ -2262,7 +2276,9 @@ pub async fn delete_agent_companion_pet_package(
         .map_err(|e| format!("Pet package path not found: {}", e))?;
 
     if !resolved.starts_with(&root) {
-        return Err("Refusing to delete path outside imported Agent companion packages".to_string());
+        return Err(
+            "Refusing to delete path outside imported Agent companion packages".to_string(),
+        );
     }
     if !resolved.is_dir() {
         return Err("Pet package is not a directory".to_string());

src/apps/desktop/src/api/config_api.rs

@@ -250,9 +250,7 @@ pub async fn set_mode_config(
     )
     .await
     {
-        Ok(_) => {
-            Ok(format!("Mode '{}' configuration set successfully", mode_id))
-        }
+        Ok(_) => Ok(format!("Mode '{}' configuration set successfully", mode_id)),
         Err(e) => {
             error!(
                 "Failed to set mode config: mode_id={}, error={}",
@@ -273,12 +271,10 @@ pub async fn reset_mode_config(
     )
     .await
     {
-        Ok(_) => {
-            Ok(format!(
-                "Mode '{}' configuration reset successfully",
-                mode_id
-            ))
-        }
+        Ok(_) => Ok(format!(
+            "Mode '{}' configuration reset successfully",
+            mode_id
+        )),
         Err(e) => {
             error!(
                 "Failed to reset mode config: mode_id={}, error={}",
@@ -348,12 +344,10 @@ pub async fn set_subagent_config(
     let config_value = to_json_value(&config, "subagent config")?;
 
     match config_service.set_config(&path, config_value).await {
-        Ok(_) => {
-            Ok(format!(
-                "SubAgent '{}' configuration set successfully",
-                subagent_id
-            ))
-        }
+        Ok(_) => Ok(format!(
+            "SubAgent '{}' configuration set successfully",
+            subagent_id
+        )),
         Err(e) => {
             error!(
                 "Failed to set subagent config: subagent_id={}, enabled={}, error={}",

src/apps/desktop/src/api/search_api.rs

@@ -49,7 +49,9 @@ async fn workspace_search_unavailable_message(root_path: &str) -> Option<String>
 }
 
 pub(crate) async fn should_use_workspace_search(root_path: &str) -> bool {
-    workspace_search_unavailable_message(root_path).await.is_none()
+    workspace_search_unavailable_message(root_path)
+        .await
+        .is_none()
 }
 
 pub(crate) async fn search_file_contents_via_workspace_search(

src/apps/desktop/src/api/snapshot_service.rs

@@ -3,8 +3,8 @@
 use bitfun_core::infrastructure::try_get_path_manager_arc;
 use bitfun_core::service::remote_ssh::workspace_state::is_remote_path;
 use bitfun_core::service::snapshot::{
-    OperationType, SnapshotConfig, SnapshotManager, ensure_snapshot_manager_for_workspace,
-    get_snapshot_manager_for_workspace, initialize_snapshot_manager_for_workspace,
+    ensure_snapshot_manager_for_workspace, get_snapshot_manager_for_workspace,
+    initialize_snapshot_manager_for_workspace, OperationType, SnapshotConfig, SnapshotManager,
 };
 use log::{info, warn};
 use serde::{Deserialize, Serialize};

src/apps/desktop/src/api/tool_api.rs

@@ -6,10 +6,10 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 
 use bitfun_core::agentic::{
-    WorkspaceBinding,
     tools::framework::ToolUseContext,
     tools::{get_all_tools, get_readonly_tools},
     workspace::{local_workspace_services, remote_workspace_services},
+    WorkspaceBinding,
 };
 use bitfun_core::service::remote_ssh::workspace_state::{
     get_remote_workspace_manager, lookup_remote_connection, resolve_workspace_session_identity,

src/apps/desktop/src/api/workspace_activation.rs

@@ -1,6 +1,6 @@
 use crate::api::app_state::AppState;
-use bitfun_core::service::search::workspace_search_runtime_available;
 use bitfun_core::service::remote_ssh::workspace_state::is_remote_path;
+use bitfun_core::service::search::workspace_search_runtime_available;
 use bitfun_core::service::workspace::{WorkspaceInfo, WorkspaceKind};
 use log::{debug, info, warn};
 use std::path::{Path, PathBuf};

src/apps/desktop/src/lib.rs

@@ -141,7 +141,8 @@ pub async fn run() {
     }
     startup_timings.record_elapsed("init_function_agents", step_started);
 
-    let workspace_search_enabled = bitfun_core::service::search::workspace_search_feature_enabled().await;
+    let workspace_search_enabled =
+        bitfun_core::service::search::workspace_search_feature_enabled().await;
     let startup_flashgrep_path = configure_workspace_search_daemon_env();
 
     let step_started = Instant::now();
@@ -395,6 +396,7 @@ pub async fn run() {
             api::agentic_api::compact_session,
             api::agentic_api::ensure_assistant_bootstrap,
             api::agentic_api::cancel_dialog_turn,
+            api::agentic_api::steer_dialog_turn,
             api::agentic_api::cancel_session,
             api::agentic_api::set_subagent_timeout,
             api::agentic_api::delete_session,
@@ -870,7 +872,6 @@ async fn init_agentic_system() -> anyhow::Result<(
     Arc<bitfun_core::service::token_usage::TokenUsageService>,
 )> {
     use bitfun_core::agentic::*;
-    use bitfun_core::service::config::get_global_config_service;
 
     let ai_client_factory = AIClientFactory::get_global().await?;
 
@@ -908,27 +909,13 @@ async fn init_agentic_system() -> anyhow::Result<(
         event_queue.clone(),
         tool_pipeline.clone(),
     ));
-    
-    // Get execution config from global settings
-    let exec_config = match get_global_config_service().await {
-        Ok(config_service) => {
-            match config_service.get_config::<bitfun_core::service::config::types::GlobalConfig>(None).await {
-                Ok(global_config) => execution::ExecutionEngineConfig {
-                    max_rounds: global_config.ai.max_rounds,
-                    ..Default::default()
-                },
-                Err(_) => Default::default(),
-            }
-        },
-        Err(_) => Default::default(),
-    };
-    
+
     let execution_engine = Arc::new(execution::ExecutionEngine::new(
         round_executor,
         event_queue.clone(),
         session_manager.clone(),
         context_compressor,
-        exec_config,
+        execution::ExecutionEngineConfig::default(),
     ));
 
     let coordinator = Arc::new(coordination::ConversationCoordinator::new(
@@ -959,6 +946,7 @@ async fn init_agentic_system() -> anyhow::Result<(
         coordination::DialogScheduler::new(coordinator.clone(), session_manager.clone());
     coordinator.set_scheduler_notifier(scheduler.outcome_sender());
     coordinator.set_round_preempt_source(scheduler.preempt_monitor());
+    coordinator.set_round_steering_source(scheduler.steering_monitor());
     coordination::set_global_scheduler(scheduler.clone());
 
     let cron_service =

src/apps/desktop/src/theme.rs

@@ -358,9 +358,8 @@ fn agent_companion_default_position(
         .as_ref()
         .map(|size| size.height)
         .unwrap_or(AGENT_COMPANION_WINDOW_MIN_SIZE);
-    let x = area_position.x + area_size.width
-        - window_width
-        - f64::from(AGENT_COMPANION_WINDOW_MARGIN);
+    let x =
+        area_position.x + area_size.width - window_width - f64::from(AGENT_COMPANION_WINDOW_MARGIN);
     let y = area_position.y + area_size.height
         - window_height
         - f64::from(AGENT_COMPANION_WINDOW_MARGIN);