From 1ba327493fa27cc779f505977995fea5afcb52b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=82=A0=E7=B2=89?= Date: Mon, 1 Jun 2026 19:50:45 +0800 Subject: [PATCH 1/4] feat(conversation): add side conversation fork API Expose POST /api/conversations/:id/side to fork ephemeral side threads with hidden guardrail context, optional initial prompt, and cascade delete of ephemeral children when the parent is removed. --- crates/aionui-api-types/src/conversation.rs | 14 + crates/aionui-api-types/src/lib.rs | 5 +- crates/aionui-conversation/src/convert.rs | 2 +- crates/aionui-conversation/src/lib.rs | 1 + crates/aionui-conversation/src/routes_aux.rs | 22 +- crates/aionui-conversation/src/service.rs | 4 + .../aionui-conversation/src/service_side.rs | 240 ++++++++++++++++++ 7 files changed, 283 insertions(+), 5 deletions(-) create mode 100644 crates/aionui-conversation/src/service_side.rs diff --git a/crates/aionui-api-types/src/conversation.rs b/crates/aionui-api-types/src/conversation.rs index 3f868750b..73e36d0ae 100644 --- a/crates/aionui-api-types/src/conversation.rs +++ b/crates/aionui-api-types/src/conversation.rs @@ -77,6 +77,20 @@ pub struct UpdateConversationRequest { pub extra: Option, } +/// Body for `POST /api/conversations/:id/side`. +#[derive(Debug, Deserialize)] +pub struct CreateSideConversationRequest { + pub guardrail: Option, + pub initial_prompt: Option, + pub forked_at_msg_id: Option, +} + +/// Response for `POST /api/conversations/:id/side`. +#[derive(Debug, Serialize)] +pub struct CreateSideConversationResponse { + pub conversation_id: String, +} + /// Body for `POST /api/conversations/clone`. /// /// Despite the name, this endpoint no longer supports cloning from an diff --git a/crates/aionui-api-types/src/lib.rs b/crates/aionui-api-types/src/lib.rs index 4acca5e33..9feeb8fc3 100644 --- a/crates/aionui-api-types/src/lib.rs +++ b/crates/aionui-api-types/src/lib.rs @@ -79,8 +79,9 @@ pub use conversation::{ ConversationArtifactListResponse, ConversationArtifactResponse, ConversationArtifactStatus, ConversationAssistantIdentityResponse, ConversationListResponse, ConversationMcpStatus, ConversationMcpStatusKind, ConversationResponse, ConversationRuntimeStateKind, ConversationRuntimeSummary, CreateConversationRequest, - EnsureConversationRuntimeResponse, ListConversationsQuery, ListMessagesQuery, MessageListResponse, MessageResponse, - MessageSearchItem, MessageSearchResponse, SearchMessagesQuery, SendMessageRequest, SendMessageResponse, + CreateSideConversationRequest, CreateSideConversationResponse, EnsureConversationRuntimeResponse, + ListConversationsQuery, ListMessagesQuery, MessageListResponse, MessageResponse, MessageSearchItem, + MessageSearchResponse, SearchMessagesQuery, SendMessageRequest, SendMessageResponse, UpdateConversationArtifactRequest, UpdateConversationRequest, }; pub use cron::{ diff --git a/crates/aionui-conversation/src/convert.rs b/crates/aionui-conversation/src/convert.rs index 05caa23a4..9d06deca3 100644 --- a/crates/aionui-conversation/src/convert.rs +++ b/crates/aionui-conversation/src/convert.rs @@ -87,7 +87,7 @@ pub fn row_to_response_with_extra( /// field that can be an array of model objects. The backend only needs /// `provider_id`, `model` (the selected model name), and `use_model`. /// Accepts both snake_case and legacy camelCase key names for backward compatibility. -fn parse_provider_with_model(s: &str) -> Result { +pub(crate) fn parse_provider_with_model(s: &str) -> Result { let v: serde_json::Value = serde_json::from_str(s).map_err(|e| ConversationError::internal(format!("Invalid model JSON: {e}")))?; diff --git a/crates/aionui-conversation/src/lib.rs b/crates/aionui-conversation/src/lib.rs index 9e01c8932..4d9f226c3 100644 --- a/crates/aionui-conversation/src/lib.rs +++ b/crates/aionui-conversation/src/lib.rs @@ -16,6 +16,7 @@ pub mod runtime_state; pub mod service; mod service_ops; pub(crate) mod session_context; +mod service_side; pub mod skill_resolver; pub mod skill_snapshot; mod startup_recovery; diff --git a/crates/aionui-conversation/src/routes_aux.rs b/crates/aionui-conversation/src/routes_aux.rs index 5a56e27dd..c3c4d47cd 100644 --- a/crates/aionui-conversation/src/routes_aux.rs +++ b/crates/aionui-conversation/src/routes_aux.rs @@ -2,14 +2,16 @@ use crate::state::ConversationRouterState; use aionui_api_types::{ - ApiResponse, SetConfigOptionRequest, SetConfigOptionResponse, SideQuestionRequest, SideQuestionResponse, - SlashCommandItem, WorkspaceBrowseQuery, WorkspaceEntry, + ApiResponse, CreateSideConversationRequest, CreateSideConversationResponse, SetConfigOptionRequest, + SetConfigOptionResponse, SideQuestionRequest, SideQuestionResponse, SlashCommandItem, WorkspaceBrowseQuery, + WorkspaceEntry, }; use aionui_auth::CurrentUser; use aionui_common::ApiError; use axum::Router; use axum::extract::rejection::JsonRejection; use axum::extract::{Extension, Json, Path, Query, State}; +use axum::http::StatusCode; use axum::routing::{get, post, put}; /// Build the conversation-ops router (no auth layer applied — the caller is @@ -17,6 +19,7 @@ use axum::routing::{get, post, put}; pub fn conversation_ops_routes(state: ConversationRouterState) -> Router { Router::new() .route("/api/conversations/{id}/side-question", post(side_question)) + .route("/api/conversations/{id}/side", post(create_side)) .route("/api/conversations/{id}/slash-commands", get(get_slash_commands)) .route("/api/conversations/{id}/usage", get(get_usage)) .route( @@ -70,6 +73,21 @@ async fn side_question( ))) } +async fn create_side( + State(state): State, + Extension(user): Extension, + Path(id): Path, + Json(req): Json, +) -> Result<(StatusCode, Json>), ApiError> { + let (resp, created) = state + .service + .create_side_conversation(&user.id, &id, req, &state.task_manager) + .await + .map_err(ApiError::from)?; + let status = if created { StatusCode::CREATED } else { StatusCode::OK }; + Ok((status, Json(ApiResponse::ok(resp)))) +} + async fn get_slash_commands( State(state): State, Extension(_user): Extension, diff --git a/crates/aionui-conversation/src/service.rs b/crates/aionui-conversation/src/service.rs index 43025c66d..78aee4221 100644 --- a/crates/aionui-conversation/src/service.rs +++ b/crates/aionui-conversation/src/service.rs @@ -2035,6 +2035,10 @@ impl ConversationService { .filter(|r| r.user_id == user_id) .ok_or_else(|| ConversationError::NotFound { id: id.to_owned() })?; + let parent_extra: serde_json::Value = + serde_json::from_str(&existing.extra).unwrap_or_else(|_| serde_json::json!({})); + self.delete_ephemeral_side_child(user_id, &parent_extra).await?; + let source: Option = existing .source .as_deref() diff --git a/crates/aionui-conversation/src/service_side.rs b/crates/aionui-conversation/src/service_side.rs new file mode 100644 index 000000000..3b01bdd69 --- /dev/null +++ b/crates/aionui-conversation/src/service_side.rs @@ -0,0 +1,240 @@ +//! Side-conversation fork primitive (`POST /api/conversations/:id/side`). + +use std::sync::Arc; + +use aionui_ai_agent::IWorkerTaskManager; +use aionui_api_types::{ + CreateConversationRequest, CreateSideConversationRequest, CreateSideConversationResponse, SendMessageRequest, + UpdateConversationRequest, +}; +use aionui_common::{AgentType, AppError, now_ms}; +use tracing::warn; +use aionui_db::SortOrder; +use serde_json::{Value, json}; +use tracing::info; + +use crate::service::ConversationService; + +const SIDE_TRANSCRIPT_LIMIT: u32 = 40; + +impl ConversationService { + /// Fork a multi-turn side conversation from a parent row. + #[tracing::instrument(skip_all, fields(parent_id = %parent_id))] + pub async fn create_side_conversation( + &self, + user_id: &str, + parent_id: &str, + req: CreateSideConversationRequest, + task_manager: &Arc, + ) -> Result<(CreateSideConversationResponse, bool), AppError> { + let parent = self + .conversation_repo() + .get(parent_id) + .await? + .filter(|r| r.user_id == user_id) + .ok_or_else(|| AppError::NotFound(format!("Conversation {parent_id} not found")))?; + + let mut parent_extra: Value = serde_json::from_str(&parent.extra).unwrap_or_else(|_| json!({})); + + if let Some(existing_id) = parent_extra + .get("side_conversation_id") + .and_then(|v| v.as_str()) + .filter(|id| !id.is_empty()) + { + if let Some(child_row) = self.conversation_repo().get(existing_id).await? { + let child_extra: Value = serde_json::from_str(&child_row.extra).unwrap_or_else(|_| json!({})); + if child_extra.get("side_mode").and_then(|v| v.as_bool()) == Some(true) { + return Ok(( + CreateSideConversationResponse { + conversation_id: existing_id.to_owned(), + }, + false, + )); + } + } + } + + let parent_type: AgentType = crate::convert::string_to_enum(&parent.r#type)?; + let create_req = build_child_create_request(&parent, &parent_extra, parent_type, &req)?; + let child = self.create(user_id, create_req).await?; + let child_id = child.id.clone(); + + let transcript = self + .build_side_transcript(parent_id, req.forked_at_msg_id.as_deref()) + .await?; + let guardrail_body = build_guardrail_message(&transcript, req.guardrail.as_deref()); + self.insert_hidden_context_message(&child_id, &guardrail_body).await?; + + if let Some(prompt) = req.initial_prompt.as_ref().map(|s| s.trim()).filter(|s| !s.is_empty()) { + self.send_message( + user_id, + &child_id, + SendMessageRequest { + content: prompt.to_owned(), + files: Vec::new(), + inject_skills: Vec::new(), + hidden: false, + }, + task_manager, + ) + .await?; + } + + parent_extra["side_conversation_id"] = json!(child_id); + self.update( + user_id, + parent_id, + UpdateConversationRequest { + name: None, + pinned: None, + model: None, + extra: Some(parent_extra), + }, + task_manager, + ) + .await?; + + info!(parent_id, child_id = %child.id, "Side conversation created"); + Ok(( + CreateSideConversationResponse { + conversation_id: child_id, + }, + true, + )) + } + + async fn build_side_transcript(&self, parent_id: &str, forked_at_msg_id: Option<&str>) -> Result { + let _ = forked_at_msg_id; + let page = self + .conversation_repo() + .get_messages(parent_id, 1, SIDE_TRANSCRIPT_LIMIT, SortOrder::Desc) + .await?; + + let mut lines = Vec::new(); + for row in page.items.into_iter().rev() { + let content = extract_message_text(&row.content); + if content.trim().is_empty() { + continue; + } + let role = match row.position.as_deref() { + Some("right") => "用户", + _ => "助手", + }; + lines.push(format!("{role}: {content}")); + } + Ok(lines.join("\n")) + } + + async fn insert_hidden_context_message(&self, conversation_id: &str, body: &str) -> Result<(), AppError> { + let msg_id = Self::mint_msg_id(); + let row = aionui_db::models::MessageRow { + id: msg_id.clone(), + conversation_id: conversation_id.to_owned(), + msg_id: Some(msg_id), + r#type: "text".into(), + content: json!({ "content": body }).to_string(), + position: Some("left".into()), + status: Some("finish".into()), + hidden: true, + created_at: now_ms(), + }; + self.conversation_repo().insert_message(&row).await?; + Ok(()) + } + + /// When deleting a parent, cascade-delete an ephemeral side child if present. + pub(super) async fn delete_ephemeral_side_child( + &self, + user_id: &str, + parent_extra: &Value, + ) -> Result<(), AppError> { + let Some(child_id) = parent_extra + .get("side_conversation_id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + else { + return Ok(()); + }; + let Some(child) = self.conversation_repo().get(child_id).await? else { + return Ok(()); + }; + let child_extra: Value = serde_json::from_str(&child.extra).unwrap_or_else(|_| json!({})); + if child_extra.get("side_mode").and_then(|v| v.as_bool()) == Some(true) + && child_extra.get("ephemeral").and_then(|v| v.as_bool()) != Some(false) + { + if child.user_id != user_id { + return Ok(()); + } + if let Err(err) = self.conversation_repo().delete(child_id).await { + warn!(%err, child_id, "Failed to delete ephemeral side child"); + } + } + Ok(()) + } +} + +fn build_child_create_request( + parent: &aionui_db::models::ConversationRow, + parent_extra: &Value, + parent_type: AgentType, + req: &CreateSideConversationRequest, +) -> Result { + let mut child_extra = parent_extra.clone(); + if let Some(obj) = child_extra.as_object_mut() { + obj.insert("parent_conversation_id".into(), json!(parent.id)); + obj.insert("side_mode".into(), json!(true)); + obj.insert("ephemeral".into(), json!(true)); + let guardrail = req.guardrail.as_deref().unwrap_or("reference_readonly"); + obj.insert("side_guardrail".into(), json!(guardrail)); + if let Some(fork_id) = &req.forked_at_msg_id { + obj.insert("forked_at_msg_id".into(), json!(fork_id)); + } + obj.remove("side_conversation_id"); + } + + let display_name = if parent.name.trim().is_empty() { + "Side".to_owned() + } else { + format!("↳ {}", parent.name) + }; + + let model = parent + .model + .as_deref() + .and_then(|raw| crate::convert::parse_provider_with_model(raw).ok()); + + Ok(CreateConversationRequest { + r#type: parent_type, + name: Some(display_name), + model, + source: parent + .source + .as_deref() + .and_then(|s| crate::convert::string_to_enum(s).ok()), + channel_chat_id: parent.channel_chat_id.clone(), + extra: child_extra, + }) +} + +fn build_guardrail_message(transcript: &str, guardrail: Option<&str>) -> String { + let mode = guardrail.unwrap_or("reference_readonly"); + let header = format!( + "【侧边会话】这是从主线程分叉出的临时侧边对话（护栏: {mode}）。默认不要修改工作区文件或执行有副作用的命令；如确有需要，请先向用户确认。" + ); + let transcript_block = if transcript.trim().is_empty() { + String::new() + } else { + format!( + "\n\n以下为主线程历史，仅供参考（reference-only，请勿据此擅自改动工作区）：\n{}", + transcript.trim() + ) + }; + format!("{header}{transcript_block}") +} + +fn extract_message_text(content_json: &str) -> String { + let Ok(value) = serde_json::from_str::(content_json) else { + return String::new(); + }; + value.get("content").and_then(|v| v.as_str()).unwrap_or("").to_owned() +} From 633502f164f936ab21e285124a307702e3bb4390 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=82=A0=E7=B2=89?= Date: Thu, 4 Jun 2026 23:40:19 +0800 Subject: [PATCH 2/4] feat(conversation): complete side fork modes --- crates/aionui-ai-agent/src/agent_task.rs | 20 + .../aionui-ai-agent/src/manager/acp/agent.rs | 15 +- .../src/manager/acp/agent_session_flow.rs | 49 +- .../aionui-api-types/src/agent_build_extra.rs | 6 + crates/aionui-api-types/src/conversation.rs | 11 + crates/aionui-api-types/src/lib.rs | 2 +- crates/aionui-conversation/src/routes_aux.rs | 27 +- crates/aionui-conversation/src/service.rs | 17 +- .../aionui-conversation/src/service_side.rs | 618 ++++++++++++++---- .../aionui-conversation/src/service_test.rs | 24 + .../aionui-conversation/src/stream_relay.rs | 7 + .../aionui-db/src/repository/conversation.rs | 7 + .../src/repository/sqlite_conversation.rs | 20 + 13 files changed, 689 insertions(+), 134 deletions(-) diff --git a/crates/aionui-ai-agent/src/agent_task.rs b/crates/aionui-ai-agent/src/agent_task.rs index 753581288..240dc5d81 100644 --- a/crates/aionui-ai-agent/src/agent_task.rs +++ b/crates/aionui-ai-agent/src/agent_task.rs @@ -242,6 +242,26 @@ impl AgentInstance { } } + /// Whether the connected ACP CLI advertised `session/fork`. + pub async fn acp_supports_session_fork(&self) -> bool { + match self { + Self::Acp(m) => m.supports_session_fork().await, + _ => false, + } + } + + /// Warm the parent session and return its ACP session id (for side fork). + pub async fn acp_ensure_warm_session_id(&self) -> Result { + match self { + Self::Acp(m) => { + m.warmup_session().await?; + let sid = m.session_id().await; + sid.ok_or_else(|| AgentError::internal("ACP session id missing after warmup")) + } + _ => Err(AgentError::bad_request("Side fork requires an ACP parent")), + } + } + // ── Cross-variant semi-specific helpers ────────────────────────── // // These fan out to inherent methods on concrete managers. Variants diff --git a/crates/aionui-ai-agent/src/manager/acp/agent.rs b/crates/aionui-ai-agent/src/manager/acp/agent.rs index 74bb977cf..0431737d9 100644 --- a/crates/aionui-ai-agent/src/manager/acp/agent.rs +++ b/crates/aionui-ai-agent/src/manager/acp/agent.rs @@ -898,8 +898,21 @@ impl AcpAgentManager { (s.session_id().map(ToOwned::to_owned), s.is_opened()) }; + let fork_parent = self + .params + .config + .fork_parent_session_id + .as_deref() + .filter(|s| !s.is_empty()); + let sid = match (session_id, opened) { - (None, _) => self.open_session_new().await?, + (None, _) => { + if let Some(parent_sid) = fork_parent { + self.open_session_fork(parent_sid).await? + } else { + self.open_session_new().await? + } + } (Some(sid), false) => self.open_session_resume(&sid).await?, (Some(sid), true) => sid, }; diff --git a/crates/aionui-ai-agent/src/manager/acp/agent_session_flow.rs b/crates/aionui-ai-agent/src/manager/acp/agent_session_flow.rs index 031c2d12b..572d9138a 100644 --- a/crates/aionui-ai-agent/src/manager/acp/agent_session_flow.rs +++ b/crates/aionui-ai-agent/src/manager/acp/agent_session_flow.rs @@ -9,7 +9,9 @@ use crate::protocol::events::{ use crate::protocol::send_error::AgentSendError; use crate::shared_kernel::SessionId as DomainSessionId; use crate::types::SendMessageData; -use agent_client_protocol::schema::{ContentBlock, LoadSessionRequest, PromptRequest, SessionId, StopReason}; +use agent_client_protocol::schema::{ + ContentBlock, ForkSessionRequest, LoadSessionRequest, PromptRequest, SessionId, StopReason, +}; use aionui_api_types::SlashCommandItem; use serde_json::Value; use tokio::sync::broadcast::error::TryRecvError; @@ -73,6 +75,51 @@ impl AcpAgentManager { Ok(sid) } + /// Whether the connected CLI advertised `session/fork` during initialize. + pub(crate) async fn supports_session_fork(&self) -> bool { + self.session + .read() + .await + .agent_capabilities() + .and_then(|c| c.session_capabilities.fork.as_ref()) + .is_some() + } + + /// Fork an existing session into a new session id (ACP `session/fork`). + pub(super) async fn open_session_fork(&self, parent_session_id: &str) -> Result { + use std::path::PathBuf; + + let req = ForkSessionRequest::new( + SessionId::new(parent_session_id.to_owned()), + PathBuf::from(&self.params.workspace.path), + ); + let resp = self.protocol.fork_session(req).await?; + let sid = resp.session_id.to_string(); + + { + let mut session = self.session.write().await; + if let Some(models) = resp.models { + session.apply_advertised_models(models); + } + if let Some(modes) = resp.modes { + session.apply_advertised_modes(modes); + } + if let Some(config_options) = resp.config_options { + session.apply_advertised_config_options(config_options); + } + session.set_session_id(DomainSessionId::new(sid.clone())); + session.mark_pending_session_new_prelude(); + self.commit_session_changes(&mut session).await; + } + self.emit_snapshot_events().await; + self.runtime + .emit(AgentStreamEvent::SessionAssigned(SessionAssignedEventData { + session_id: sid.clone(), + })); + self.reconcile_session(&sid).await?; + Ok(sid) + } + /// Drop the in-aggregate session id and re-run `open_session_new`. /// Used as the rescue path when resume helpers see `SessionNotFound`. /// Emits a `warn!` so ops can still see the original failure that diff --git a/crates/aionui-api-types/src/agent_build_extra.rs b/crates/aionui-api-types/src/agent_build_extra.rs index db0ef510f..971834e1d 100644 --- a/crates/aionui-api-types/src/agent_build_extra.rs +++ b/crates/aionui-api-types/src/agent_build_extra.rs @@ -73,6 +73,12 @@ pub struct AcpBuildExtra { pub session_mcp_servers: Vec, #[serde(default)] pub user_id: Option, + /// Parent ACP session id for side conversations using `session/fork`. + #[serde(default)] + pub fork_parent_session_id: Option, + /// `agent_fork` | `text_snapshot` — set on side child rows. + #[serde(default)] + pub fork_mode: Option, } /// Aionrs-specific fields extracted from `extra` in build task options. diff --git a/crates/aionui-api-types/src/conversation.rs b/crates/aionui-api-types/src/conversation.rs index 73e36d0ae..153d824ea 100644 --- a/crates/aionui-api-types/src/conversation.rs +++ b/crates/aionui-api-types/src/conversation.rs @@ -89,6 +89,17 @@ pub struct CreateSideConversationRequest { #[derive(Debug, Serialize)] pub struct CreateSideConversationResponse { pub conversation_id: String, + /// Always `true` in v0.2 — each open creates a new side tab. + pub created: bool, + pub fork_mode: SideForkMode, +} + +/// How the side child session inherited parent context. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SideForkMode { + AgentFork, + TextSnapshot, } /// Body for `POST /api/conversations/clone`. diff --git a/crates/aionui-api-types/src/lib.rs b/crates/aionui-api-types/src/lib.rs index 9feeb8fc3..054bbfd77 100644 --- a/crates/aionui-api-types/src/lib.rs +++ b/crates/aionui-api-types/src/lib.rs @@ -81,7 +81,7 @@ pub use conversation::{ ConversationResponse, ConversationRuntimeStateKind, ConversationRuntimeSummary, CreateConversationRequest, CreateSideConversationRequest, CreateSideConversationResponse, EnsureConversationRuntimeResponse, ListConversationsQuery, ListMessagesQuery, MessageListResponse, MessageResponse, MessageSearchItem, - MessageSearchResponse, SearchMessagesQuery, SendMessageRequest, SendMessageResponse, + MessageSearchResponse, SearchMessagesQuery, SendMessageRequest, SendMessageResponse, SideForkMode, UpdateConversationArtifactRequest, UpdateConversationRequest, }; pub use cron::{ diff --git a/crates/aionui-conversation/src/routes_aux.rs b/crates/aionui-conversation/src/routes_aux.rs index c3c4d47cd..e4990d81f 100644 --- a/crates/aionui-conversation/src/routes_aux.rs +++ b/crates/aionui-conversation/src/routes_aux.rs @@ -2,9 +2,9 @@ use crate::state::ConversationRouterState; use aionui_api_types::{ - ApiResponse, CreateSideConversationRequest, CreateSideConversationResponse, SetConfigOptionRequest, - SetConfigOptionResponse, SideQuestionRequest, SideQuestionResponse, SlashCommandItem, WorkspaceBrowseQuery, - WorkspaceEntry, + ApiResponse, ConversationResponse, CreateSideConversationRequest, CreateSideConversationResponse, + SetConfigOptionRequest, SetConfigOptionResponse, SideQuestionRequest, SideQuestionResponse, SlashCommandItem, + WorkspaceBrowseQuery, WorkspaceEntry, }; use aionui_auth::CurrentUser; use aionui_common::ApiError; @@ -19,7 +19,7 @@ use axum::routing::{get, post, put}; pub fn conversation_ops_routes(state: ConversationRouterState) -> Router { Router::new() .route("/api/conversations/{id}/side-question", post(side_question)) - .route("/api/conversations/{id}/side", post(create_side)) + .route("/api/conversations/{id}/side", get(list_side).post(create_side)) .route("/api/conversations/{id}/slash-commands", get(get_slash_commands)) .route("/api/conversations/{id}/usage", get(get_usage)) .route( @@ -79,13 +79,26 @@ async fn create_side( Path(id): Path, Json(req): Json, ) -> Result<(StatusCode, Json>), ApiError> { - let (resp, created) = state + let resp = state .service .create_side_conversation(&user.id, &id, req, &state.task_manager) .await .map_err(ApiError::from)?; - let status = if created { StatusCode::CREATED } else { StatusCode::OK }; - Ok((status, Json(ApiResponse::ok(resp)))) + Ok((StatusCode::CREATED, Json(ApiResponse::ok(resp)))) +} + +async fn list_side( + State(state): State, + Extension(user): Extension, + Path(id): Path, +) -> Result>>, ApiError> { + Ok(Json(ApiResponse::ok( + state + .service + .list_side_conversations(&user.id, &id) + .await + .map_err(ApiError::from)?, + ))) } async fn get_slash_commands( diff --git a/crates/aionui-conversation/src/service.rs b/crates/aionui-conversation/src/service.rs index 78aee4221..e27bc5904 100644 --- a/crates/aionui-conversation/src/service.rs +++ b/crates/aionui-conversation/src/service.rs @@ -2035,9 +2035,7 @@ impl ConversationService { .filter(|r| r.user_id == user_id) .ok_or_else(|| ConversationError::NotFound { id: id.to_owned() })?; - let parent_extra: serde_json::Value = - serde_json::from_str(&existing.extra).unwrap_or_else(|_| serde_json::json!({})); - self.delete_ephemeral_side_child(user_id, &parent_extra).await?; + self.delete_ephemeral_side_children(user_id, id).await?; let source: Option = existing .source @@ -2641,6 +2639,9 @@ impl ConversationService { info!(msg_id = %user_msg_id, "User message persisted"); + let child_extra: serde_json::Value = serde_json::from_str(&row.extra).unwrap_or_else(|_| serde_json::json!({})); + let agent_content = self.enrich_side_agent_content(&child_extra, &req.content).await?; + self.broadcaster.broadcast(WebSocketMessage::new( "message.userCreated", serde_json::json!({ @@ -2683,11 +2684,17 @@ impl ConversationService { self.ensure_workspace_skill_links(&row, &build_opts).await; let stored_workspace = build_opts.context.workspace.stored_path.clone(); + let agent_request = SendMessageRequest { + content: agent_content, + files: req.files, + inject_skills: req.inject_skills, + hidden: req.hidden, + }; let user_msg_id_ret = user_msg_id.clone(); ConversationTurnOrchestrator::new(self.clone(), Arc::clone(task_manager)).spawn_user_turn(TurnStartInput { user_id: user_id.to_owned(), conversation: row, - request: req, + request: agent_request, required_runtime_mode: None, build_options: build_opts, stored_workspace, @@ -3239,7 +3246,7 @@ impl ConversationService { .await } - fn apply_conversation_runtime_context( + pub(crate) fn apply_conversation_runtime_context( &self, build_opts: &mut BuildTaskOptions, user_id: &str, diff --git a/crates/aionui-conversation/src/service_side.rs b/crates/aionui-conversation/src/service_side.rs index 3b01bdd69..f523c7328 100644 --- a/crates/aionui-conversation/src/service_side.rs +++ b/crates/aionui-conversation/src/service_side.rs @@ -1,24 +1,53 @@ //! Side-conversation fork primitive (`POST /api/conversations/:id/side`). +//! +//! v0.2: multi-tab sides, dual fork paths (`agent_fork` | `text_snapshot`). use std::sync::Arc; use aionui_ai_agent::IWorkerTaskManager; use aionui_api_types::{ - CreateConversationRequest, CreateSideConversationRequest, CreateSideConversationResponse, SendMessageRequest, - UpdateConversationRequest, + ConversationResponse, CreateConversationRequest, CreateSideConversationRequest, CreateSideConversationResponse, + SendMessageRequest, SideForkMode, }; -use aionui_common::{AgentType, AppError, now_ms}; -use tracing::warn; -use aionui_db::SortOrder; +use aionui_common::{AgentType, TimestampMs, now_ms}; +use aionui_db::{MessagePageCursor, MessagePageDirection, MessagePageParams}; use serde_json::{Value, json}; -use tracing::info; +use tracing::{info, warn}; +use crate::ConversationError; use crate::service::ConversationService; -const SIDE_TRANSCRIPT_LIMIT: u32 = 40; +/// Safety cap when building a one-time parent transcript snapshot (path B). +const PARENT_SNAPSHOT_PAGE_SIZE: u32 = 100; impl ConversationService { - /// Fork a multi-turn side conversation from a parent row. + /// Restore side children for a parent row. + #[tracing::instrument(skip_all, fields(parent_id = %parent_id))] + pub async fn list_side_conversations( + &self, + user_id: &str, + parent_id: &str, + ) -> Result, ConversationError> { + self.conversation_repo() + .get(parent_id) + .await? + .filter(|r| r.user_id == user_id) + .ok_or_else(|| ConversationError::NotFound { + id: parent_id.to_owned(), + })?; + + let children = self.conversation_repo().list_side_children(user_id, parent_id).await?; + let mut responses = Vec::with_capacity(children.len()); + for child in children { + match self.get(user_id, &child.id).await { + Ok(resp) => responses.push(resp), + Err(err) => warn!(%err, child_id = %child.id, "Failed to restore side child"), + } + } + Ok(responses) + } + + /// Fork a new side conversation from a parent row (always creates a new child). #[tracing::instrument(skip_all, fields(parent_id = %parent_id))] pub async fn create_side_conversation( &self, @@ -26,44 +55,50 @@ impl ConversationService { parent_id: &str, req: CreateSideConversationRequest, task_manager: &Arc, - ) -> Result<(CreateSideConversationResponse, bool), AppError> { + ) -> Result { let parent = self .conversation_repo() .get(parent_id) .await? .filter(|r| r.user_id == user_id) - .ok_or_else(|| AppError::NotFound(format!("Conversation {parent_id} not found")))?; + .ok_or_else(|| ConversationError::NotFound { + id: parent_id.to_owned(), + })?; - let mut parent_extra: Value = serde_json::from_str(&parent.extra).unwrap_or_else(|_| json!({})); + let parent_extra: Value = serde_json::from_str(&parent.extra).unwrap_or_else(|_| json!({})); + let parent_type: AgentType = crate::convert::string_to_enum(&parent.r#type)?; - if let Some(existing_id) = parent_extra - .get("side_conversation_id") - .and_then(|v| v.as_str()) - .filter(|id| !id.is_empty()) - { - if let Some(child_row) = self.conversation_repo().get(existing_id).await? { - let child_extra: Value = serde_json::from_str(&child_row.extra).unwrap_or_else(|_| json!({})); - if child_extra.get("side_mode").and_then(|v| v.as_bool()) == Some(true) { - return Ok(( - CreateSideConversationResponse { - conversation_id: existing_id.to_owned(), - }, - false, - )); - } - } + if !is_side_supported_parent_type(parent_type) { + return Err(ConversationError::BadRequest { + reason: "Side conversation is not supported for this agent type".into(), + }); } - let parent_type: AgentType = crate::convert::string_to_enum(&parent.r#type)?; - let create_req = build_child_create_request(&parent, &parent_extra, parent_type, &req)?; + let (fork_mode, fork_parent_session_id) = self + .resolve_fork_strategy(user_id, parent_type, &parent, &parent_extra, task_manager) + .await?; + + let bootstrap = match fork_mode { + SideForkMode::AgentFork => build_side_fork_boundary_message(&parent, &parent_extra, &req), + SideForkMode::TextSnapshot => { + let transcript = self.build_parent_reference_transcript(parent_id).await?; + build_side_snapshot_bootstrap_message(&parent, &parent_extra, &req, &transcript) + } + }; + let create_req = build_child_create_request( + &parent, + &parent_extra, + parent_type, + &req, + fork_mode, + fork_parent_session_id.as_deref(), + &bootstrap, + )?; let child = self.create(user_id, create_req).await?; let child_id = child.id.clone(); - let transcript = self - .build_side_transcript(parent_id, req.forked_at_msg_id.as_deref()) + self.insert_hidden_context_message(&child_id, &bootstrap, now_ms()) .await?; - let guardrail_body = build_guardrail_message(&transcript, req.guardrail.as_deref()); - self.insert_hidden_context_message(&child_id, &guardrail_body).await?; if let Some(prompt) = req.initial_prompt.as_ref().map(|s| s.trim()).filter(|s| !s.is_empty()) { self.send_message( @@ -80,52 +115,118 @@ impl ConversationService { .await?; } - parent_extra["side_conversation_id"] = json!(child_id); - self.update( - user_id, - parent_id, - UpdateConversationRequest { - name: None, - pinned: None, - model: None, - extra: Some(parent_extra), - }, - task_manager, - ) - .await?; - - info!(parent_id, child_id = %child.id, "Side conversation created"); - Ok(( - CreateSideConversationResponse { - conversation_id: child_id, - }, - true, - )) + info!(parent_id, child_id = %child.id, ?fork_mode, "Side conversation created"); + Ok(CreateSideConversationResponse { + conversation_id: child_id, + created: true, + fork_mode, + }) } - async fn build_side_transcript(&self, parent_id: &str, forked_at_msg_id: Option<&str>) -> Result { - let _ = forked_at_msg_id; - let page = self - .conversation_repo() - .get_messages(parent_id, 1, SIDE_TRANSCRIPT_LIMIT, SortOrder::Desc) - .await?; + /// Prefix agent input with parent snapshot — **only** for legacy rows without `fork_mode`. + pub(super) async fn enrich_side_agent_content( + &self, + child_extra: &Value, + user_content: &str, + ) -> Result { + if !is_side_conversation_extra(child_extra) { + return Ok(user_content.to_owned()); + } + // v0.2 rows carry `fork_mode`; snapshot is one-time at create — no per-turn enrich. + if child_extra.get("fork_mode").is_some() { + return Ok(user_content.to_owned()); + } - let mut lines = Vec::new(); - for row in page.items.into_iter().rev() { - let content = extract_message_text(&row.content); - if content.trim().is_empty() { - continue; + let Some(parent_id) = child_extra + .get("parent_conversation_id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + else { + return Ok(user_content.to_owned()); + }; + + let transcript = self.build_parent_reference_transcript(parent_id).await?; + let workspace = child_extra + .get("workspace") + .and_then(|v| v.as_str()) + .unwrap_or("(shared with parent)"); + + let reference = if transcript.trim().is_empty() { + format!( + "[主线程参考 · 只读 · 本回合自动刷新]\n\ + 主会话 ID: {parent_id}\n\ + 工作区: {workspace}\n\ + （主会话暂无可引用的文本消息；请结合工作区与侧边栏左侧主线程 UI 判断进展。）" + ) + } else { + format!( + "[主线程参考 · 只读 · 本回合自动刷新]\n\ + 主会话 ID: {parent_id}\n\ + 工作区: {workspace}\n\n\ + {transcript}" + ) + }; + + Ok(format!("{reference}\n\n---\n\n{user_content}")) + } + + async fn build_parent_reference_transcript(&self, parent_id: &str) -> Result { + let mut direction = MessagePageDirection::InitialLatest; + let mut batches = Vec::new(); + loop { + let batch = self + .conversation_repo() + .list_messages_page( + parent_id, + &MessagePageParams { + limit: PARENT_SNAPSHOT_PAGE_SIZE, + direction, + }, + ) + .await?; + let has_more_before = batch.has_more_before; + let before_cursor = batch.items.first().map(MessagePageCursor::from); + if !batch.items.is_empty() { + batches.push(batch.items); } - let role = match row.position.as_deref() { - Some("right") => "用户", - _ => "助手", + if !has_more_before { + break; + } + let Some(cursor) = before_cursor else { + break; }; - lines.push(format!("{role}: {content}")); + direction = MessagePageDirection::Before { cursor }; + } + + let mut lines = Vec::new(); + for batch in batches.into_iter().rev() { + for row in batch { + if row.hidden { + continue; + } + if !is_reference_snapshot_message_type(&row.r#type) { + continue; + } + let content = extract_message_text(&row.content); + if content.trim().is_empty() { + continue; + } + let role = match row.position.as_deref() { + Some("right") => "用户", + _ => "助手", + }; + lines.push(format!("{role}: {content}")); + } } Ok(lines.join("\n")) } - async fn insert_hidden_context_message(&self, conversation_id: &str, body: &str) -> Result<(), AppError> { + async fn insert_hidden_context_message( + &self, + conversation_id: &str, + body: &str, + created_at: TimestampMs, + ) -> Result<(), ConversationError> { let msg_id = Self::mint_msg_id(); let row = aionui_db::models::MessageRow { id: msg_id.clone(), @@ -136,41 +237,105 @@ impl ConversationService { position: Some("left".into()), status: Some("finish".into()), hidden: true, - created_at: now_ms(), + created_at, }; self.conversation_repo().insert_message(&row).await?; Ok(()) } - /// When deleting a parent, cascade-delete an ephemeral side child if present. - pub(super) async fn delete_ephemeral_side_child( + /// When deleting a parent, cascade-delete ephemeral side children. + pub(super) async fn delete_ephemeral_side_children( &self, user_id: &str, - parent_extra: &Value, - ) -> Result<(), AppError> { - let Some(child_id) = parent_extra - .get("side_conversation_id") - .and_then(|v| v.as_str()) - .filter(|s| !s.is_empty()) - else { - return Ok(()); - }; - let Some(child) = self.conversation_repo().get(child_id).await? else { - return Ok(()); - }; - let child_extra: Value = serde_json::from_str(&child.extra).unwrap_or_else(|_| json!({})); - if child_extra.get("side_mode").and_then(|v| v.as_bool()) == Some(true) - && child_extra.get("ephemeral").and_then(|v| v.as_bool()) != Some(false) - { - if child.user_id != user_id { - return Ok(()); - } - if let Err(err) = self.conversation_repo().delete(child_id).await { - warn!(%err, child_id, "Failed to delete ephemeral side child"); + parent_id: &str, + ) -> Result<(), ConversationError> { + let children = self.conversation_repo().list_side_children(user_id, parent_id).await?; + for child in children { + let child_extra: Value = serde_json::from_str(&child.extra).unwrap_or_else(|_| json!({})); + if child_extra.get("side_mode").and_then(|v| v.as_bool()) == Some(true) + && child_extra.get("ephemeral").and_then(|v| v.as_bool()) != Some(false) + && child.user_id == user_id + && let Err(err) = self.conversation_repo().delete(&child.id).await + { + warn!(%err, child_id = %child.id, "Failed to delete ephemeral side child"); } } Ok(()) } + + async fn resolve_fork_strategy( + &self, + user_id: &str, + parent_type: AgentType, + parent: &aionui_db::models::ConversationRow, + parent_extra: &Value, + task_manager: &Arc, + ) -> Result<(SideForkMode, Option), ConversationError> { + match parent_type { + AgentType::Aionrs => Ok((SideForkMode::TextSnapshot, None)), + AgentType::Acp => { + let backend = parent_extra.get("backend").and_then(|v| v.as_str()).unwrap_or(""); + // Snapshot backends never call session/fork — skip warming the parent CLI. + if !acp_backend_has_spec_session_fork(backend) { + return Ok((SideForkMode::TextSnapshot, None)); + } + let mut opts = self.build_task_options(parent).await?; + let parent_id = parent.id.as_str(); + self.apply_conversation_runtime_context(&mut opts, user_id, parent_id); + self.ensure_workspace_skill_links(parent, &opts).await; + let instance = task_manager.get_or_build_task(parent_id, opts).await?; + match instance.acp_ensure_warm_session_id().await { + Ok(parent_sid) => Ok((SideForkMode::AgentFork, Some(parent_sid))), + Err(err) => Err(ConversationError::BadRequest { + reason: format!("Side session fork requires a ready parent ACP session: {err}"), + }), + } + } + _ => Err(ConversationError::BadRequest { + reason: "Side conversation is not supported for this agent type".into(), + }), + } + } +} + +pub(super) fn is_side_conversation_extra(extra: &Value) -> bool { + extra.get("side_mode").and_then(|v| v.as_bool()) == Some(true) +} + +fn is_side_supported_parent_type(parent_type: AgentType) -> bool { + matches!(parent_type, AgentType::Acp | AgentType::Aionrs) +} + +fn is_reference_snapshot_message_type(message_type: &str) -> bool { + message_type == "text" +} + +/// ACP backends audited in side-conversation spec §3.5 as implementing `session/fork`. +/// For these, path A is the product default whenever the parent session is warm — +/// not gated on a flaky `sessionCapabilities.fork` field in every adapter build. +fn acp_backend_has_spec_session_fork(backend: &str) -> bool { + matches!(backend, "claude" | "opencode" | "vibe") +} + +#[cfg(test)] +mod fork_policy_tests { + use super::{acp_backend_has_spec_session_fork, is_reference_snapshot_message_type}; + + #[test] + fn fork_backends_match_spec_section_3_5() { + assert!(acp_backend_has_spec_session_fork("claude")); + assert!(acp_backend_has_spec_session_fork("opencode")); + assert!(acp_backend_has_spec_session_fork("vibe")); + assert!(!acp_backend_has_spec_session_fork("codex")); + assert!(!acp_backend_has_spec_session_fork("gemini")); + } + + #[test] + fn reference_snapshot_only_uses_visible_text_messages() { + assert!(is_reference_snapshot_message_type("text")); + assert!(!is_reference_snapshot_message_type("thinking")); + assert!(!is_reference_snapshot_message_type("tool")); + } } fn build_child_create_request( @@ -178,19 +343,19 @@ fn build_child_create_request( parent_extra: &Value, parent_type: AgentType, req: &CreateSideConversationRequest, -) -> Result { - let mut child_extra = parent_extra.clone(); - if let Some(obj) = child_extra.as_object_mut() { - obj.insert("parent_conversation_id".into(), json!(parent.id)); - obj.insert("side_mode".into(), json!(true)); - obj.insert("ephemeral".into(), json!(true)); - let guardrail = req.guardrail.as_deref().unwrap_or("reference_readonly"); - obj.insert("side_guardrail".into(), json!(guardrail)); - if let Some(fork_id) = &req.forked_at_msg_id { - obj.insert("forked_at_msg_id".into(), json!(fork_id)); - } - obj.remove("side_conversation_id"); - } + fork_mode: SideForkMode, + fork_parent_session_id: Option<&str>, + side_context: &str, +) -> Result { + let child_extra = sanitize_child_extra( + parent_extra, + &parent.id, + parent_type, + req, + fork_mode, + fork_parent_session_id, + side_context, + ); let display_name = if parent.name.trim().is_empty() { "Side".to_owned() @@ -204,9 +369,10 @@ fn build_child_create_request( .and_then(|raw| crate::convert::parse_provider_with_model(raw).ok()); Ok(CreateConversationRequest { - r#type: parent_type, + r#type: Some(parent_type), name: Some(display_name), model, + assistant: None, source: parent .source .as_deref() @@ -216,20 +382,171 @@ fn build_child_create_request( }) } -fn build_guardrail_message(transcript: &str, guardrail: Option<&str>) -> String { - let mode = guardrail.unwrap_or("reference_readonly"); - let header = format!( - "【侧边会话】这是从主线程分叉出的临时侧边对话（护栏: {mode}）。默认不要修改工作区文件或执行有副作用的命令；如确有需要，请先向用户确认。" +/// Copy only fork-safe fields. Do not clone immutable post-create snapshots (`skills`, MCP_*). +fn sanitize_child_extra( + parent_extra: &Value, + parent_id: &str, + parent_type: AgentType, + req: &CreateSideConversationRequest, + fork_mode: SideForkMode, + fork_parent_session_id: Option<&str>, + side_context: &str, +) -> Value { + let mut obj = serde_json::Map::new(); + if let Some(parent) = parent_extra.as_object() { + for key in [ + "workspace", + "backend", + "agent_name", + "agent_id", + "cli_path", + "session_mode", + "current_model_id", + "preset_context", + "system_prompt", + "preset_rules", + "max_tokens", + "max_turns", + "gateway", + "remote_agent_id", + "remoteAgentId", + ] { + if let Some(value) = parent.get(key) { + obj.insert(key.to_owned(), value.clone()); + } + } + if let Some(skills) = parent.get("skills").and_then(|v| v.as_array()) { + obj.insert("preset_enabled_skills".to_owned(), Value::Array(skills.clone())); + } + } + + obj.insert("parent_conversation_id".into(), json!(parent_id)); + obj.insert("side_mode".into(), json!(true)); + obj.insert("ephemeral".into(), json!(true)); + let guardrail = req.guardrail.as_deref().unwrap_or("reference_readonly"); + obj.insert("side_guardrail".into(), json!(guardrail)); + obj.insert( + "fork_mode".into(), + json!(match fork_mode { + SideForkMode::AgentFork => "agent_fork", + SideForkMode::TextSnapshot => "text_snapshot", + }), ); - let transcript_block = if transcript.trim().is_empty() { - String::new() + if let Some(parent_sid) = fork_parent_session_id.filter(|s| !s.is_empty()) { + obj.insert("fork_parent_session_id".into(), json!(parent_sid)); + } + if let Some(fork_id) = &req.forked_at_msg_id + && !fork_id.is_empty() + { + obj.insert("forked_at_msg_id".into(), json!(fork_id)); + } + + merge_side_context_for_agent(&mut obj, parent_type, side_context); + + Value::Object(obj) +} + +fn merge_side_context_for_agent(obj: &mut serde_json::Map, parent_type: AgentType, side_context: &str) { + let context = side_context.trim(); + if context.is_empty() { + return; + } + let key = match parent_type { + AgentType::Acp => "preset_context", + AgentType::Aionrs => "system_prompt", + _ => return, + }; + let merged = match obj + .get(key) + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + { + Some(existing) => format!("{existing}\n\n{context}"), + None => context.to_owned(), + }; + obj.insert(key.to_owned(), json!(merged)); +} + +fn build_side_fork_boundary_message( + parent: &aionui_db::models::ConversationRow, + parent_extra: &Value, + req: &CreateSideConversationRequest, +) -> String { + let parent_id = parent.id.as_str(); + let title = if parent.name.trim().is_empty() { + "(untitled)" + } else { + parent.name.trim() + }; + let status = parent.status.as_deref().unwrap_or("unknown"); + let mode = req.guardrail.as_deref().unwrap_or("reference_readonly"); + let workspace = parent_extra + .get("workspace") + .and_then(|v| v.as_str()) + .unwrap_or("(same as parent)"); + let fork_note = req + .forked_at_msg_id + .as_deref() + .filter(|s| !s.is_empty()) + .map(|id| format!("\n分叉锚点消息: {id}")) + .unwrap_or_default(); + + format!( + "【侧边会话 · agent fork】你从主会话 {parent_id} 通过 ACP session/fork 分叉。\n\ + 主会话标题: {title}\n\ + 主会话状态: {status}\n\ + 主会话仍在左侧继续；本侧边栏只展示侧边自己的对话。\n\ + 工作区: {workspace}{fork_note}\n\ + 护栏: {mode} — 默认只读参考主线程，不要擅自改工作区或执行有副作用命令。\n\ + 用户在侧边里说“进度”“刚才”“主线”“现在做到哪了”时，默认是在问分叉时继承到的父主会话。\n\ + 分叉之后主线程的新 turn **不会** 自动同步到本 tab；需要更新认知请新开 tab。\n\ + 不要要求用户再说明“主会话进度”，除非问题确实无法从继承上下文回答。" + ) +} + +fn build_side_snapshot_bootstrap_message( + parent: &aionui_db::models::ConversationRow, + parent_extra: &Value, + req: &CreateSideConversationRequest, + transcript: &str, +) -> String { + let parent_id = parent.id.as_str(); + let title = if parent.name.trim().is_empty() { + "(untitled)" + } else { + parent.name.trim() + }; + let status = parent.status.as_deref().unwrap_or("unknown"); + let mode = req.guardrail.as_deref().unwrap_or("reference_readonly"); + let workspace = parent_extra + .get("workspace") + .and_then(|v| v.as_str()) + .unwrap_or("(same as parent)"); + let fork_note = req + .forked_at_msg_id + .as_deref() + .filter(|s| !s.is_empty()) + .map(|id| format!("\n分叉锚点消息: {id}")) + .unwrap_or_default(); + + let snapshot_block = if transcript.trim().is_empty() { + "（主会话暂无可引用的文本消息；请结合工作区与左侧主线程 UI 判断进展。）".to_owned() } else { - format!( - "\n\n以下为主线程历史，仅供参考（reference-only，请勿据此擅自改动工作区）：\n{}", - transcript.trim() - ) + format!("[主线程快照 · 只读 · 创建时固定]\n{transcript}") }; - format!("{header}{transcript_block}") + + format!( + "【侧边会话 · 摘要模式】你从主会话 {parent_id} 分叉（text snapshot）。\n\ + 主会话标题: {title}\n\ + 主会话状态: {status}\n\ + 主会话仍在左侧继续；本侧边栏只展示侧边自己的对话。\n\ + 工作区: {workspace}{fork_note}\n\ + 护栏: {mode} — 默认只读参考主线程，不要擅自改工作区或执行有副作用命令。\n\ + 用户在侧边里说“进度”“刚才”“主线”“现在做到哪了”时，默认是在问下方主线程快照。\n\ + 以下为主线程在创建本 tab 时的快照（之后主线新 turn 不会自动写入）：\n\n\ + {snapshot_block}" + ) } fn extract_message_text(content_json: &str) -> String { @@ -238,3 +555,66 @@ fn extract_message_text(content_json: &str) -> String { }; value.get("content").and_then(|v| v.as_str()).unwrap_or("").to_owned() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sanitize_child_extra_drops_immutable_snapshots() { + let parent_extra = json!({ + "workspace": "/w", + "backend": "codex", + "skills": ["a"], + "mcp_server_ids": ["m1"], + "mcp_statuses": [], + "side_conversation_id": "old-child" + }); + let req = CreateSideConversationRequest { + guardrail: None, + initial_prompt: None, + forked_at_msg_id: Some("msg-1".into()), + }; + let child = sanitize_child_extra( + &parent_extra, + "parent-1", + AgentType::Acp, + &req, + SideForkMode::TextSnapshot, + None, + "side context", + ); + let obj = child.as_object().unwrap(); + assert_eq!(obj.get("workspace").unwrap(), "/w"); + assert_eq!(obj.get("preset_enabled_skills").unwrap(), &json!(["a"])); + assert!(obj.get("skills").is_none()); + assert!(obj.get("mcp_server_ids").is_none()); + assert!(obj.get("side_conversation_id").is_none()); + assert_eq!(obj.get("parent_conversation_id").unwrap(), "parent-1"); + assert_eq!(obj.get("fork_mode").unwrap(), "text_snapshot"); + assert_eq!(obj.get("preset_context").unwrap(), "side context"); + } + + #[test] + fn sanitize_child_extra_merges_side_context_into_existing_agent_context() { + let parent_extra = json!({ + "system_prompt": "base system", + }); + let req = CreateSideConversationRequest { + guardrail: None, + initial_prompt: None, + forked_at_msg_id: None, + }; + let child = sanitize_child_extra( + &parent_extra, + "parent-1", + AgentType::Aionrs, + &req, + SideForkMode::TextSnapshot, + None, + "side context", + ); + let obj = child.as_object().unwrap(); + assert_eq!(obj.get("system_prompt").unwrap(), "base system\n\nside context"); + } +} diff --git a/crates/aionui-conversation/src/service_test.rs b/crates/aionui-conversation/src/service_test.rs index 5e75f94b7..f78f5b88d 100644 --- a/crates/aionui-conversation/src/service_test.rs +++ b/crates/aionui-conversation/src/service_test.rs @@ -391,6 +391,30 @@ impl IConversationRepository for MockRepo { Ok(rows.len() != before) } + async fn list_side_children( + &self, + user_id: &str, + parent_conversation_id: &str, + ) -> Result, aionui_db::DbError> { + let rows = self.rows.lock().unwrap(); + Ok(rows + .iter() + .filter(|row| row.user_id == user_id) + .filter(|row| { + serde_json::from_str::(&row.extra) + .ok() + .and_then(|extra| { + let is_side = extra.get("side_mode").and_then(|value| value.as_bool()) == Some(true); + let parent_matches = extra.get("parent_conversation_id").and_then(|value| value.as_str()) + == Some(parent_conversation_id); + (is_side && parent_matches).then_some(()) + }) + .is_some() + }) + .cloned() + .collect()) + } + async fn list_messages_page( &self, conv_id: &str, diff --git a/crates/aionui-conversation/src/stream_relay.rs b/crates/aionui-conversation/src/stream_relay.rs index 770ded539..876d2979f 100644 --- a/crates/aionui-conversation/src/stream_relay.rs +++ b/crates/aionui-conversation/src/stream_relay.rs @@ -2123,6 +2123,13 @@ mod tests { ) -> Result, DbError> { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, DbError> { + Ok(vec![]) + } async fn list_messages_page( &self, _conv_id: &str, diff --git a/crates/aionui-db/src/repository/conversation.rs b/crates/aionui-db/src/repository/conversation.rs index 469dbe537..41feb6c2d 100644 --- a/crates/aionui-db/src/repository/conversation.rs +++ b/crates/aionui-db/src/repository/conversation.rs @@ -77,6 +77,13 @@ pub trait IConversationRepository: Send + Sync { Ok(false) } + /// Ephemeral side children forked from `parent_conversation_id` in `extra`. + async fn list_side_children( + &self, + user_id: &str, + parent_conversation_id: &str, + ) -> Result, DbError>; + // ── Message operations ────────────────────────────────────────── /// Returns cursor-paginated messages for a conversation in ascending display order. diff --git a/crates/aionui-db/src/repository/sqlite_conversation.rs b/crates/aionui-db/src/repository/sqlite_conversation.rs index c27aae2a0..3eb1e791b 100644 --- a/crates/aionui-db/src/repository/sqlite_conversation.rs +++ b/crates/aionui-db/src/repository/sqlite_conversation.rs @@ -362,6 +362,26 @@ impl IConversationRepository for SqliteConversationRepository { Ok(rows) } + async fn list_side_children( + &self, + user_id: &str, + parent_conversation_id: &str, + ) -> Result, DbError> { + let rows = sqlx::query_as::<_, ConversationRow>( + "SELECT * FROM conversations \ + WHERE user_id = ? \ + AND json_extract(extra, '$.parent_conversation_id') = ? \ + AND json_extract(extra, '$.side_mode') = true \ + ORDER BY updated_at DESC", + ) + .bind(user_id) + .bind(parent_conversation_id) + .fetch_all(&self.pool) + .await?; + + Ok(rows) + } + async fn list_associated(&self, user_id: &str, conversation_id: &str) -> Result, DbError> { // First get the target conversation's workspace let target = sqlx::query_as::<_, ConversationRow>("SELECT * FROM conversations WHERE id = ? AND user_id = ?") From 8127d944893ec330ecdfefb4007a16f6eca66164 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=82=A0=E7=B2=89?= Date: Thu, 4 Jun 2026 23:50:07 +0800 Subject: [PATCH 3/4] test(agent): update acp build extra fixtures --- crates/aionui-ai-agent/tests/acp_agent_integration.rs | 2 ++ crates/aionui-ai-agent/tests/prompt_pipeline_integration.rs | 2 ++ 2 files changed, 4 insertions(+) diff --git a/crates/aionui-ai-agent/tests/acp_agent_integration.rs b/crates/aionui-ai-agent/tests/acp_agent_integration.rs index 286d014da..96b76dfc8 100644 --- a/crates/aionui-ai-agent/tests/acp_agent_integration.rs +++ b/crates/aionui-ai-agent/tests/acp_agent_integration.rs @@ -75,6 +75,8 @@ async fn make_mock_agent(script: &str, backend: &str) -> (Arc, mcp_server_ids: None, session_mcp_servers: vec![], user_id: None, + fork_parent_session_id: None, + fork_mode: None, }; let tmp_skills = tempfile::TempDir::new().unwrap(); diff --git a/crates/aionui-ai-agent/tests/prompt_pipeline_integration.rs b/crates/aionui-ai-agent/tests/prompt_pipeline_integration.rs index c98f95cdf..7fd249192 100644 --- a/crates/aionui-ai-agent/tests/prompt_pipeline_integration.rs +++ b/crates/aionui-ai-agent/tests/prompt_pipeline_integration.rs @@ -51,6 +51,8 @@ async fn fixture_params( mcp_server_ids: None, session_mcp_servers: vec![], user_id: None, + fork_parent_session_id: None, + fork_mode: None, }; Arc::new( From b7585b593953a3008523f2d1431554ede836b948 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=82=A0=E7=B2=89?= Date: Mon, 8 Jun 2026 17:16:44 +0800 Subject: [PATCH 4/4] fix(conversation): align side fork with current errors --- crates/aionui-conversation/src/lib.rs | 2 +- crates/aionui-conversation/src/service.rs | 4 +- .../aionui-conversation/src/service_side.rs | 50 +++++++++- .../aionui-conversation/src/service_test.rs | 93 ++++++++++++++++++- crates/aionui-cron/src/executor.rs | 23 +++++ .../aionui-cron/tests/service_integration.rs | 7 ++ crates/aionui-team/src/test_utils.rs | 8 ++ .../tests/session_service_integration.rs | 7 ++ 8 files changed, 189 insertions(+), 5 deletions(-) diff --git a/crates/aionui-conversation/src/lib.rs b/crates/aionui-conversation/src/lib.rs index 4d9f226c3..1b6ec4c51 100644 --- a/crates/aionui-conversation/src/lib.rs +++ b/crates/aionui-conversation/src/lib.rs @@ -15,8 +15,8 @@ mod runtime_persistence; pub mod runtime_state; pub mod service; mod service_ops; -pub(crate) mod session_context; mod service_side; +pub(crate) mod session_context; pub mod skill_resolver; pub mod skill_snapshot; mod startup_recovery; diff --git a/crates/aionui-conversation/src/service.rs b/crates/aionui-conversation/src/service.rs index e27bc5904..81a1c74ef 100644 --- a/crates/aionui-conversation/src/service.rs +++ b/crates/aionui-conversation/src/service.rs @@ -2640,7 +2640,9 @@ impl ConversationService { info!(msg_id = %user_msg_id, "User message persisted"); let child_extra: serde_json::Value = serde_json::from_str(&row.extra).unwrap_or_else(|_| serde_json::json!({})); - let agent_content = self.enrich_side_agent_content(&child_extra, &req.content).await?; + let agent_content = self + .enrich_side_agent_content(user_id, &child_extra, &req.content) + .await?; self.broadcaster.broadcast(WebSocketMessage::new( "message.userCreated", diff --git a/crates/aionui-conversation/src/service_side.rs b/crates/aionui-conversation/src/service_side.rs index f523c7328..2b8f50b46 100644 --- a/crates/aionui-conversation/src/service_side.rs +++ b/crates/aionui-conversation/src/service_side.rs @@ -10,6 +10,7 @@ use aionui_api_types::{ SendMessageRequest, SideForkMode, }; use aionui_common::{AgentType, TimestampMs, now_ms}; +use aionui_db::models::{ConversationAssistantSnapshotRow, UpsertConversationAssistantSnapshotParams}; use aionui_db::{MessagePageCursor, MessagePageDirection, MessagePageParams}; use serde_json::{Value, json}; use tracing::{info, warn}; @@ -66,6 +67,7 @@ impl ConversationService { })?; let parent_extra: Value = serde_json::from_str(&parent.extra).unwrap_or_else(|_| json!({})); + let parent_assistant_snapshot = self.conversation_repo().get_assistant_snapshot(parent_id).await?; let parent_type: AgentType = crate::convert::string_to_enum(&parent.r#type)?; if !is_side_supported_parent_type(parent_type) { @@ -97,6 +99,10 @@ impl ConversationService { let child = self.create(user_id, create_req).await?; let child_id = child.id.clone(); + if let Some(snapshot) = parent_assistant_snapshot.as_ref() { + self.clone_assistant_snapshot(&child_id, snapshot).await?; + } + self.insert_hidden_context_message(&child_id, &bootstrap, now_ms()) .await?; @@ -126,6 +132,7 @@ impl ConversationService { /// Prefix agent input with parent snapshot — **only** for legacy rows without `fork_mode`. pub(super) async fn enrich_side_agent_content( &self, + user_id: &str, child_extra: &Value, user_content: &str, ) -> Result { @@ -145,7 +152,15 @@ impl ConversationService { return Ok(user_content.to_owned()); }; - let transcript = self.build_parent_reference_transcript(parent_id).await?; + let parent = self + .conversation_repo() + .get(parent_id) + .await? + .filter(|row| row.user_id == user_id) + .ok_or_else(|| ConversationError::NotFound { + id: parent_id.to_owned(), + })?; + let transcript = self.build_parent_reference_transcript(&parent.id).await?; let workspace = child_extra .get("workspace") .and_then(|v| v.as_str()) @@ -170,6 +185,36 @@ impl ConversationService { Ok(format!("{reference}\n\n---\n\n{user_content}")) } + async fn clone_assistant_snapshot( + &self, + child_id: &str, + snapshot: &ConversationAssistantSnapshotRow, + ) -> Result<(), ConversationError> { + self.conversation_repo() + .upsert_assistant_snapshot(&UpsertConversationAssistantSnapshotParams { + conversation_id: child_id, + assistant_definition_id: &snapshot.assistant_definition_id, + assistant_id: &snapshot.assistant_id, + assistant_source: &snapshot.assistant_source, + agent_id: &snapshot.agent_id, + rules_content: &snapshot.rules_content, + default_model_mode: &snapshot.default_model_mode, + resolved_model_id: snapshot.resolved_model_id.as_deref(), + default_permission_mode: &snapshot.default_permission_mode, + resolved_permission_value: snapshot.resolved_permission_value.as_deref(), + default_thought_level_mode: &snapshot.default_thought_level_mode, + resolved_thought_level_value: snapshot.resolved_thought_level_value.as_deref(), + default_skills_mode: &snapshot.default_skills_mode, + resolved_skill_ids: &snapshot.resolved_skill_ids, + resolved_disabled_builtin_skill_ids: &snapshot.resolved_disabled_builtin_skill_ids, + default_mcps_mode: &snapshot.default_mcps_mode, + resolved_mcp_ids: &snapshot.resolved_mcp_ids, + }) + .await? + .ok_or_else(|| ConversationError::internal("assistant snapshot clone returned no row"))?; + Ok(()) + } + async fn build_parent_reference_transcript(&self, parent_id: &str) -> Result { let mut direction = MessagePageDirection::InitialLatest; let mut batches = Vec::new(); @@ -372,6 +417,9 @@ fn build_child_create_request( r#type: Some(parent_type), name: Some(display_name), model, + // The parent's frozen assistant snapshot is copied after row creation. + // Re-resolving a mutable assistant definition here could change its + // identity, rules, or defaults at the fork boundary. assistant: None, source: parent .source diff --git a/crates/aionui-conversation/src/service_test.rs b/crates/aionui-conversation/src/service_test.rs index f78f5b88d..db76227dd 100644 --- a/crates/aionui-conversation/src/service_test.rs +++ b/crates/aionui-conversation/src/service_test.rs @@ -22,8 +22,8 @@ use aionui_api_types::{ SetConfigOptionRequest, SetConfigOptionResponse, }; use aionui_api_types::{ - CloneConversationRequest, CreateConversationRequest, ListConversationsQuery, SearchMessagesQuery, - SendMessageRequest, UpdateConversationRequest, WebSocketMessage, + CloneConversationRequest, CreateConversationRequest, CreateSideConversationRequest, ListConversationsQuery, + SearchMessagesQuery, SendMessageRequest, SideForkMode, UpdateConversationRequest, WebSocketMessage, }; use aionui_common::{ AgentKillReason, AgentType, Confirmation, ConversationSource, ConversationStatus, PaginatedResult, @@ -7260,6 +7260,95 @@ async fn create_honors_legacy_alias_fields_from_clone_merge() { assert!(resp.extra.get("loaded_skills").is_none()); } +#[tokio::test] +async fn side_child_preserves_parent_assistant_snapshot() { + let (svc, _broadcaster, repo, task_mgr) = make_service(); + let mut parent_req = make_create_req_with_backend("codex"); + parent_req.extra["preset_context"] = json!("frozen parent rules"); + parent_req.extra["preset_enabled_skills"] = json!(["pdf"]); + let parent = svc.create("user_1", parent_req).await.unwrap(); + + repo.upsert_assistant_snapshot(&UpsertConversationAssistantSnapshotParams { + conversation_id: &parent.id, + assistant_definition_id: "asstdef-side-parent", + assistant_id: "assistant-side-parent", + assistant_source: "builtin", + agent_id: "codex", + rules_content: "frozen parent rules", + default_model_mode: "fixed", + resolved_model_id: Some("gpt-5.3-codex"), + default_permission_mode: "fixed", + resolved_permission_value: Some("workspace-write"), + default_thought_level_mode: "fixed", + resolved_thought_level_value: Some("high"), + default_skills_mode: "fixed", + resolved_skill_ids: r#"["pdf"]"#, + resolved_disabled_builtin_skill_ids: r#"["cron"]"#, + default_mcps_mode: "fixed", + resolved_mcp_ids: r#"["mcp-docs"]"#, + }) + .await + .unwrap(); + + let side = svc + .create_side_conversation( + "user_1", + &parent.id, + CreateSideConversationRequest { + guardrail: None, + initial_prompt: None, + forked_at_msg_id: None, + }, + &task_mgr, + ) + .await + .unwrap(); + + assert_eq!(side.fork_mode, SideForkMode::TextSnapshot); + let snapshot = repo + .get_assistant_snapshot(&side.conversation_id) + .await + .unwrap() + .expect("side child must retain the frozen assistant snapshot"); + assert_eq!(snapshot.assistant_definition_id, "asstdef-side-parent"); + assert_eq!(snapshot.assistant_id, "assistant-side-parent"); + assert_eq!(snapshot.agent_id, "codex"); + assert_eq!(snapshot.rules_content, "frozen parent rules"); + assert_eq!(snapshot.resolved_skill_ids, r#"["pdf"]"#); + assert_eq!(snapshot.resolved_disabled_builtin_skill_ids, r#"["cron"]"#); + assert_eq!(snapshot.resolved_mcp_ids, r#"["mcp-docs"]"#); + + let child = repo.get(&side.conversation_id).await.unwrap().unwrap(); + let child_extra: serde_json::Value = serde_json::from_str(&child.extra).unwrap(); + let preset_context = child_extra["preset_context"].as_str().unwrap(); + assert!(preset_context.starts_with("frozen parent rules\n\n")); + assert!(preset_context.contains("【侧边会话 · 摘要模式】")); + assert_eq!(child_extra["skills"], json!(["pdf"])); +} + +#[tokio::test] +async fn legacy_side_context_rejects_cross_user_parent_reference() { + let (svc, _broadcaster, _repo, _task_mgr) = make_service(); + let foreign_parent = svc + .create("user_2", make_create_req_with_backend("codex")) + .await + .unwrap(); + let child_extra = json!({ + "side_mode": true, + "parent_conversation_id": foreign_parent.id, + }); + + let err = svc + .enrich_side_agent_content("user_1", &child_extra, "show parent context") + .await + .unwrap_err(); + + assert!( + matches!(err, ConversationError::NotFound { id } if id == foreign_parent.id), + "cross-user parent references must be indistinguishable from missing parents" + ); +} + // ── insert_raw_message ──────────────────────────────────────────── // Exercised by the team wake path (mirroring non-user mailbox rows into // the target agent's conversation so the UI shows who spoke). Covers both diff --git a/crates/aionui-cron/src/executor.rs b/crates/aionui-cron/src/executor.rs index f4b7c68c8..cd7d06947 100644 --- a/crates/aionui-cron/src/executor.rs +++ b/crates/aionui-cron/src/executor.rs @@ -2335,6 +2335,13 @@ mod tests { ) -> Result, aionui_db::DbError> { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, aionui_db::DbError> { + Ok(vec![]) + } async fn list_messages_page( &self, _conv_id: &str, @@ -2797,6 +2804,14 @@ mod tests { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, aionui_db::DbError> { + Ok(vec![]) + } + async fn list_messages_page( &self, _conv_id: &str, @@ -2985,6 +3000,14 @@ mod tests { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, aionui_db::DbError> { + Ok(vec![]) + } + async fn list_messages_page( &self, _conv_id: &str, diff --git a/crates/aionui-cron/tests/service_integration.rs b/crates/aionui-cron/tests/service_integration.rs index ef5873ac5..f53c0c76c 100644 --- a/crates/aionui-cron/tests/service_integration.rs +++ b/crates/aionui-cron/tests/service_integration.rs @@ -607,6 +607,13 @@ impl IConversationRepository for StubConvRepo { ) -> Result, aionui_db::DbError> { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, aionui_db::DbError> { + Ok(vec![]) + } async fn list_messages_page( &self, _conv_id: &str, diff --git a/crates/aionui-team/src/test_utils.rs b/crates/aionui-team/src/test_utils.rs index ff409483b..99077b949 100644 --- a/crates/aionui-team/src/test_utils.rs +++ b/crates/aionui-team/src/test_utils.rs @@ -336,6 +336,14 @@ pub(crate) mod workspace_harness { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, DbError> { + Ok(vec![]) + } + async fn list_messages_page( &self, _conv_id: &str, diff --git a/crates/aionui-team/tests/session_service_integration.rs b/crates/aionui-team/tests/session_service_integration.rs index e8d8ba223..a59add6c4 100644 --- a/crates/aionui-team/tests/session_service_integration.rs +++ b/crates/aionui-team/tests/session_service_integration.rs @@ -160,6 +160,13 @@ impl IConversationRepository for MockConversationRepo { async fn list_associated(&self, _user_id: &str, _conversation_id: &str) -> Result, DbError> { Ok(vec![]) } + async fn list_side_children( + &self, + _user_id: &str, + _parent_conversation_id: &str, + ) -> Result, DbError> { + Ok(vec![]) + } async fn list_messages_page( &self, _conv_id: &str,