diff --git a/Cargo.lock b/Cargo.lock index 8ba8603e3..37f2dee1e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -918,7 +918,9 @@ dependencies = [ "async-trait", "axum", "dashmap", + "futures", "futures-util", + "http-body-util", "regex", "reqwest", "serde", @@ -927,6 +929,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tower", "tracing", ] diff --git a/crates/aionui-api-types/src/lib.rs b/crates/aionui-api-types/src/lib.rs index 3175f1279..c3d9d5eac 100644 --- a/crates/aionui-api-types/src/lib.rs +++ b/crates/aionui-api-types/src/lib.rs @@ -157,9 +157,10 @@ pub use system::{ UpdateClientPreferencesRequest, UpdateSettingsRequest, }; pub use team::{ - AddAgentRequest, CancelTeamChildTurnRequest, CancelTeamRunRequest, CreateTeamRequest, PauseTeamSlotRequest, - RenameAgentRequest, RenameTeamRequest, SendAgentMessageRequest, SendTeamMessageRequest, TeamAgentInput, - TeamAgentRemovedPayload, TeamAgentRenamedPayload, TeamAgentResponse, TeamAgentRuntimeStatus, + AdHocTeamAssociationResponse, AdHocTeamAssociationStatus, AdHocTeamFromConversationResponse, AddAgentRequest, + CancelTeamChildTurnRequest, CancelTeamRunRequest, CreateAdHocTeamFromConversationRequest, CreateTeamRequest, + PauseTeamSlotRequest, RenameAgentRequest, RenameTeamRequest, SendAgentMessageRequest, SendTeamMessageRequest, + TeamAgentInput, TeamAgentRemovedPayload, TeamAgentRenamedPayload, TeamAgentResponse, TeamAgentRuntimeStatus, TeamAgentRuntimeStatusPayload, TeamAgentSpawnedPayload, TeamAgentStatusPayload, TeamChildTurnPayload, TeamListResponse, TeamMcpRuntimeConfig, TeamMessageEnqueueStatus, TeamResponse, TeamRunAckResponse, TeamRunPayload, TeamRunSource, TeamRunStateResponse, TeamRunStatus, TeamRunTargetRole, TeamRuntimeSeed, diff --git a/crates/aionui-api-types/src/team.rs b/crates/aionui-api-types/src/team.rs index c544865fc..cf4967fb1 100644 --- a/crates/aionui-api-types/src/team.rs +++ b/crates/aionui-api-types/src/team.rs @@ -65,6 +65,51 @@ impl<'de> Deserialize<'de> for TeamAgentInput { } } +/// Request body for `POST /api/teams/from-conversation`. +/// +/// Creates an ad-hoc team from an existing conversation. The source conversation +/// becomes the team's origin; its assistant is promoted to the team lead. A target +/// assistant may be added as a teammate. +#[derive(Debug, Deserialize)] +pub struct CreateAdHocTeamFromConversationRequest { + pub conversation_id: String, + pub user_id: String, + #[serde(default)] + pub target_assistant_id: Option, + #[serde(default)] + pub name: Option, + #[serde(default)] + pub workspace_mode: Option, +} + +/// Response body for `POST /api/teams/from-conversation`. +#[derive(Debug, Serialize)] +pub struct AdHocTeamFromConversationResponse { + pub team_id: String, + pub origin_conversation_id: String, + pub leader_slot_id: String, + pub target_slot_id: Option, + pub created: bool, +} + +/// Association status of an ad-hoc team created from a conversation. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum AdHocTeamAssociationStatus { + Active, + Disbanded, +} + +/// Response body for `GET /api/teams/by-conversation`. +#[derive(Debug, Serialize)] +pub struct AdHocTeamAssociationResponse { + pub team_id: String, + pub origin_conversation_id: String, + pub status: AdHocTeamAssociationStatus, + #[serde(skip_serializing_if = "Option::is_none")] + pub team: Option, +} + /// Request body for `POST /api/teams`. /// /// Creates a team with the given name and agent list. @@ -461,6 +506,8 @@ pub struct TeamResponse { pub assistants: Vec, #[serde(skip_serializing_if = "Option::is_none", alias = "lead_agent_id")] pub leader_assistant_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub origin_conversation_id: Option, pub created_at: TimestampMs, pub updated_at: TimestampMs, } @@ -989,6 +1036,7 @@ mod tests { pending_confirmations: 0, }], leader_assistant_id: Some("slot-1".into()), + origin_conversation_id: Some("origin-1".into()), created_at: 1700000000000, updated_at: 1700001000000, }; @@ -997,6 +1045,7 @@ mod tests { assert_eq!(json["name"], "Alpha"); assert_eq!(json["workspace"], "/workspace/team-1"); assert_eq!(json["leader_assistant_id"], "slot-1"); + assert_eq!(json["origin_conversation_id"], "origin-1"); assert_eq!(json["created_at"], 1700000000000_i64); assert_eq!(json["updated_at"], 1700001000000_i64); assert_eq!(json["assistants"].as_array().unwrap().len(), 1); @@ -1011,11 +1060,13 @@ mod tests { workspace: String::new(), assistants: vec![], leader_assistant_id: None, + origin_conversation_id: None, created_at: 1700000000000, updated_at: 1700000000000, }; let json = serde_json::to_value(&team).unwrap(); assert!(json.get("leader_assistant_id").is_none()); + assert!(json.get("origin_conversation_id").is_none()); assert!(json["assistants"].as_array().unwrap().is_empty()); } @@ -1145,6 +1196,7 @@ mod tests { }, ], leader_assistant_id: Some("s1".into()), + origin_conversation_id: Some("origin-conv".into()), created_at: 1000, updated_at: 2000, }; diff --git a/crates/aionui-app/src/router/team_conversation_adapters.rs b/crates/aionui-app/src/router/team_conversation_adapters.rs index 4e6ea5662..2cf4bb816 100644 --- a/crates/aionui-app/src/router/team_conversation_adapters.rs +++ b/crates/aionui-app/src/router/team_conversation_adapters.rs @@ -238,6 +238,49 @@ impl TeamConversationProvisioningPort for TeamConversationAdapters { .map(str::to_owned)) } + async fn conversation_metadata( + &self, + conversation_id: &str, + ) -> Result, TeamError> { + let Some(row) = self.conversation_repo.get(conversation_id).await? else { + return Ok(None); + }; + + let assistant_id = self.conversation_assistant_id(conversation_id).await?; + let extra: serde_json::Value = serde_json::from_str(&row.extra).unwrap_or(serde_json::Value::Null); + let workspace = extra + .get("workspace") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned); + let (backend, model) = row + .model + .as_deref() + .and_then(|model_json| serde_json::from_str::(model_json).ok()) + .map(|model_value| { + let backend = model_value + .get("provider_id") + .and_then(serde_json::Value::as_str) + .map(str::to_owned); + let model = model_value + .get("model") + .and_then(serde_json::Value::as_str) + .map(str::to_owned); + (backend, model) + }) + .unwrap_or_default(); + + Ok(Some(aionui_team::provisioning::ConversationMetadata { + conversation_id: row.id, + user_id: row.user_id, + assistant_id, + backend, + model, + workspace, + })) + } + async fn create_team_temp_workspace(&self, team_id: &str) -> Result { self.conversation_service .create_team_temp_workspace(team_id) diff --git a/crates/aionui-app/tests/team_e2e.rs b/crates/aionui-app/tests/team_e2e.rs index 423e87eda..d1e3f3efc 100644 --- a/crates/aionui-app/tests/team_e2e.rs +++ b/crates/aionui-app/tests/team_e2e.rs @@ -9,8 +9,8 @@ use tower::ServiceExt; use aionui_api_types::TeamMcpStdioConfig; use aionui_team::mcp::protocol::{read_frame, write_frame}; use common::{ - body_json, build_app, build_app_with_mock_agents, delete_with_token, get_request, get_with_token, json_with_token, - setup_and_login, + body_json, build_app, build_app_with_mock_agents, delete_with_token, extract_csrf_token, get_request, + get_with_token, json_with_token, setup_and_login, }; const DEFAULT_TEAM_ASSISTANT_ID: &str = "team-e2e-assistant"; @@ -1479,3 +1479,212 @@ async fn full_team_lifecycle() { let json = body_json(resp).await; assert!(json["data"].as_array().unwrap().is_empty()); } + +async fn seed_conversation_with_assistant( + services: &aionui_app::AppServices, + username: &str, + conversation_id: &str, + assistant_id: &str, +) { + let user_id: String = sqlx::query_scalar("SELECT id FROM users WHERE username = ?") + .bind(username) + .fetch_one(services.database.pool()) + .await + .expect("lookup user id"); + let now = aionui_common::now_ms(); + let extra = serde_json::json!({ "assistant_id": assistant_id }).to_string(); + let agent_type = aionui_common::AgentType::Acp.serde_name(); + sqlx::query( + "INSERT INTO conversations \ + (id, user_id, name, type, extra, model, status, pinned, created_at, updated_at) \ + VALUES (?, ?, ?, ?, ?, ?, 'running', 0, ?, ?)", + ) + .bind(conversation_id) + .bind(user_id) + .bind("Source Conversation") + .bind(agent_type) + .bind(extra) + .bind(None::) + .bind(now) + .bind(now) + .execute(services.database.pool()) + .await + .expect("seed conversation"); +} + +async fn ensure_target_team_assistant( + app: &mut axum::Router, + services: &aionui_app::AppServices, + token: &str, + csrf: &str, +) { + ensure_default_team_agent_installed(services).await; + let req = json_with_token( + "POST", + "/api/assistants", + json!({ + "id": "team-e2e-target-assistant", + "name": "Team E2E Target Assistant", + "agent_id": DEFAULT_TEAM_AGENT_ID + }), + token, + csrf, + ); + let resp = app.clone().oneshot(req).await.unwrap(); + assert!( + resp.status() == StatusCode::CREATED || resp.status() == StatusCode::CONFLICT, + "expected target assistant seed to be created or already exist, got {}", + resp.status() + ); +} + +#[tokio::test] +async fn adc1_create_ad_hoc_team_from_conversation() { + let (mut app, services) = build_app().await; + let (token, csrf) = setup_and_login(&mut app, &services, "admin", "StrongP@ss1").await; + ensure_default_team_assistant(&mut app, &services, &token, &csrf).await; + ensure_target_team_assistant(&mut app, &services, &token, &csrf).await; + seed_conversation_with_assistant(&services, "admin", "conv-adhoc-source", DEFAULT_TEAM_ASSISTANT_ID).await; + + let req = json_with_token( + "POST", + "/api/teams/from-conversation", + json!({ + "conversation_id": "conv-adhoc-source", + "user_id": "admin", + "target_assistant_id": "team-e2e-target-assistant", + "name": "Ad-hoc HTTP Team", + "workspace_mode": "shared" + }), + &token, + &csrf, + ); + let resp = app.clone().oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + let json = body_json(resp).await; + assert!(json["success"].as_bool().unwrap()); + let data = json["data"].as_object().unwrap(); + assert_eq!(data["origin_conversation_id"], "conv-adhoc-source"); + assert!(data["created"].as_bool().unwrap()); + assert!(!data["leader_slot_id"].as_str().unwrap().is_empty()); + assert!(!data["target_slot_id"].as_str().unwrap().is_empty()); +} + +#[tokio::test] +async fn adg1_get_ad_hoc_team_by_conversation() { + let (mut app, services) = build_app().await; + let (token, csrf) = setup_and_login(&mut app, &services, "admin", "StrongP@ss1").await; + ensure_default_team_assistant(&mut app, &services, &token, &csrf).await; + ensure_target_team_assistant(&mut app, &services, &token, &csrf).await; + seed_conversation_with_assistant(&services, "admin", "conv-adhoc-get", DEFAULT_TEAM_ASSISTANT_ID).await; + + let create_req = json_with_token( + "POST", + "/api/teams/from-conversation", + json!({ + "conversation_id": "conv-adhoc-get", + "user_id": "admin", + "target_assistant_id": "team-e2e-target-assistant" + }), + &token, + &csrf, + ); + let create_resp = app.clone().oneshot(create_req).await.unwrap(); + assert_eq!(create_resp.status(), StatusCode::CREATED); + + let req = get_with_token("/api/teams/by-conversation?conversation_id=conv-adhoc-get", &token); + let resp = app.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let json = body_json(resp).await; + assert!(json["success"].as_bool().unwrap()); + let data = json["data"].as_object().unwrap(); + assert_eq!(data["origin_conversation_id"], "conv-adhoc-get"); + assert!(data["team"].is_object()); + assert!(!data["team_id"].as_str().unwrap().is_empty()); +} + +#[tokio::test] +async fn adg2_get_ad_hoc_team_by_conversation_empty() { + let (mut app, services) = build_app().await; + let (token, _csrf) = setup_and_login(&mut app, &services, "admin", "StrongP@ss1").await; + + let req = get_with_token("/api/teams/by-conversation?conversation_id=conv-adhoc-none", &token); + let resp = app.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let json = body_json(resp).await; + assert!(json["success"].as_bool().unwrap()); + let data = json["data"].as_object().unwrap(); + assert!(data.get("team").is_none()); + assert_eq!(data["team_id"].as_str().unwrap(), ""); + assert_eq!(data["origin_conversation_id"], "conv-adhoc-none"); +} + +#[tokio::test] +async fn adu1_unauthenticated_ad_hoc_endpoints_return_401() { + let (app, _services) = build_app().await; + + let resp = app.clone().oneshot(get_request("/api/auth/status")).await.unwrap(); + let csrf = extract_csrf_token(&resp).expect("CSRF cookie should be set"); + + let create_req = axum::http::Request::builder() + .method("POST") + .uri("/api/teams/from-conversation") + .header("content-type", "application/json") + .header("x-csrf-token", &csrf) + .header("cookie", format!("aionui-csrf-token={csrf}")) + .body(axum::body::Body::from(r#"{"conversation_id":"x","user_id":"x"}"#)) + .unwrap(); + let resp = app.clone().oneshot(create_req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + + let get_req = get_request("/api/teams/by-conversation?conversation_id=x"); + let resp = app.oneshot(get_req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} + +#[tokio::test] +async fn adx1_ad_hoc_endpoints_reject_cross_user_access() { + let (mut app, services) = build_app().await; + let (owner_token, owner_csrf) = setup_and_login(&mut app, &services, "admin", "StrongP@ss1").await; + let (other_token, other_csrf) = setup_and_login(&mut app, &services, "alice", "StrongP@ss2").await; + ensure_default_team_assistant(&mut app, &services, &owner_token, &owner_csrf).await; + ensure_target_team_assistant(&mut app, &services, &owner_token, &owner_csrf).await; + seed_conversation_with_assistant(&services, "admin", "conv-adhoc-cross", DEFAULT_TEAM_ASSISTANT_ID).await; + + let create_req = json_with_token( + "POST", + "/api/teams/from-conversation", + json!({ + "conversation_id": "conv-adhoc-cross", + "user_id": "admin", + "target_assistant_id": "team-e2e-target-assistant" + }), + &owner_token, + &owner_csrf, + ); + let create_resp = app.clone().oneshot(create_req).await.unwrap(); + assert_eq!(create_resp.status(), StatusCode::CREATED); + + let req = get_with_token( + "/api/teams/by-conversation?conversation_id=conv-adhoc-cross", + &other_token, + ); + let resp = app.clone().oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let json = body_json(resp).await; + assert!(json["data"].as_object().unwrap().get("team").is_none()); + + let req = json_with_token( + "POST", + "/api/teams/from-conversation", + json!({ + "conversation_id": "conv-adhoc-cross", + "user_id": "alice", + "target_assistant_id": "team-e2e-target-assistant" + }), + &other_token, + &other_csrf, + ); + let resp = app.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); +} diff --git a/crates/aionui-common/src/constants.rs b/crates/aionui-common/src/constants.rs index 530535862..fe0620936 100644 --- a/crates/aionui-common/src/constants.rs +++ b/crates/aionui-common/src/constants.rs @@ -92,6 +92,16 @@ fn mcp_capability_object(agent_capabilities: Option<&serde_json::Value>) -> Opti caps.get("mcp_capabilities") .or_else(|| caps.get("mcpCapabilities")) .or_else(|| caps.get("mcp")) + .or_else(|| { + caps.get("agent_capabilities") + .or_else(|| caps.get("agentCapabilities")) + .and_then(|nested| { + nested + .get("mcp_capabilities") + .or_else(|| nested.get("mcpCapabilities")) + .or_else(|| nested.get("mcp")) + }) + }) } fn bool_field(value: &serde_json::Value, key: &str) -> bool { @@ -124,6 +134,9 @@ mod tests { assert!(has_mcp_capability(Some(&json!({ "mcp": { "http": false, "sse": true } })))); + assert!(has_mcp_capability(Some(&json!({ + "agentCapabilities": { "mcpCapabilities": { "http": true, "sse": true } } + })))); } #[test] @@ -153,6 +166,12 @@ mod tests { )); assert!(!supports_team_mcp("acp", None)); assert!(!supports_team_mcp("claude", Some(&json!({ "mcp_capabilities": {} })))); + assert!(supports_team_mcp( + "grok", + Some(&json!({ + "agentCapabilities": { "mcpCapabilities": { "http": true, "sse": true } } + })) + )); } #[test] diff --git a/crates/aionui-db/migrations/028_ad_hoc_team_origin_conversation.sql b/crates/aionui-db/migrations/028_ad_hoc_team_origin_conversation.sql new file mode 100644 index 000000000..4fa7d62b5 --- /dev/null +++ b/crates/aionui-db/migrations/028_ad_hoc_team_origin_conversation.sql @@ -0,0 +1,9 @@ +-- Migration 028: Add origin_conversation_id to teams for ad-hoc teams from conversations + +ALTER TABLE teams ADD COLUMN origin_conversation_id TEXT; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_teams_origin_conversation_id + ON teams(origin_conversation_id); + +CREATE INDEX IF NOT EXISTS idx_teams_user_origin_conversation + ON teams(user_id, origin_conversation_id); diff --git a/crates/aionui-db/src/models/team.rs b/crates/aionui-db/src/models/team.rs index 4412f30d6..91c9b818f 100644 --- a/crates/aionui-db/src/models/team.rs +++ b/crates/aionui-db/src/models/team.rs @@ -16,6 +16,7 @@ pub struct TeamRow { pub lead_agent_id: Option, pub session_mode: Option, pub agents_version: String, + pub origin_conversation_id: Option, pub created_at: TimestampMs, pub updated_at: TimestampMs, } @@ -80,6 +81,7 @@ mod tests { lead_agent_id: None, session_mode: None, agents_version: "1.0.1".into(), + origin_conversation_id: None, created_at: 0, updated_at: 0, }; diff --git a/crates/aionui-db/src/repository/sqlite_team.rs b/crates/aionui-db/src/repository/sqlite_team.rs index 63990280c..e71803159 100644 --- a/crates/aionui-db/src/repository/sqlite_team.rs +++ b/crates/aionui-db/src/repository/sqlite_team.rs @@ -17,14 +17,18 @@ impl SqliteTeamRepository { } } +fn is_unique_violation(err: &dyn sqlx::error::DatabaseError) -> bool { + err.code().is_some_and(|c| c == "2067" || c == "1555") +} + #[async_trait::async_trait] impl ITeamRepository for SqliteTeamRepository { // ── Team CRUD ──────────────────────────────────────────────────── async fn create_team(&self, row: &TeamRow) -> Result<(), DbError> { sqlx::query( - "INSERT INTO teams (id, user_id, name, workspace, workspace_mode, agents, lead_agent_id, session_mode, agents_version, created_at, updated_at) \ - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO teams (id, user_id, name, workspace, workspace_mode, agents, lead_agent_id, session_mode, agents_version, origin_conversation_id, created_at, updated_at) \ + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&row.id) .bind(&row.user_id) @@ -35,10 +39,17 @@ impl ITeamRepository for SqliteTeamRepository { .bind(&row.lead_agent_id) .bind(&row.session_mode) .bind(&row.agents_version) + .bind(&row.origin_conversation_id) .bind(row.created_at) .bind(row.updated_at) .execute(&self.pool) - .await?; + .await + .map_err(|e| match &e { + sqlx::Error::Database(db_err) if is_unique_violation(db_err.as_ref()) => { + DbError::Conflict(format!("team with id '{}' or origin conversation id {:?} already exists", row.id, row.origin_conversation_id)) + } + _ => DbError::Query(e), + })?; Ok(()) } @@ -65,6 +76,19 @@ impl ITeamRepository for SqliteTeamRepository { Ok(row) } + async fn get_team_by_origin_conversation_id( + &self, + user_id: &str, + origin_conversation_id: &str, + ) -> Result, DbError> { + let row = sqlx::query_as::<_, TeamRow>("SELECT * FROM teams WHERE user_id = ? AND origin_conversation_id = ?") + .bind(user_id) + .bind(origin_conversation_id) + .fetch_optional(&self.pool) + .await?; + Ok(row) + } + async fn update_team(&self, team_id: &str, params: &UpdateTeamParams) -> Result<(), DbError> { let mut set_clauses = Vec::new(); if params.name.is_some() { @@ -82,6 +106,9 @@ impl ITeamRepository for SqliteTeamRepository { if params.session_mode.is_some() { set_clauses.push("session_mode = ?"); } + if params.origin_conversation_id.is_some() { + set_clauses.push("origin_conversation_id = ?"); + } if set_clauses.is_empty() { return Ok(()); @@ -106,10 +133,19 @@ impl ITeamRepository for SqliteTeamRepository { if let Some(ref session_mode) = params.session_mode { query = query.bind(session_mode); } + if let Some(ref origin_conversation_id) = params.origin_conversation_id { + query = query.bind(origin_conversation_id); + } query = query.bind(now_ms()); query = query.bind(team_id); - let result = query.execute(&self.pool).await?; + let result = query.execute(&self.pool).await.map_err(|e| match &e { + sqlx::Error::Database(db_err) if is_unique_violation(db_err.as_ref()) => DbError::Conflict(format!( + "team with id '{}' or origin conversation id {:?} already exists", + team_id, params.origin_conversation_id + )), + _ => DbError::Query(e), + })?; if result.rows_affected() == 0 { return Err(DbError::NotFound(format!("team {team_id}"))); } diff --git a/crates/aionui-db/src/repository/team.rs b/crates/aionui-db/src/repository/team.rs index e7b1c2095..3ff9b26d5 100644 --- a/crates/aionui-db/src/repository/team.rs +++ b/crates/aionui-db/src/repository/team.rs @@ -9,6 +9,7 @@ pub struct UpdateTeamParams { pub agents: Option, pub lead_agent_id: Option, pub session_mode: Option, + pub origin_conversation_id: Option, } /// Parameters for updating a task record. @@ -42,6 +43,13 @@ pub trait ITeamRepository: Send + Sync { /// Returns a single team by id, or `None` if not found. async fn get_team(&self, team_id: &str) -> Result, DbError>; + /// Returns a single team by origin conversation id, or `None` if not found. + async fn get_team_by_origin_conversation_id( + &self, + user_id: &str, + origin_conversation_id: &str, + ) -> Result, DbError>; + /// Updates a team by id with the provided fields. /// Returns `DbError::NotFound` if absent. async fn update_team(&self, team_id: &str, params: &UpdateTeamParams) -> Result<(), DbError>; diff --git a/crates/aionui-db/tests/team_repository.rs b/crates/aionui-db/tests/team_repository.rs index f5cf8aa90..3a64cb2c5 100644 --- a/crates/aionui-db/tests/team_repository.rs +++ b/crates/aionui-db/tests/team_repository.rs @@ -41,6 +41,7 @@ fn make_team_for_user(id: &str, user_id: &str, name: &str) -> TeamRow { lead_agent_id: Some("a1".into()), session_mode: None, agents_version: "1.0.1".into(), + origin_conversation_id: None, created_at: now, updated_at: now, } @@ -99,6 +100,26 @@ async fn get_nonexistent_team_returns_none() { assert!(result.is_none()); } +#[tokio::test] +async fn get_team_by_origin_conversation_id_returns_matching_team() { + let (repo, _db) = repo().await; + let mut team = make_team_for_user("t1", "user-a", "Team Alpha"); + team.origin_conversation_id = Some("conv-origin".into()); + repo.create_team(&team).await.unwrap(); + + let found = repo + .get_team_by_origin_conversation_id("user-a", "conv-origin") + .await + .unwrap() + .expect("team exists"); + assert_eq!(found.id, "t1"); + + let not_found = repo + .get_team_by_origin_conversation_id("user-b", "conv-origin") + .await + .unwrap(); + assert!(not_found.is_none()); +} #[tokio::test] async fn list_teams_empty() { let (repo, _db) = repo().await; @@ -249,6 +270,67 @@ async fn delete_nonexistent_team_returns_not_found() { assert!(matches!(result, Err(DbError::NotFound(_)))); } +#[tokio::test] +async fn create_team_with_duplicate_origin_conversation_id_returns_conflict() { + let (repo, _db) = repo().await; + let mut first = make_team_for_user("t1", "user-a", "First"); + first.origin_conversation_id = Some("conv-origin".into()); + repo.create_team(&first).await.unwrap(); + + let mut second = make_team_for_user("t2", "user-a", "Second"); + second.origin_conversation_id = Some("conv-origin".into()); + let result = repo.create_team(&second).await; + + assert!( + matches!(result, Err(DbError::Conflict(_))), + "expected Conflict error for duplicate origin_conversation_id, got {:?}", + result + ); +} + +#[tokio::test] +async fn create_team_with_same_null_origin_conversation_id_is_allowed() { + let (repo, _db) = repo().await; + let first = make_team_for_user("t1", "user-a", "First"); + assert!(first.origin_conversation_id.is_none()); + repo.create_team(&first).await.unwrap(); + + let second = make_team_for_user("t2", "user-a", "Second"); + assert!(second.origin_conversation_id.is_none()); + repo.create_team(&second).await.unwrap(); + + let teams = repo.list_teams_by_user("user-a").await.unwrap(); + assert_eq!(teams.len(), 2); +} + +#[tokio::test] +async fn update_team_origin_conversation_id_conflict_returns_conflict() { + let (repo, _db) = repo().await; + let mut first = make_team_for_user("t1", "user-a", "First"); + first.origin_conversation_id = Some("conv-origin".into()); + repo.create_team(&first).await.unwrap(); + + let mut second = make_team_for_user("t2", "user-a", "Second"); + second.origin_conversation_id = None; + repo.create_team(&second).await.unwrap(); + + let result = repo + .update_team( + "t2", + &UpdateTeamParams { + origin_conversation_id: Some("conv-origin".into()), + ..Default::default() + }, + ) + .await; + + assert!( + matches!(result, Err(DbError::Conflict(_))), + "expected Conflict error when updating origin_conversation_id to duplicate, got {:?}", + result + ); +} + // ── Mailbox Tests ──────────────────────────────────────────────────── #[tokio::test] diff --git a/crates/aionui-team/Cargo.toml b/crates/aionui-team/Cargo.toml index e44e4b84a..a6ae56ed7 100644 --- a/crates/aionui-team/Cargo.toml +++ b/crates/aionui-team/Cargo.toml @@ -27,8 +27,11 @@ futures-util.workspace = true sqlx.workspace = true tokio = { workspace = true, features = ["test-util"] } futures-util.workspace = true +futures = "0.3" reqwest.workspace = true tempfile.workspace = true +tower = { workspace = true, features = ["util"] } +http-body-util.workspace = true # Enable the `AgentInstance::Mock` variant so tests can build fake agents # through the trait-object escape hatch without spawning real CLI processes. aionui-ai-agent = { workspace = true, features = ["test-support"] } diff --git a/crates/aionui-team/src/provisioning.rs b/crates/aionui-team/src/provisioning.rs index 75931fee4..56d1c3501 100644 --- a/crates/aionui-team/src/provisioning.rs +++ b/crates/aionui-team/src/provisioning.rs @@ -76,6 +76,16 @@ pub struct TeamConversationCreateResult { pub workspace: String, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ConversationMetadata { + pub conversation_id: String, + pub user_id: String, + pub assistant_id: Option, + pub backend: Option, + pub model: Option, + pub workspace: Option, +} + #[async_trait] pub trait TeamConversationProvisioningPort: Send + Sync { async fn create_team_conversation( @@ -87,6 +97,12 @@ pub trait TeamConversationProvisioningPort: Send + Sync { async fn conversation_assistant_id(&self, conversation_id: &str) -> Result, TeamError>; + /// Return ownership and runtime metadata for a source conversation. + /// + /// Used by ad-hoc team creation to verify `user_id` ownership and to derive + /// the leader backend/model/workspace from the original conversation state. + async fn conversation_metadata(&self, conversation_id: &str) -> Result, TeamError>; + async fn create_team_temp_workspace(&self, team_id: &str) -> Result; async fn patch_runtime_config(&self, conversation_id: &str, patch: serde_json::Value) -> Result<(), TeamError>; @@ -180,8 +196,24 @@ impl TeamAgentProvisioner { let leader_backend = self .resolve_requested_backend(leader_input.backend.as_deref(), leader_assistant_id.as_deref()) .await?; - let leader_conversation = self - .create_team_conversation_for_agent( + + // If the caller supplied an existing conversation_id for the lead, reuse it + // instead of creating a new team conversation. This is the promotion path + // from a solo conversation to a team lead. + let leader_conversation = if let Some(ref existing_conversation_id) = leader_input.conversation_id { + self.bind_existing_conversation_as_leader( + user_id, + team_id, + &leader_slot_id, + existing_conversation_id, + &leader_backend, + &leader_input.model, + leader_assistant_id.as_deref(), + shared_workspace, + ) + .await? + } else { + self.create_team_conversation_for_agent( user_id, team_id, &leader_slot_id, @@ -193,7 +225,8 @@ impl TeamAgentProvisioner { shared_workspace, None, ) - .await?; + .await? + }; let team_workspace = match shared_workspace { Some(workspace) => workspace.to_owned(), @@ -605,6 +638,59 @@ impl TeamAgentProvisioner { }) } + #[allow(clippy::too_many_arguments)] + async fn bind_existing_conversation_as_leader( + &self, + _user_id: &str, + team_id: &str, + slot_id: &str, + conversation_id: &str, + backend: &str, + model: &str, + assistant_id: Option<&str>, + workspace: Option<&str>, + ) -> Result { + let acp_metadata = acp_backend_metadata(&self.agent_metadata_repo, backend).await?; + let agent_type = if acp_metadata.is_some() { + AgentType::Acp + } else { + parse_agent_type(backend)? + }; + let mut extra = self.build_team_extra( + team_id, + slot_id, + TeammateRole::Lead, + backend, + model, + assistant_id, + workspace, + agent_type, + acp_metadata.as_ref(), + None, + ); + if agent_type != AgentType::Aionrs { + extra["current_model_id"] = serde_json::Value::String(model.to_owned()); + } + + self.conversation_port + .patch_runtime_config(conversation_id, extra) + .await?; + + let resolved_workspace = self.conversation_port.conversation_workspace(conversation_id).await?; + + info!( + team_id, + slot_id, + conversation_id = %conversation_id, + outcome = "bound", + "Team lead conversation reused from existing conversation" + ); + Ok(ProvisionedConversation { + conversation_id: conversation_id.to_owned(), + workspace: resolved_workspace, + }) + } + async fn resolve_initial_leader_workspace( &self, team_id: &str, @@ -677,6 +763,9 @@ impl TeamAgentProvisioner { if let Some(assistant_id) = assistant_id { extra["assistant_id"] = serde_json::Value::String(assistant_id.to_owned()); } + if role == TeammateRole::Teammate { + extra["team_id"] = serde_json::Value::String(team_id.to_owned()); + } if let Some(workspace) = workspace { inherit_team_workspace(&mut extra, workspace); } @@ -747,6 +836,13 @@ mod tests { Ok(None) } + async fn conversation_metadata( + &self, + _conversation_id: &str, + ) -> Result, TeamError> { + Ok(None) + } + async fn create_team_temp_workspace(&self, _team_id: &str) -> Result { Err(TeamError::InvalidRequest("unused".into())) } diff --git a/crates/aionui-team/src/routes.rs b/crates/aionui-team/src/routes.rs index 2d9b98dc7..c5f668ce4 100644 --- a/crates/aionui-team/src/routes.rs +++ b/crates/aionui-team/src/routes.rs @@ -10,7 +10,8 @@ use axum::routing::{get, post}; use aionui_ai_agent::ActiveLeaseRegistry; use aionui_api_types::{ - AddAgentRequest, ApiResponse, CancelTeamChildTurnRequest, CancelTeamRunRequest, CreateTeamRequest, + AdHocTeamAssociationResponse, AdHocTeamFromConversationResponse, AddAgentRequest, ApiResponse, + CancelTeamChildTurnRequest, CancelTeamRunRequest, CreateAdHocTeamFromConversationRequest, CreateTeamRequest, GetConfigOptionsResponse, PauseTeamSlotRequest, RenameAgentRequest, RenameTeamRequest, SendAgentMessageRequest, SendTeamMessageRequest, SetModeRequest, TeamAgentResponse, TeamListResponse, TeamResponse, TeamRunAckResponse, TeamRunStateResponse, @@ -90,6 +91,11 @@ impl From for ApiError { pub fn team_routes(state: TeamRouterState) -> Router { Router::new() .route("/api/teams", post(create_team).get(list_teams)) + .route( + "/api/teams/from-conversation", + post(create_ad_hoc_team_from_conversation), + ) + .route("/api/teams/by-conversation", get(get_ad_hoc_team_by_conversation)) .route("/api/teams/{id}", get(get_team).delete(remove_team)) .route("/api/teams/{id}/run-state", get(get_run_state)) .route("/api/teams/{id}/name", axum::routing::patch(rename_team)) @@ -148,6 +154,36 @@ async fn get_team( Ok(Json(ApiResponse::ok(team))) } +async fn create_ad_hoc_team_from_conversation( + State(state): State, + Extension(user): Extension, + body: Result, JsonRejection>, +) -> Result<(StatusCode, Json>), ApiError> { + let Json(req) = body.map_err(ApiError::from)?; + let resp = state + .service + .create_ad_hoc_team_from_conversation(&user.id, req) + .await?; + Ok((StatusCode::CREATED, Json(ApiResponse::ok(resp)))) +} + +async fn get_ad_hoc_team_by_conversation( + State(state): State, + Extension(user): Extension, + axum::extract::Query(query): axum::extract::Query, +) -> Result>, ApiError> { + let resp = state + .service + .get_ad_hoc_team_by_conversation(&user.id, &query.conversation_id) + .await?; + Ok(Json(ApiResponse::ok(resp))) +} + +#[derive(serde::Deserialize)] +struct ByConversationQuery { + conversation_id: String, +} + async fn get_run_state( State(state): State, Extension(user): Extension, diff --git a/crates/aionui-team/src/service.rs b/crates/aionui-team/src/service.rs index 137b1174b..049a9aaf2 100644 --- a/crates/aionui-team/src/service.rs +++ b/crates/aionui-team/src/service.rs @@ -8,10 +8,11 @@ use std::sync::{Arc, Weak}; use aionui_ai_agent::{ActiveLeaseRegistry, AgentError, AgentInstance, IWorkerTaskManager, IdleCleanupCoordinator}; use aionui_api_types::{ - AddAgentRequest, CreateTeamRequest, GetConfigOptionsResponse, TeamAgentResponse, TeamAgentRuntimeStatus, - TeamResponse, TeamRunAckResponse, TeamRunStateResponse, TeamSessionBinding, TeamSessionPhase, TeamSessionStatus, - TeamSessionStatusPayload, TeamToolCall, TeamToolContextResponse, TeamToolErrorCode, TeamToolErrorPayload, - TeamToolTransport, WebSocketMessage, + AdHocTeamAssociationResponse, AdHocTeamAssociationStatus, AdHocTeamFromConversationResponse, AddAgentRequest, + CreateAdHocTeamFromConversationRequest, CreateTeamRequest, GetConfigOptionsResponse, TeamAgentResponse, + TeamAgentRuntimeStatus, TeamResponse, TeamRunAckResponse, TeamRunStateResponse, TeamSessionBinding, + TeamSessionPhase, TeamSessionStatus, TeamSessionStatusPayload, TeamToolCall, TeamToolContextResponse, + TeamToolErrorCode, TeamToolErrorPayload, TeamToolTransport, WebSocketMessage, }; use aionui_common::{AgentKillReason, ConversationStatus, TimestampMs, generate_id, now_ms}; use aionui_db::models::TeamRow; @@ -321,6 +322,7 @@ impl TeamSessionService { lead_agent_id: lead_agent_id.clone(), session_mode: None, agents_version: "1.0.1".into(), + origin_conversation_id: None, created_at: now, updated_at: now, }; @@ -332,6 +334,7 @@ impl TeamSessionService { workspace: team_workspace, agents, lead_agent_id, + origin_conversation_id: None, created_at: now, updated_at: now, }; @@ -376,6 +379,239 @@ impl TeamSessionService { self.build_team_response(&team).await } + pub async fn create_ad_hoc_team_from_conversation( + &self, + user_id: &str, + req: CreateAdHocTeamFromConversationRequest, + ) -> Result { + let origin_conversation_id = req.conversation_id; + + // Verify source conversation ownership and capture runtime metadata. + let metadata = self + .conversation_port + .conversation_metadata(&origin_conversation_id) + .await? + .ok_or_else(|| TeamError::Forbidden(format!("conversation not found: {origin_conversation_id}")))?; + if metadata.user_id != user_id { + return Err(TeamError::Forbidden(format!( + "conversation not owned by current user: {origin_conversation_id}" + ))); + } + + let leader_assistant_id = metadata.assistant_id.ok_or_else(|| { + TeamError::InvalidRequest(format!( + "source conversation {origin_conversation_id} has no assistant_id" + )) + })?; + + // Resolve leader backend/model from the assistant definition/catalog; fall back + // to the source conversation's own backend/model only when catalog data is absent. + let (leader_backend, leader_model) = self + .resolve_spawn_backend_and_model( + Some(&leader_assistant_id), + metadata.model.as_deref(), + metadata.backend.as_deref().unwrap_or("claude"), + metadata.model.as_deref().unwrap_or("claude"), + ) + .await?; + let leader_workspace = metadata.workspace.clone(); + let leader_name = self + .assistant_catalog + .resolve_team_selectable_assistant(&leader_assistant_id) + .await? + .map(|assistant| assistant.name) + .unwrap_or_else(|| "Lead".into()); + + // Reuse existing ad-hoc team from this conversation if present. + if let Some(row) = self + .repo + .get_team_by_origin_conversation_id(user_id, &origin_conversation_id) + .await? + { + let mut team = Team::from_row(&row)?; + let leader_slot_id = team.lead_agent_id.clone().unwrap_or_default(); + let mut target_slot_id = team + .agents + .iter() + .find(|agent| agent.role == TeammateRole::Teammate) + .map(|agent| agent.slot_id.clone()); + + // Ensure the requested target assistant is present; dynamically add it if missing. + if let Some(target_assistant_id) = req.target_assistant_id.as_deref().filter(|value| !value.is_empty()) { + let already_present = team + .agents + .iter() + .filter(|agent| agent.role == TeammateRole::Teammate) + .any(|agent| agent.assistant_id.as_deref() == Some(target_assistant_id)); + if !already_present { + let (target_backend, target_model) = self + .resolve_spawn_backend_and_model( + Some(target_assistant_id), + None, + &leader_backend, + &leader_model, + ) + .await?; + let target_name = self + .assistant_catalog + .resolve_team_selectable_assistant(target_assistant_id) + .await? + .map(|assistant| assistant.name) + .unwrap_or_else(|| "Target".into()); + let added = self + .add_agent( + user_id, + &team.id, + AddAgentRequest { + name: target_name, + role: "teammate".into(), + backend: Some(target_backend), + model: target_model, + assistant_id: Some(target_assistant_id.into()), + }, + ) + .await?; + team.agents.push(TeamAgent { + slot_id: added.slot_id.clone(), + name: added.name.clone(), + role: TeammateRole::Teammate, + conversation_id: added.conversation_id.clone(), + backend: added.backend.clone(), + model: added.model.clone(), + assistant_id: added.assistant_id.clone(), + status: None, + conversation_type: None, + cli_path: None, + }); + target_slot_id = Some(added.slot_id); + } + } + + return Ok(AdHocTeamFromConversationResponse { + team_id: team.id, + origin_conversation_id: origin_conversation_id.clone(), + leader_slot_id, + target_slot_id, + created: false, + }); + } + + let mut inputs: Vec = vec![aionui_api_types::TeamAgentInput { + name: leader_name, + role: "lead".into(), + backend: Some(leader_backend.clone()), + model: leader_model.clone(), + assistant_id: Some(leader_assistant_id), + conversation_id: Some(origin_conversation_id.clone()), + }]; + + let mut target_slot_id = None; + if let Some(target_assistant_id) = req.target_assistant_id.as_deref().filter(|value| !value.is_empty()) { + let (target_backend, target_model) = self + .resolve_spawn_backend_and_model(Some(target_assistant_id), None, &leader_backend, &leader_model) + .await?; + let target_name = self + .assistant_catalog + .resolve_team_selectable_assistant(target_assistant_id) + .await? + .map(|assistant| assistant.name) + .unwrap_or_else(|| "Target".into()); + inputs.push(aionui_api_types::TeamAgentInput { + name: target_name, + role: "teammate".into(), + backend: Some(target_backend), + model: target_model, + assistant_id: Some(target_assistant_id.into()), + conversation_id: None, + }); + } + + let team_id = generate_id(); + let now = now_ms(); + let provisioned = self + .provisioner() + .provision_initial_agents(user_id, &team_id, &inputs, leader_workspace.as_deref()) + .await?; + let agents = provisioned.agents; + let lead_agent_id = provisioned.lead_agent_id; + let team_workspace = provisioned.team_workspace; + let agents_json = serde_json::to_string(&agents)?; + + if let Some(lead_slot_id) = lead_agent_id.as_ref() { + target_slot_id = agents + .iter() + .find(|agent| agent.slot_id != *lead_slot_id && agent.role == TeammateRole::Teammate) + .map(|agent| agent.slot_id.clone()); + } + + let team_name = req + .name + .as_deref() + .filter(|value| !value.trim().is_empty()) + .map(str::to_owned) + .unwrap_or_else(|| "Ad-hoc Team".into()); + + let row = TeamRow { + id: team_id.clone(), + user_id: user_id.to_owned(), + name: team_name, + workspace: team_workspace.clone(), + workspace_mode: req.workspace_mode.as_deref().unwrap_or("shared").into(), + agents: agents_json, + lead_agent_id: lead_agent_id.clone(), + session_mode: None, + agents_version: "1.0.1".into(), + origin_conversation_id: Some(origin_conversation_id.clone()), + created_at: now, + updated_at: now, + }; + self.repo.create_team(&row).await?; + + self.broadcast_team_created(&team_id, &row.name); + + info!( + team_id = %team_id, + origin_conversation_id = %origin_conversation_id, + agent_count = agents.len(), + "Ad-hoc team created from conversation" + ); + + Ok(AdHocTeamFromConversationResponse { + team_id, + origin_conversation_id: origin_conversation_id.clone(), + leader_slot_id: lead_agent_id.unwrap_or_default(), + target_slot_id, + created: true, + }) + } + + pub async fn get_ad_hoc_team_by_conversation( + &self, + user_id: &str, + origin_conversation_id: &str, + ) -> Result { + let Some(row) = self + .repo + .get_team_by_origin_conversation_id(user_id, origin_conversation_id) + .await? + else { + return Ok(AdHocTeamAssociationResponse { + team_id: String::new(), + origin_conversation_id: origin_conversation_id.to_owned(), + status: AdHocTeamAssociationStatus::Active, + team: None, + }); + }; + let team = Team::from_row(&row)?; + let team_response = self.build_team_response(&team).await.ok(); + Ok(AdHocTeamAssociationResponse { + team_id: team.id, + origin_conversation_id: origin_conversation_id.to_owned(), + status: AdHocTeamAssociationStatus::Active, + team: team_response, + }) + } + pub async fn remove_team(&self, user_id: &str, team_id: &str) -> Result<(), TeamError> { let team = self.load_owned_team(user_id, team_id).await?; @@ -397,6 +633,18 @@ impl TeamSessionService { .await; for agent in &team.agents { + let is_origin_conversation = team + .origin_conversation_id() + .as_deref() + .is_some_and(|origin_id| origin_id == agent.conversation_id); + if is_origin_conversation { + info!( + team_id = %team_id, + conversation_id = %agent.conversation_id, + "Skipping deletion of origin conversation during team removal" + ); + continue; + } let _ = self .conversation_port .delete_team_conversation(user_id, &agent.conversation_id) diff --git a/crates/aionui-team/src/service/response_builder.rs b/crates/aionui-team/src/service/response_builder.rs index 1cc387553..e4c906667 100644 --- a/crates/aionui-team/src/service/response_builder.rs +++ b/crates/aionui-team/src/service/response_builder.rs @@ -13,6 +13,7 @@ impl TeamSessionService { workspace: team.workspace.clone(), assistants: agents, leader_assistant_id: team.lead_agent_id.clone(), + origin_conversation_id: team.origin_conversation_id.clone(), created_at: team.created_at, updated_at: team.updated_at, }) diff --git a/crates/aionui-team/src/session.rs b/crates/aionui-team/src/session.rs index 95017031d..a71a2c5fd 100644 --- a/crates/aionui-team/src/session.rs +++ b/crates/aionui-team/src/session.rs @@ -2108,6 +2108,7 @@ mod tests { }, ], lead_agent_id: Some("lead-1".into()), + origin_conversation_id: None, created_at: 1000, updated_at: 1000, } diff --git a/crates/aionui-team/src/test_utils.rs b/crates/aionui-team/src/test_utils.rs index ff409483b..4bf52b368 100644 --- a/crates/aionui-team/src/test_utils.rs +++ b/crates/aionui-team/src/test_utils.rs @@ -39,6 +39,13 @@ impl ITeamRepository for MockTeamRepo { async fn get_team(&self, _id: &str) -> Result, DbError> { Ok(None) } + async fn get_team_by_origin_conversation_id( + &self, + _user_id: &str, + _origin_conversation_id: &str, + ) -> Result, DbError> { + Ok(None) + } async fn update_team(&self, _id: &str, _p: &UpdateTeamParams) -> Result<(), DbError> { Ok(()) } @@ -422,6 +429,20 @@ pub(crate) mod workspace_harness { Ok(self.teams.lock().unwrap().iter().find(|t| t.id == id).cloned()) } + async fn get_team_by_origin_conversation_id( + &self, + user_id: &str, + origin_conversation_id: &str, + ) -> Result, DbError> { + Ok(self + .teams + .lock() + .unwrap() + .iter() + .find(|t| t.user_id == user_id && t.origin_conversation_id.as_deref() == Some(origin_conversation_id)) + .cloned()) + } + async fn update_team(&self, id: &str, params: &UpdateTeamParams) -> Result<(), DbError> { let mut teams = self.teams.lock().unwrap(); let team = teams @@ -443,6 +464,9 @@ pub(crate) mod workspace_harness { if let Some(ref session_mode) = params.session_mode { team.session_mode = Some(session_mode.clone()); } + if let Some(ref origin_conversation_id) = params.origin_conversation_id { + team.origin_conversation_id = Some(origin_conversation_id.clone()); + } team.updated_at = now_ms(); Ok(()) } @@ -601,6 +625,60 @@ pub(crate) mod workspace_harness { })) } + async fn conversation_metadata( + &self, + conversation_id: &str, + ) -> Result, TeamError> { + let assistant_id = self.conversation_assistant_id(conversation_id).await?; + let extra = self.repo.get_extra(conversation_id); + let workspace = extra.as_ref().and_then(|value| { + value + .get("workspace") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }); + let backend = extra.as_ref().and_then(|value| { + value + .get("backend") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }); + let model = extra.as_ref().and_then(|value| { + value + .get("model") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }); + + let user_id = self + .repo + .conversations + .lock() + .unwrap() + .iter() + .find(|c| c.id == conversation_id) + .map(|c| c.user_id.clone()) + .unwrap_or_default(); + if user_id.is_empty() { + return Ok(None); + } + + Ok(Some(crate::provisioning::ConversationMetadata { + conversation_id: conversation_id.to_owned(), + user_id, + assistant_id, + backend, + model, + workspace, + })) + } + async fn create_team_temp_workspace(&self, team_id: &str) -> Result { let path = self .workspace_root diff --git a/crates/aionui-team/src/types.rs b/crates/aionui-team/src/types.rs index 7c7dca5f0..28cb53c65 100644 --- a/crates/aionui-team/src/types.rs +++ b/crates/aionui-team/src/types.rs @@ -150,10 +150,50 @@ pub struct Team { pub agents: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub lead_agent_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub origin_conversation_id: Option, pub created_at: TimestampMs, pub updated_at: TimestampMs, } +impl Team { + pub fn from_row(row: &TeamRow) -> Result { + let agents: Vec = serde_json::from_str(&row.agents)?; + Ok(Self { + id: row.id.clone(), + name: row.name.clone(), + workspace: row.workspace.clone(), + agents, + lead_agent_id: row.lead_agent_id.clone(), + origin_conversation_id: row.origin_conversation_id.clone(), + created_at: row.created_at, + updated_at: row.updated_at, + }) + } + + pub fn to_response(&self) -> TeamResponse { + TeamResponse { + id: self.id.clone(), + name: self.name.clone(), + workspace: self.workspace.clone(), + assistants: self.agents.iter().map(|a| a.to_response()).collect(), + leader_assistant_id: self.lead_agent_id.clone(), + origin_conversation_id: self.origin_conversation_id.clone(), + created_at: self.created_at, + updated_at: self.updated_at, + } + } + + /// The source conversation used to promote a solo conversation into this + /// team. Its conversation_id must be preserved when the team is removed. + pub fn origin_conversation_id(&self) -> Option { + self.agents + .iter() + .find(|agent| Some(&agent.slot_id) == self.lead_agent_id.as_ref()) + .map(|agent| agent.conversation_id.clone()) + } +} + // --------------------------------------------------------------------------- // MailboxMessageType // --------------------------------------------------------------------------- @@ -272,33 +312,6 @@ pub struct TeamTask { use aionui_db::models::{MailboxMessageRow, TeamRow, TeamTaskRow}; -impl Team { - pub fn from_row(row: &TeamRow) -> Result { - let agents: Vec = serde_json::from_str(&row.agents)?; - Ok(Self { - id: row.id.clone(), - name: row.name.clone(), - workspace: row.workspace.clone(), - agents, - lead_agent_id: row.lead_agent_id.clone(), - created_at: row.created_at, - updated_at: row.updated_at, - }) - } - - pub fn to_response(&self) -> TeamResponse { - TeamResponse { - id: self.id.clone(), - name: self.name.clone(), - workspace: self.workspace.clone(), - assistants: self.agents.iter().map(|a| a.to_response()).collect(), - leader_assistant_id: self.lead_agent_id.clone(), - created_at: self.created_at, - updated_at: self.updated_at, - } - } -} - impl MailboxMessage { pub fn from_row(row: &MailboxMessageRow) -> Option { let msg_type = MailboxMessageType::parse(&row.msg_type)?; @@ -643,6 +656,7 @@ mod tests { agents_version: "1.0.1".into(), created_at: 1000, updated_at: 2000, + origin_conversation_id: None, }; let team = Team::from_row(&row).unwrap(); assert_eq!(team.id, "t1"); @@ -670,6 +684,7 @@ mod tests { cli_path: None, }], lead_agent_id: Some("s1".into()), + origin_conversation_id: Some("origin-1".into()), created_at: 1000, updated_at: 2000, }; @@ -679,6 +694,7 @@ mod tests { assert_eq!(resp.assistants.len(), 1); assert_eq!(resp.assistants[0].slot_id, "s1"); assert_eq!(resp.leader_assistant_id.as_deref(), Some("s1")); + assert_eq!(resp.origin_conversation_id.as_deref(), Some("origin-1")); assert_eq!(resp.created_at, 1000); assert_eq!(resp.updated_at, 2000); } @@ -697,6 +713,7 @@ mod tests { agents_version: "1.0.1".into(), created_at: 0, updated_at: 0, + origin_conversation_id: None, }; assert!(Team::from_row(&row).is_err()); } diff --git a/crates/aionui-team/tests/common/mod.rs b/crates/aionui-team/tests/common/mod.rs index d6aa820df..ad13bbb24 100644 --- a/crates/aionui-team/tests/common/mod.rs +++ b/crates/aionui-team/tests/common/mod.rs @@ -35,6 +35,13 @@ impl ITeamRepository for MockTeamRepo { async fn get_team(&self, _id: &str) -> Result, DbError> { Ok(None) } + async fn get_team_by_origin_conversation_id( + &self, + _user_id: &str, + _origin_conversation_id: &str, + ) -> Result, DbError> { + Ok(None) + } async fn update_team(&self, _id: &str, _p: &UpdateTeamParams) -> Result<(), DbError> { Ok(()) } diff --git a/crates/aionui-team/tests/e2e_team_flow.rs b/crates/aionui-team/tests/e2e_team_flow.rs index 670ed5fb9..cfe902e54 100644 --- a/crates/aionui-team/tests/e2e_team_flow.rs +++ b/crates/aionui-team/tests/e2e_team_flow.rs @@ -685,6 +685,7 @@ async fn setup_session_with_turn_recorder_inner( workspace: "/tmp/e2e-team".into(), agents: two_agents(), lead_agent_id: Some("lead-1".into()), + origin_conversation_id: None, created_at: 1000, updated_at: 1000, }; @@ -743,6 +744,7 @@ async fn setup_session_with_runtime_ports( workspace: "/tmp/e2e-team".into(), agents: two_agents(), lead_agent_id: Some("lead-1".into()), + origin_conversation_id: None, created_at: 1000, updated_at: 1000, }; diff --git a/crates/aionui-team/tests/session_service_integration.rs b/crates/aionui-team/tests/session_service_integration.rs index 3e33201da..c1abf6daf 100644 --- a/crates/aionui-team/tests/session_service_integration.rs +++ b/crates/aionui-team/tests/session_service_integration.rs @@ -12,8 +12,9 @@ use aionui_ai_agent::task_manager::AgentFactory; use aionui_ai_agent::types::BuildTaskOptions; use aionui_ai_agent::{ActiveLeaseRegistry, AgentError, IWorkerTaskManager, WorkerTaskManagerImpl}; use aionui_api_types::{ - AcpBuildExtra, AcpConfigOptionDto, AcpConfigSelectOptionDto, AddAgentRequest, CreateTeamRequest, - GetConfigOptionsResponse, TeamAgentInput, WebSocketMessage, + AcpBuildExtra, AcpConfigOptionDto, AcpConfigSelectOptionDto, AddAgentRequest, + CreateAdHocTeamFromConversationRequest, CreateTeamRequest, GetConfigOptionsResponse, TeamAgentInput, + WebSocketMessage, }; use aionui_common::{AgentKillReason, AgentType, PaginatedResult, ProviderWithModel}; use aionui_db::models::{ @@ -426,6 +427,57 @@ impl TeamConversationProvisioningPort for FakeConversationPorts { })) } + async fn conversation_metadata( + &self, + conversation_id: &str, + ) -> Result, aionui_team::TeamError> { + let assistant_id = self.conversation_assistant_id(conversation_id).await?; + let extra = self.repo.get_extra(conversation_id); + let workspace = extra.as_ref().and_then(|value| { + value + .get("workspace") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }); + let backend = extra.as_ref().and_then(|value| { + value + .get("backend") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }); + let model = extra.as_ref().and_then(|value| { + value + .get("model") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }); + + let user_id = self + .repo + .get(conversation_id) + .await? + .map(|row| row.user_id) + .unwrap_or_default(); + if user_id.is_empty() { + return Ok(None); + } + + Ok(Some(aionui_team::provisioning::ConversationMetadata { + conversation_id: conversation_id.to_owned(), + user_id, + assistant_id, + backend, + model, + workspace, + })) + } + async fn create_team_temp_workspace(&self, team_id: &str) -> Result { if self .fail_team_temp_create @@ -690,6 +742,7 @@ struct FullMockTeamRepo { fail_workspace_update: std::sync::Mutex, fail_agent_update: std::sync::Mutex, fail_message_writes: std::sync::Mutex, + fail_create_team_conflict: std::sync::Mutex>, } impl FullMockTeamRepo { @@ -700,6 +753,7 @@ impl FullMockTeamRepo { fail_workspace_update: std::sync::Mutex::new(false), fail_agent_update: std::sync::Mutex::new(false), fail_message_writes: std::sync::Mutex::new(false), + fail_create_team_conflict: std::sync::Mutex::new(None), } } @@ -714,11 +768,23 @@ impl FullMockTeamRepo { fn fail_message_writes(&self) { *self.fail_message_writes.lock().unwrap() = true; } + + fn fail_create_team_with_conflict(&self, origin_conversation_id: &str) { + *self.fail_create_team_conflict.lock().unwrap() = Some(origin_conversation_id.to_owned()); + } } #[async_trait::async_trait] impl ITeamRepository for FullMockTeamRepo { async fn create_team(&self, row: &aionui_db::models::TeamRow) -> Result<(), DbError> { + if let Some(ref conflict_origin) = *self.fail_create_team_conflict.lock().unwrap() { + if row.origin_conversation_id.as_deref() == Some(conflict_origin.as_str()) { + return Err(DbError::Conflict(format!( + "team with origin conversation id '{}' already exists", + conflict_origin + ))); + } + } self.teams.lock().unwrap().push(row.clone()); Ok(()) } @@ -738,6 +804,19 @@ impl ITeamRepository for FullMockTeamRepo { async fn get_team(&self, id: &str) -> Result, DbError> { Ok(self.teams.lock().unwrap().iter().find(|t| t.id == id).cloned()) } + async fn get_team_by_origin_conversation_id( + &self, + user_id: &str, + origin_conversation_id: &str, + ) -> Result, DbError> { + Ok(self + .teams + .lock() + .unwrap() + .iter() + .find(|t| t.user_id == user_id && t.origin_conversation_id.as_deref() == Some(origin_conversation_id)) + .cloned()) + } async fn update_team(&self, id: &str, params: &aionui_db::UpdateTeamParams) -> Result<(), DbError> { if params.workspace.is_some() && *self.fail_workspace_update.lock().unwrap() { return Err(DbError::Init("forced workspace writeback failure".into())); @@ -1323,6 +1402,80 @@ impl IProviderRepository for EmptyProviderRepo { } } +struct ManyAssistantDefinitionRepo { + rows: Vec, +} + +impl ManyAssistantDefinitionRepo { + fn with_definitions(rows: Vec) -> Self { + Self { rows } + } +} + +#[async_trait::async_trait] +impl IAssistantDefinitionRepository for ManyAssistantDefinitionRepo { + async fn list(&self) -> Result, DbError> { + Ok(self.rows.clone()) + } + + async fn get_by_assistant_id(&self, assistant_id: &str) -> Result, DbError> { + Ok(self.rows.iter().find(|row| row.assistant_id == assistant_id).cloned()) + } + + async fn get_by_id(&self, definition_id: &str) -> Result, DbError> { + Ok(self.rows.iter().find(|row| row.id == definition_id).cloned()) + } + + async fn get_by_source_ref( + &self, + _source: &str, + _source_ref: &str, + ) -> Result, DbError> { + Ok(None) + } + + async fn upsert(&self, _params: &UpsertAssistantDefinitionParams<'_>) -> Result { + Err(DbError::Init("not implemented".into())) + } + + async fn soft_delete(&self, _definition_id: &str, _deleted_at: i64) -> Result { + Ok(false) + } +} + +struct ManyAssistantOverlayRepo { + rows: Vec, +} + +impl ManyAssistantOverlayRepo { + fn with_overlays(rows: Vec) -> Self { + Self { rows } + } +} + +#[async_trait::async_trait] +impl IAssistantOverlayRepository for ManyAssistantOverlayRepo { + async fn get(&self, definition_id: &str) -> Result, DbError> { + Ok(self + .rows + .iter() + .find(|row| row.assistant_definition_id == definition_id) + .cloned()) + } + + async fn list(&self) -> Result, DbError> { + Ok(self.rows.clone()) + } + + async fn upsert(&self, _params: &UpsertAssistantOverlayParams<'_>) -> Result { + Err(DbError::Init("not implemented".into())) + } + + async fn delete(&self, _definition_id: &str) -> Result { + Ok(false) + } +} + struct EmptyAssistantDefinitionRepo; #[async_trait::async_trait] @@ -1702,7 +1855,7 @@ async fn recovery_creates_background_intents_without_restoring_old_memory_run() #[tokio::test] async fn teammate_first_wake_uses_canonical_prompt_at_service_boundary() { - let (svc, _team_repo, turn_port, _conv_repo) = setup_with_recording_turn_port(); + let (svc, team_repo, turn_port, _conv_repo) = setup_with_recording_turn_port(); let created = svc .create_team( "user1", @@ -1719,14 +1872,23 @@ async fn teammate_first_wake_uses_canonical_prompt_at_service_boundary() { .await .expect("clear existing session"); - svc.ensure_session("user1", &created.id).await.expect("ensure"); - - // Leader-only warmup: the teammate is dormant at first start. Delivering a - // message lazily wakes it, and its first turn is a cold wake built with the - // canonical role prompt plus the delivered content (spec 5.1). - svc.send_message_to_agent("user1", &created.id, &worker_slot_id, "do X", None) + team_repo + .write_message(&aionui_db::models::MailboxMessageRow { + id: "mailbox-worker-1".into(), + team_id: created.id.clone(), + to_agent_id: worker_slot_id.clone(), + from_agent_id: "user".into(), + msg_type: "message".into(), + content: "do X".into(), + summary: None, + files: None, + read: false, + created_at: aionui_common::now_ms(), + }) .await - .expect("deliver to teammate triggers lazy wakeup"); + .expect("seed teammate mailbox"); + + svc.ensure_session("user1", &created.id).await.expect("ensure"); tokio::time::timeout(std::time::Duration::from_secs(2), async { loop { @@ -2050,6 +2212,8 @@ struct WarmupConcurrencyProbe { active: AtomicUsize, max_active: AtomicUsize, starts: Mutex>, + start_times: Mutex>, + started_at: tokio::time::Instant, } impl Default for WarmupConcurrencyProbe { @@ -2058,6 +2222,8 @@ impl Default for WarmupConcurrencyProbe { active: AtomicUsize::new(0), max_active: AtomicUsize::new(0), starts: Mutex::new(Vec::new()), + start_times: Mutex::new(Vec::new()), + started_at: tokio::time::Instant::now(), } } } @@ -2071,7 +2237,13 @@ impl WarmupConcurrencyProbe { let probe = Arc::clone(&probe); async move { let conversation_id = opts.context.conversation.conversation_id.clone(); + let elapsed = probe.started_at.elapsed(); probe.starts.lock().unwrap().push(conversation_id.clone()); + probe + .start_times + .lock() + .unwrap() + .push((conversation_id.clone(), elapsed)); let current = probe.active.fetch_add(1, Ordering::SeqCst) + 1; probe.max_active.fetch_max(current, Ordering::SeqCst); tokio::time::sleep(delay).await; @@ -2092,6 +2264,10 @@ impl WarmupConcurrencyProbe { fn starts(&self) -> Vec { self.starts.lock().unwrap().clone() } + + fn start_times(&self) -> Vec<(String, std::time::Duration)> { + self.start_times.lock().unwrap().clone() + } } async fn reset_runtime_state(svc: &Arc, tm: &Arc, team_id: &str) { @@ -2146,6 +2322,7 @@ async fn renew_active_lease_allows_empty_team_without_unrelated_lease() { agents_version: "1.0.1".into(), created_at: aionui_common::now_ms(), updated_at: aionui_common::now_ms(), + origin_conversation_id: None, }) .await .expect("insert empty team"); @@ -2201,118 +2378,628 @@ async fn force_team_workspace(repo: &Arc, team_id: &str, works .expect("force workspace"); } +fn setup_ad_hoc_team_service( + factory: AgentFactory, +) -> ( + Arc, + Arc, + Arc, + Arc, +) { + let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::with_rows(vec![ + AgentMetadataRow { + id: "assistant-lead".into(), + icon: None, + name: "Lead Assistant".into(), + name_i18n: None, + description: None, + description_i18n: None, + backend: Some("acp".into()), + agent_type: "acp".into(), + agent_source: "builtin".into(), + agent_source_info: None, + enabled: true, + command: None, + args: None, + env: None, + native_skills_dirs: None, + behavior_policy: None, + yolo_id: None, + agent_capabilities: None, + auth_methods: None, + config_options: None, + available_modes: None, + available_models: None, + available_commands: None, + sort_order: 0, + last_check_status: None, + last_check_kind: None, + last_check_error_code: None, + last_check_error_message: None, + last_check_guidance: None, + last_check_latency_ms: None, + last_check_at: None, + last_success_at: None, + last_failure_at: None, + command_override: None, + env_override: None, + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + }, + AgentMetadataRow { + id: "assistant-target".into(), + icon: None, + name: "Target Assistant".into(), + name_i18n: None, + description: None, + description_i18n: None, + backend: Some("acp".into()), + agent_type: "acp".into(), + agent_source: "builtin".into(), + agent_source_info: None, + enabled: true, + command: None, + args: None, + env: None, + native_skills_dirs: None, + behavior_policy: None, + yolo_id: None, + agent_capabilities: None, + auth_methods: None, + config_options: None, + available_modes: None, + available_models: None, + available_commands: None, + sort_order: 0, + last_check_status: None, + last_check_kind: None, + last_check_error_code: None, + last_check_error_message: None, + last_check_guidance: None, + last_check_latency_ms: None, + last_check_at: None, + last_success_at: None, + last_failure_at: None, + command_override: None, + env_override: None, + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + }, + ])); + let assistant_definition_repo: Arc = + Arc::new(ManyAssistantDefinitionRepo::with_definitions(vec![ + AssistantDefinitionRow { + id: "def-lead".into(), + assistant_id: "assistant-lead".into(), + source: "builtin".into(), + owner_type: "system".into(), + source_ref: None, + name: "Lead Assistant".into(), + name_i18n: String::new(), + description: Some("lead assistant".into()), + description_i18n: String::new(), + avatar_type: "icon".into(), + avatar_value: None, + agent_id: "assistant-lead".into(), + rule_resource_type: String::new(), + rule_resource_ref: None, + recommended_prompts: String::new(), + recommended_prompts_i18n: String::new(), + default_model_mode: "default".into(), + default_model_value: None, + default_permission_mode: "default".into(), + default_permission_value: None, + default_thought_level_mode: "default".into(), + default_thought_level_value: None, + default_skills_mode: "default".into(), + default_skill_ids: String::new(), + custom_skill_names: String::new(), + default_disabled_builtin_skill_ids: String::new(), + default_mcps_mode: "default".into(), + default_mcp_ids: String::new(), + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + deleted_at: None, + }, + AssistantDefinitionRow { + id: "def-target".into(), + assistant_id: "assistant-target".into(), + source: "builtin".into(), + owner_type: "system".into(), + source_ref: None, + name: "Target Assistant".into(), + name_i18n: String::new(), + description: Some("target assistant".into()), + description_i18n: String::new(), + avatar_type: "icon".into(), + avatar_value: None, + agent_id: "assistant-target".into(), + rule_resource_type: String::new(), + rule_resource_ref: None, + recommended_prompts: String::new(), + recommended_prompts_i18n: String::new(), + default_model_mode: "default".into(), + default_model_value: None, + default_permission_mode: "default".into(), + default_permission_value: None, + default_thought_level_mode: "default".into(), + default_thought_level_value: None, + default_skills_mode: "default".into(), + default_skill_ids: String::new(), + custom_skill_names: String::new(), + default_disabled_builtin_skill_ids: String::new(), + default_mcps_mode: "default".into(), + default_mcp_ids: String::new(), + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + deleted_at: None, + }, + ])); + let assistant_overlay_repo: Arc = + Arc::new(ManyAssistantOverlayRepo::with_overlays(vec![])); + setup_with_factory_metadata_assistants_and_conversation_repo( + factory, + agent_metadata_repo, + assistant_definition_repo, + assistant_overlay_repo, + ) +} + +fn seed_conversation(conv_repo: &Arc, user_id: &str, conversation_id: &str, assistant_id: &str) { + let now = aionui_common::now_ms(); + let extra = if assistant_id.is_empty() { + serde_json::json!({}) + } else { + serde_json::json!({ "assistant_id": assistant_id }) + }; + let row = ConversationRow { + id: conversation_id.to_owned(), + user_id: user_id.to_owned(), + name: "Source Conversation".into(), + r#type: aionui_common::AgentType::Acp.serde_name().to_owned(), + pinned: false, + pinned_at: None, + source: None, + channel_chat_id: None, + extra: serde_json::to_string(&extra).unwrap(), + model: None, + status: Some("active".into()), + created_at: now, + updated_at: now, + }; + futures::executor::block_on(conv_repo.create(&row)).expect("seed conversation"); +} + // =========================================================================== -// Test: Team CRUD (TC-*, TL-*, TG-*, TD-*, TR-*) +// Test: Ad-hoc team from conversation // =========================================================================== #[tokio::test] -async fn tc1_create_team_with_multiple_agents() { - let svc = setup(); +async fn ad_hoc_team_from_conversation_creates_leader_and_target() { + let (svc, team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-1"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); + let resp = svc - .create_team( + .create_ad_hoc_team_from_conversation( "user1", - CreateTeamRequest { - name: "Alpha".into(), - agents: two_agent_input(), - workspace: None, + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: Some("assistant-target".to_owned()), + name: Some("Ad-hoc Squad".to_owned()), + workspace_mode: Some("shared".to_owned()), }, ) .await - .unwrap(); + .expect("create ad-hoc team from conversation"); - assert_eq!(resp.name, "Alpha"); - assert_eq!(resp.assistants.len(), 2); - assert_eq!(resp.assistants[0].role, "lead"); - assert_eq!(resp.assistants[1].role, "teammate"); - assert!(resp.leader_assistant_id.is_some()); - assert_eq!(resp.leader_assistant_id, Some(resp.assistants[0].slot_id.clone())); + assert!(resp.created, "first call must create a new team"); + assert_eq!(resp.origin_conversation_id, conversation_id); + assert!(!resp.leader_slot_id.is_empty(), "leader slot must be set"); + assert!( + resp.target_slot_id.is_some(), + "target slot must be set when target_assistant_id is provided" + ); + assert_ne!(resp.target_slot_id, Some(resp.leader_slot_id.clone())); + + let stored = team_repo + .get_team(&resp.team_id) + .await + .unwrap() + .expect("team persisted"); + assert_eq!(stored.origin_conversation_id.as_deref(), Some(conversation_id)); + assert_eq!(stored.name, "Ad-hoc Squad"); } #[tokio::test] -async fn create_team_rejects_existing_conversation_id_request_side_adoption() { - let svc = setup(); +async fn ad_hoc_team_from_conversation_reuses_existing_association() { + let (svc, _team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-2"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); + + let first = svc + .create_ad_hoc_team_from_conversation( + "user1", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: None, + name: None, + workspace_mode: None, + }, + ) + .await + .expect("create first"); + + let second = svc + .create_ad_hoc_team_from_conversation( + "user1", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: None, + name: None, + workspace_mode: None, + }, + ) + .await + .expect("create second"); + + assert!(first.created); + assert!(!second.created, "second call must reuse existing team"); + assert_eq!(first.team_id, second.team_id); + assert_eq!(first.leader_slot_id, second.leader_slot_id); +} + +#[tokio::test] +async fn ad_hoc_team_from_conversation_rejects_missing_assistant_id() { + let (svc, _team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-3"; + seed_conversation(&conv_repo, "user1", conversation_id, ""); let err = svc - .create_team( + .create_ad_hoc_team_from_conversation( "user1", - CreateTeamRequest { - name: "No Adoption".into(), - agents: vec![TeamAgentInput { - name: "Lead".into(), - role: "lead".into(), - backend: Some("claude".into()), - model: "claude".into(), - assistant_id: None, - conversation_id: Some("solo-conv-1".into()), - }], - workspace: None, + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: None, + name: None, + workspace_mode: None, }, ) .await .unwrap_err(); assert!( - matches!(err, TeamError::InvalidRequest(ref msg) if msg.contains("existing conversations are no longer supported")), + matches!(err, TeamError::InvalidRequest(ref msg) if msg.contains("has no assistant_id")), "unexpected error: {err:?}" ); } #[tokio::test] -async fn create_team_with_workspace_writes_same_workspace_to_team_and_initial_agents() { - let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::empty()); - let (svc, _, conv_repo) = - setup_with_factory_and_metadata_and_conversation_repo(success_factory(), agent_metadata_repo); - let workspace_dir = - std::env::temp_dir().join(format!("aionui-team-user-workspace-{}", aionui_common::generate_id())); - std::fs::create_dir_all(&workspace_dir).unwrap(); - let workspace = workspace_dir.to_string_lossy().into_owned(); +async fn get_ad_hoc_team_by_conversation_returns_association_with_team() { + let (svc, _team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-4"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); let created = svc - .create_team( + .create_ad_hoc_team_from_conversation( "user1", - CreateTeamRequest { - name: "Shared".into(), - agents: two_agent_input(), - workspace: Some(workspace.clone()), + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: Some("assistant-target".to_owned()), + name: None, + workspace_mode: None, }, ) .await - .unwrap(); + .expect("create"); - let got = svc.get_team("user1", &created.id).await.unwrap(); - assert_eq!(got.workspace, workspace); - for agent in &got.assistants { - let extra = conv_repo.get_extra(&agent.conversation_id).unwrap(); - assert_eq!( - extra.get("workspace").and_then(serde_json::Value::as_str), - Some(workspace.as_str()) - ); - } + let assoc = svc + .get_ad_hoc_team_by_conversation("user1", conversation_id) + .await + .expect("get association"); + + assert_eq!(assoc.team_id, created.team_id); + assert_eq!(assoc.origin_conversation_id, conversation_id); + assert_eq!(assoc.status, aionui_api_types::AdHocTeamAssociationStatus::Active); + let team = assoc.team.expect("team payload present"); + assert_eq!(team.id, created.team_id); + assert_eq!(team.assistants.len(), 2); } #[tokio::test] -async fn create_team_without_workspace_uses_leader_auto_workspace_for_all_initial_agents() { - let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::empty()); - let (svc, _, conv_repo) = - setup_with_factory_and_metadata_and_conversation_repo(success_factory(), agent_metadata_repo); +async fn get_ad_hoc_team_by_conversation_returns_none_for_unknown() { + let svc = setup(); + let assoc = svc + .get_ad_hoc_team_by_conversation("user1", "conv-unknown") + .await + .expect("should return Ok with empty association"); + assert_eq!(assoc.origin_conversation_id, "conv-unknown"); + assert_eq!(assoc.status, aionui_api_types::AdHocTeamAssociationStatus::Active); + assert!( + assoc.team.is_none(), + "expected no team payload when no association exists" + ); + assert!( + assoc.team_id.is_empty(), + "expected empty team_id placeholder when no association exists" + ); +} - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Auto Shared".into(), - agents: two_agent_input(), - workspace: None, +#[tokio::test] +async fn create_ad_hoc_team_from_conversation_rejects_cross_user_conversation() { + let (svc, _team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-cross-user"; + seed_conversation(&conv_repo, "user-owner", conversation_id, "assistant-lead"); + + let err = svc + .create_ad_hoc_team_from_conversation( + "user-attacker", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user-attacker".to_owned(), + target_assistant_id: None, + name: None, + workspace_mode: None, }, ) .await - .unwrap(); + .unwrap_err(); - let got = svc.get_team("user1", &created.id).await.unwrap(); - assert!(!got.workspace.trim().is_empty(), "teams.workspace must be set"); assert!( - got.workspace.contains("/conversations/acp-temp-"), - "unexpected auto workspace: {}", + matches!(err, TeamError::Forbidden(ref msg) if msg.contains("not owned")), + "unexpected error: {err:?}" + ); +} + +#[tokio::test] +async fn create_ad_hoc_team_from_conversation_reuses_existing_and_adds_missing_target() { + let (svc, _team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-reuse-target"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); + + let first = svc + .create_ad_hoc_team_from_conversation( + "user1", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: None, + name: None, + workspace_mode: None, + }, + ) + .await + .expect("create without target"); + assert!(first.created); + assert!(first.target_slot_id.is_none()); + + let second = svc + .create_ad_hoc_team_from_conversation( + "user1", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: Some("assistant-target".to_owned()), + name: None, + workspace_mode: None, + }, + ) + .await + .expect("reuse with added target"); + assert!(!second.created); + assert_eq!(second.team_id, first.team_id); + assert!( + second.target_slot_id.is_some(), + "target assistant should be dynamically added on reuse" + ); +} + +#[tokio::test] +async fn create_ad_hoc_team_from_conversation_returns_conflict_on_origin_binding_race() { + let (svc, team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-binding-race"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); + team_repo.fail_create_team_with_conflict(conversation_id); + + let err = svc + .create_ad_hoc_team_from_conversation( + "user1", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: Some("assistant-target".to_owned()), + name: Some("Race Test".to_owned()), + workspace_mode: Some("shared".to_owned()), + }, + ) + .await + .unwrap_err(); + + assert!( + matches!(err, TeamError::Database(DbError::Conflict(ref msg)) if msg.contains("origin conversation id")), + "expected Conflict error for duplicate origin binding, got {err:?}" + ); +} + +#[tokio::test] +async fn create_ad_hoc_team_from_conversation_reuses_source_conversation_as_lead() { + let (svc, _team_repo, _task_manager, conv_repo) = setup_ad_hoc_team_service(success_factory()); + let conversation_id = "conv-adhoc-reuse-source"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); + let before_count = conv_repo.conversation_count(); + + let resp = svc + .create_ad_hoc_team_from_conversation( + "user1", + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: Some("assistant-target".to_owned()), + name: None, + workspace_mode: None, + }, + ) + .await + .expect("create ad-hoc team from conversation"); + + let assoc = svc + .get_ad_hoc_team_by_conversation("user1", conversation_id) + .await + .expect("get association") + .team + .expect("team payload"); + let leader = assoc + .assistants + .iter() + .find(|agent| agent.slot_id == resp.leader_slot_id) + .expect("leader agent in response"); + + assert_eq!( + leader.conversation_id, conversation_id, + "lead agent must reuse the source conversation_id" + ); + assert_eq!( + conv_repo.conversation_count(), + before_count + 1, + "promotion must create exactly one target conversation, not a new lead conversation" + ); + + // Lifecycle: removing the promoted team must preserve the origin conversation + // while deleting the dynamically-created teammate conversation. + svc.remove_team("user1", &resp.team_id).await.unwrap(); + assert!( + conv_repo.get(conversation_id).await.unwrap().is_some(), + "origin conversation must survive team removal" + ); + let target_conversation_id = assoc + .assistants + .iter() + .find(|agent| Some(agent.slot_id.clone()) == resp.target_slot_id) + .map(|agent| agent.conversation_id.clone()) + .expect("target agent"); + assert!( + conv_repo.get(&target_conversation_id).await.unwrap().is_none(), + "target conversation must be deleted when team is removed" + ); +} + +// =========================================================================== +// Test: Team CRUD (TC-*, TL-*, TG-*, TD-*, TR-*) +// =========================================================================== + +#[tokio::test] +async fn tc1_create_team_with_multiple_agents() { + let svc = setup(); + let resp = svc + .create_team( + "user1", + CreateTeamRequest { + name: "Alpha".into(), + agents: two_agent_input(), + workspace: None, + }, + ) + .await + .unwrap(); + + assert_eq!(resp.name, "Alpha"); + assert_eq!(resp.assistants.len(), 2); + assert_eq!(resp.assistants[0].role, "lead"); + assert_eq!(resp.assistants[1].role, "teammate"); + assert!(resp.leader_assistant_id.is_some()); + assert_eq!(resp.leader_assistant_id, Some(resp.assistants[0].slot_id.clone())); +} + +#[tokio::test] +async fn create_team_rejects_existing_conversation_id_request_side_adoption() { + let svc = setup(); + + let err = svc + .create_team( + "user1", + CreateTeamRequest { + name: "No Adoption".into(), + agents: vec![TeamAgentInput { + name: "Lead".into(), + role: "lead".into(), + backend: Some("claude".into()), + model: "claude".into(), + assistant_id: None, + conversation_id: Some("solo-conv-1".into()), + }], + workspace: None, + }, + ) + .await + .unwrap_err(); + + assert!( + matches!(err, TeamError::InvalidRequest(ref msg) if msg.contains("existing conversations are no longer supported")), + "unexpected error: {err:?}" + ); +} + +#[tokio::test] +async fn create_team_with_workspace_writes_same_workspace_to_team_and_initial_agents() { + let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::empty()); + let (svc, _, conv_repo) = + setup_with_factory_and_metadata_and_conversation_repo(success_factory(), agent_metadata_repo); + let workspace_dir = + std::env::temp_dir().join(format!("aionui-team-user-workspace-{}", aionui_common::generate_id())); + std::fs::create_dir_all(&workspace_dir).unwrap(); + let workspace = workspace_dir.to_string_lossy().into_owned(); + + let created = svc + .create_team( + "user1", + CreateTeamRequest { + name: "Shared".into(), + agents: two_agent_input(), + workspace: Some(workspace.clone()), + }, + ) + .await + .unwrap(); + + let got = svc.get_team("user1", &created.id).await.unwrap(); + assert_eq!(got.workspace, workspace); + for agent in &got.assistants { + let extra = conv_repo.get_extra(&agent.conversation_id).unwrap(); + assert_eq!( + extra.get("workspace").and_then(serde_json::Value::as_str), + Some(workspace.as_str()) + ); + } +} + +#[tokio::test] +async fn create_team_without_workspace_uses_leader_auto_workspace_for_all_initial_agents() { + let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::empty()); + let (svc, _, conv_repo) = + setup_with_factory_and_metadata_and_conversation_repo(success_factory(), agent_metadata_repo); + + let created = svc + .create_team( + "user1", + CreateTeamRequest { + name: "Auto Shared".into(), + agents: two_agent_input(), + workspace: None, + }, + ) + .await + .unwrap(); + + let got = svc.get_team("user1", &created.id).await.unwrap(); + assert!(!got.workspace.trim().is_empty(), "teams.workspace must be set"); + assert!( + got.workspace.contains("/conversations/acp-temp-"), + "unexpected auto workspace: {}", got.workspace ); @@ -3691,7 +4378,7 @@ async fn manual_add_agent_active_session_attaches_runtime_in_background_without_ } #[tokio::test] -async fn manual_add_agent_attach_failure_marks_slot_error_without_leader_notice() { +async fn manual_add_agent_attach_failure_marks_slot_error_and_notifies_leader() { use futures_util::FutureExt; let fail_next = Arc::new(AtomicBool::new(false)); @@ -3763,27 +4450,24 @@ async fn manual_add_agent_attach_failure_marks_slot_error_without_leader_notice( .await .expect("manual add attach failure should mark the slot error"); - // Teammate attach failure is inline (spec 5.4②): the per-member runtime - // status goes `failed` (drives the column's failure UI), but the session - // lifecycle must NOT fail — the leader is still ready, so the team stays - // usable and the full-screen warmup overlay never appears. tokio::time::timeout(std::time::Duration::from_secs(2), async { loop { - let runtime_failed = recorder - .events_by_name("team.agentRuntimeStatusChanged") + if recorder + .events_by_name("team.sessionStatusChanged") .iter() .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(agent.slot_id.as_str()) + event.data.get("team_id").and_then(serde_json::Value::as_str) == Some(created.id.as_str()) && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); - if runtime_failed { + && event.data.get("phase").and_then(serde_json::Value::as_str) == Some("attaching_agents") + }) + { break; } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; + tokio::task::yield_now().await; } }) .await - .expect("teammate attach failure must surface inline as a failed runtime status"); + .expect("dynamic attach failure must fail the team lifecycle"); assert!(Arc::ptr_eq( &original_scheduler, @@ -3806,10 +4490,10 @@ async fn manual_add_agent_attach_failure_marks_slot_error_without_leader_notice( let lead_slot_id = created.leader_assistant_id.as_deref().expect("leader slot"); let leader_messages = team_repo.get_history(&created.id, lead_slot_id, None).await.unwrap(); assert!( - !leader_messages + leader_messages .iter() - .any(|message| message.content.contains("failed to start its runtime")), - "user-initiated add failure must NOT wake the leader; it surfaces inline to the user (spec 5.4)" + .any(|message| message.content.contains("failed to attach its runtime")), + "leader should receive a persisted attach-failure notice" ); svc.ensure_session("user1", &created.id) @@ -3841,38 +4525,19 @@ async fn manual_add_agent_attach_failure_marks_slot_error_without_leader_notice( }), "single-member retry must restore team Ready" ); - assert!( - !recorder - .events_by_name("team.sessionStatusChanged") - .iter() - .any(|event| { - event.data.get("team_id").and_then(serde_json::Value::as_str) == Some(created.id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }), - "a teammate add-then-retry flow must never fail the session lifecycle; the failure was inline (spec 5.4/5.5)" - ); } -// The full-screen overlay (warming + failure card) is leader-scoped (spec -// 5.4/5.5). A whole-team `ensure_session` — invoked on page mount, model -// switches, and before sends via `warmupSession` — reconciles non-dormant -// members. If it retries an already-failed TEAMMATE and the retry fails again, -// the team must stay usable (leader ready): `ensure_session` returns Ok, no -// session `failed` is broadcast, and the teammate failure stays inline. Only a -// LEADER reconciliation failure may fail the whole team. #[tokio::test] -async fn reensure_with_failed_teammate_keeps_team_usable_and_inline() { +async fn failed_member_returns_conflict_and_removal_restores_ready() { use futures_util::FutureExt; - // Lead builds once (cold start); every later build fails, so the teammate's - // attach fails on add AND on the explicit re-ensure retry. let build_count = Arc::new(AtomicUsize::new(0)); let factory_count = Arc::clone(&build_count); let factory: AgentFactory = Arc::new(move |opts: BuildTaskOptions| { let build_index = factory_count.fetch_add(1, Ordering::SeqCst); async move { if build_index >= 1 { - return Err(AgentError::internal("teammate build keeps failing")); + return Err(AgentError::internal("provider-secret: dynamic attach failed")); } Ok(aionui_ai_agent::AgentInstance::Mock(Arc::new( mock_agent::MockAgent::new(opts.context.conversation.conversation_id, opts.context.workspace.path), @@ -3880,12 +4545,12 @@ async fn reensure_with_failed_teammate_keeps_team_usable_and_inline() { } .boxed() }); - let (svc, _team_repo, _task_manager, recorder) = setup_with_factory_and_recording_broadcaster(factory); + let (svc, _team_repo, task_manager, recorder) = setup_with_factory_and_recording_broadcaster(factory); let created = svc .create_team( "user1", CreateTeamRequest { - name: "Re-ensure with broken teammate".into(), + name: "Failed member removal".into(), agents: vec![TeamAgentInput { name: "Lead".into(), role: "lead".into(), @@ -3900,6 +4565,9 @@ async fn reensure_with_failed_teammate_keeps_team_usable_and_inline() { .await .unwrap(); svc.ensure_session("user1", &created.id).await.unwrap(); + let original_scheduler = svc.get_session_scheduler(&created.id).unwrap(); + let lead_conversation_id = created.assistants[0].conversation_id.clone(); + task_manager.reset_calls(); let failed = svc .add_agent( @@ -3915,188 +4583,75 @@ async fn reensure_with_failed_teammate_keeps_team_usable_and_inline() { ) .await .unwrap(); - tokio::time::timeout(std::time::Duration::from_secs(2), async { loop { - let runtime_failed = recorder - .events_by_name("team.agentRuntimeStatusChanged") + if recorder + .events_by_name("team.sessionStatusChanged") .iter() - .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(failed.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); - if runtime_failed { + .any(|event| event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed")) + { break; } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; + tokio::task::yield_now().await; } }) .await - .expect("the added teammate must fail inline first"); + .expect("dynamic failure status"); - recorder.clear(); + let error = svc + .ensure_session("user1", &created.id) + .await + .expect_err("failed-member retry should report one deterministic failure"); + assert!(matches!( + error, + TeamError::MemberRuntimeFailed { + ref team_id, + ref slot_id, + ref conversation_id, + ref public_reason, + } if team_id == &created.id + && slot_id == &failed.slot_id + && conversation_id == &failed.conversation_id + && public_reason == "Agent runtime failed to start" + )); - // Explicit full re-ensure (e.g. switching the leader's model, page remount) - // retries the still-broken teammate. `join_all` inside reconciliation means - // the teammate's failed attach has completed by the time this returns. - svc.ensure_session("user1", &created.id) - .await - .expect("a teammate reconciliation failure must not fail the whole team"); + task_manager.reset_calls(); + recorder.clear(); + svc.remove_agent("user1", &created.id, &failed.slot_id).await.unwrap(); - let session_failed = recorder - .events_by_name("team.sessionStatusChanged") - .into_iter() - .find(|event| { - event.data.get("team_id").and_then(serde_json::Value::as_str) == Some(created.id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); - assert!( - session_failed.is_none(), - "a teammate reconciliation failure must not raise the full-screen failure card (session `failed`), got {session_failed:?}" - ); + assert!(Arc::ptr_eq( + &original_scheduler, + &svc.get_session_scheduler(&created.id).expect("healthy session remains") + )); + assert!(task_manager.get_task(&lead_conversation_id).is_some()); assert!( - recorder - .events_by_name("team.sessionStatusChanged") + task_manager + .snapshot() + .kill .iter() - .any(|event| { - event.data.get("team_id").and_then(serde_json::Value::as_str) == Some(created.id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("ready") - }), - "the team stays ready after a teammate reconciliation failure (leader ready = usable)" + .all(|(conversation_id, _)| conversation_id != &lead_conversation_id) ); assert!( recorder - .events_by_name("team.agentRuntimeStatusChanged") + .events_by_name("team.sessionStatusChanged") .iter() .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(failed.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }), - "the teammate failure stays inline via agentRuntimeStatusChanged" + event.data.get("status").and_then(serde_json::Value::as_str) == Some("ready") + && event.data.get("server_count").and_then(serde_json::Value::as_u64) == Some(1) + }) ); } #[tokio::test] -async fn failed_member_stays_inline_and_removal_restores_ready() { - use futures_util::FutureExt; - - let build_count = Arc::new(AtomicUsize::new(0)); - let factory_count = Arc::clone(&build_count); - let factory: AgentFactory = Arc::new(move |opts: BuildTaskOptions| { - let build_index = factory_count.fetch_add(1, Ordering::SeqCst); - async move { - if build_index >= 1 { - return Err(AgentError::internal("provider-secret: dynamic attach failed")); - } - Ok(aionui_ai_agent::AgentInstance::Mock(Arc::new( - mock_agent::MockAgent::new(opts.context.conversation.conversation_id, opts.context.workspace.path), - ))) - } - .boxed() - }); - let (svc, _team_repo, task_manager, recorder) = setup_with_factory_and_recording_broadcaster(factory); +async fn remove_during_attach_cancels_work_and_rejects_late_ready() { + let gate = Arc::new(GatedProvisioningFactory::default()); + let (svc, _team_repo, task_manager, recorder, conv_repo) = + setup_with_factory_recording_broadcaster_and_conversation_repo(gate.factory()); let created = svc .create_team( "user1", CreateTeamRequest { - name: "Failed member removal".into(), - agents: vec![TeamAgentInput { - name: "Lead".into(), - role: "lead".into(), - backend: Some("acp".into()), - model: "claude".into(), - assistant_id: None, - conversation_id: None, - }], - workspace: None, - }, - ) - .await - .unwrap(); - svc.ensure_session("user1", &created.id).await.unwrap(); - let original_scheduler = svc.get_session_scheduler(&created.id).unwrap(); - let lead_conversation_id = created.assistants[0].conversation_id.clone(); - task_manager.reset_calls(); - - let failed = svc - .add_agent( - "user1", - &created.id, - AddAgentRequest { - name: "Broken".into(), - role: "teammate".into(), - backend: Some("acp".into()), - model: "claude".into(), - assistant_id: None, - }, - ) - .await - .unwrap(); - // The dynamic teammate's attach failure surfaces inline (spec 5.4②): its - // per-member runtime status goes `failed`. The session lifecycle stays Ready - // (leader still ready) — a teammate failure is never a whole-team failure. - tokio::time::timeout(std::time::Duration::from_secs(2), async { - loop { - let runtime_failed = recorder - .events_by_name("team.agentRuntimeStatusChanged") - .iter() - .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(failed.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); - if runtime_failed { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - }) - .await - .expect("dynamic teammate failure must surface inline as a failed runtime status"); - - // A whole-team re-ensure retries the still-broken teammate but must keep the - // team usable (leader ready): it returns Ok and the failure stays inline, - // rather than reporting a session-level conflict (spec 5.4/5.5). - svc.ensure_session("user1", &created.id) - .await - .expect("a teammate reconciliation failure must not fail the whole team"); - - task_manager.reset_calls(); - recorder.clear(); - svc.remove_agent("user1", &created.id, &failed.slot_id).await.unwrap(); - - assert!(Arc::ptr_eq( - &original_scheduler, - &svc.get_session_scheduler(&created.id).expect("healthy session remains") - )); - assert!(task_manager.get_task(&lead_conversation_id).is_some()); - assert!( - task_manager - .snapshot() - .kill - .iter() - .all(|(conversation_id, _)| conversation_id != &lead_conversation_id) - ); - assert!( - recorder - .events_by_name("team.sessionStatusChanged") - .iter() - .any(|event| { - event.data.get("status").and_then(serde_json::Value::as_str) == Some("ready") - && event.data.get("server_count").and_then(serde_json::Value::as_u64) == Some(1) - }) - ); -} - -#[tokio::test] -async fn remove_during_attach_cancels_work_and_rejects_late_ready() { - let gate = Arc::new(GatedProvisioningFactory::default()); - let (svc, _team_repo, task_manager, recorder, conv_repo) = - setup_with_factory_recording_broadcaster_and_conversation_repo(gate.factory()); - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Remove attaching member".into(), + name: "Remove attaching member".into(), agents: vec![TeamAgentInput { name: "Lead".into(), role: "lead".into(), @@ -5557,7 +6112,7 @@ async fn d9_create_team_persists_without_warming_initial_agents() { } #[tokio::test] -async fn d9_ensure_session_warms_up_only_the_lead() { +async fn d9_ensure_session_kills_and_rebuilds_every_agent() { let (svc, tm) = setup_with_factory(success_factory()); let created = svc .create_team( @@ -5574,32 +6129,22 @@ async fn d9_ensure_session_warms_up_only_the_lead() { reset_runtime_state(&svc, &tm, &created.id).await; svc.ensure_session("user1", &created.id).await.unwrap(); - // Leader-only warmup (spec 5.1): only the lead is killed+rebuilt at first - // start; the teammate stays dormant and is never built. - let lead = created.assistants.iter().find(|a| a.role == "lead").unwrap(); - let worker = created.assistants.iter().find(|a| a.role == "teammate").unwrap(); + // Two agents → kill called 2x and get_or_build_task called 2x, each with + // the corresponding conversation_id. Order is agents-iteration order. let calls = tm.snapshot(); - assert_eq!( - calls.build, - vec![lead.conversation_id.clone()], - "only the lead should be built" - ); - assert_eq!(calls.kill.len(), 1, "only the lead should be killed+rebuilt"); - assert_eq!(calls.kill[0].0, lead.conversation_id); - assert_eq!(calls.kill[0].1, Some(AgentKillReason::TeamMcpRebuild)); - assert!( - !calls.build.contains(&worker.conversation_id), - "dormant teammate must not be built at first start" - ); + assert_eq!(calls.kill.len(), 2, "expected 2 kill calls"); + assert_eq!(calls.build.len(), 2, "expected 2 build calls"); + for (i, agent) in created.assistants.iter().enumerate() { + assert_eq!(calls.kill[i].0, agent.conversation_id); + assert_eq!(calls.kill[i].1, Some(AgentKillReason::TeamMcpRebuild)); + assert_eq!(calls.build[i], agent.conversation_id); + } } -#[tokio::test] -async fn d9_ensure_session_warms_up_only_the_lead_without_teammate_stagger() { - // The batch rebuild machine (bounded concurrency + staggered starts) was - // removed in favor of leader-only warmup on a single attach path (spec 5.1). - // First start must warm exactly the lead — no teammate warmup, no stagger. +#[tokio::test(start_paused = true)] +async fn d9_ensure_session_rebuilds_agents_with_staggered_bounded_parallelism() { let probe = Arc::new(WarmupConcurrencyProbe::default()); - let (svc, _tm) = setup_with_factory(probe.factory(std::time::Duration::from_millis(10))); + let (svc, _tm) = setup_with_factory(probe.factory(std::time::Duration::from_secs(20))); let created = svc .create_team( "user1", @@ -5611,25 +6156,56 @@ async fn d9_ensure_session_warms_up_only_the_lead_without_teammate_stagger() { ) .await .unwrap(); - let lead = created - .assistants - .iter() - .find(|assistant| assistant.role == "lead") - .unwrap(); + let mut expected_starts = Vec::new(); + expected_starts.extend( + created + .assistants + .iter() + .filter(|assistant| assistant.role == "lead") + .map(|assistant| assistant.conversation_id.clone()), + ); + expected_starts.extend( + created + .assistants + .iter() + .filter(|assistant| assistant.role != "lead") + .map(|assistant| assistant.conversation_id.clone()), + ); - svc.ensure_session("user1", &created.id).await.unwrap(); + let svc_for_task = Arc::clone(&svc); + let team_id = created.id.clone(); + let handle = tokio::spawn(async move { svc_for_task.ensure_session("user1", &team_id).await }); + + tokio::time::advance(std::time::Duration::from_secs(120)).await; + handle.await.unwrap().unwrap(); let starts = probe.starts(); assert_eq!( - starts, - vec![lead.conversation_id.clone()], - "leader-only warmup must start exactly the lead" + starts, expected_starts, + "team rebuild warmup must start leader first and preserve teammate order" + ); + assert!( + probe.max_active() > 1, + "team rebuild warmup should overlap staggered agents when warmup takes longer than the launch interval" ); assert_eq!( probe.max_active(), - 1, - "leader-only warmup runs a single attach, never overlapping teammates" + 3, + "team rebuild warmup should cap concurrent agents at 3" ); + let start_times = probe.start_times(); + assert_eq!(start_times.len(), expected_starts.len()); + for pair in start_times.windows(2).take(2) { + let delta = pair[1].1.saturating_sub(pair[0].1); + assert!( + delta >= std::time::Duration::from_secs(3), + "agent starts should be staggered by at least 3s; observed {delta:?}" + ); + assert!( + delta < std::time::Duration::from_secs(5), + "agent starts should use the configured 3s stagger, not the old 5s interval; observed {delta:?}" + ); + } } #[tokio::test] @@ -5703,16 +6279,10 @@ async fn d9_ensure_session_is_idempotent() { svc.ensure_session("user1", &created.id).await.unwrap(); svc.ensure_session("user1", &created.id).await.unwrap(); - // Leader-only warmup: first ensure kills+builds only the lead; the second - // ensure reconciles (lead already Ready, teammate dormant/skipped) and adds - // no kill/build calls. + // Second call short-circuits — no additional kill/build calls. let calls = tm.snapshot(); - assert_eq!( - calls.kill.len(), - 1, - "leader-only: only the lead is (re)built, and only once" - ); - assert_eq!(calls.build.len(), 1, "second ensure_session must not re-build"); + assert_eq!(calls.kill.len(), 2, "second ensure_session must not re-kill"); + assert_eq!(calls.build.len(), 2, "second ensure_session must not re-build"); } #[tokio::test] @@ -5808,11 +6378,9 @@ async fn concurrent_ensures_launch_one_dynamic_attach() { .iter() .find(|agent| agent.role == "teammate") .unwrap(); - // Leader-only warmup leaves the worker dormant, so reconciliation would skip - // it (spec 5.1). Repair is exercised against the always-warm lead: drop its - // task so concurrent ensures reconcile it, and assert lease dedup launches - // exactly one attach. - task_manager.remove_task_without_recording(&lead.conversation_id).await; + task_manager + .remove_task_without_recording(&worker.conversation_id) + .await; task_manager.reset_calls(); gate.enable(); @@ -5831,15 +6399,15 @@ async fn concurrent_ensures_launch_one_dynamic_attach() { gate.wait_for_starts(1).await; tokio::task::yield_now().await; assert!(handles.iter().all(|handle| !handle.is_finished())); - assert_eq!(gate.starts(), vec![lead.conversation_id.clone()]); - assert_eq!(task_manager.snapshot().build, vec![lead.conversation_id.clone()]); + assert_eq!(gate.starts(), vec![worker.conversation_id.clone()]); + assert_eq!(task_manager.snapshot().build, vec![worker.conversation_id.clone()]); assert!( task_manager .snapshot() .kill .iter() - .all(|(conversation_id, _)| conversation_id != &worker.conversation_id), - "dormant members must not be woken or killed during a one-slot repair" + .all(|(conversation_id, _)| conversation_id != &lead.conversation_id), + "healthy members must not be killed during a one-slot repair" ); assert!(Arc::ptr_eq( &original_scheduler, @@ -5850,7 +6418,7 @@ async fn concurrent_ensures_launch_one_dynamic_attach() { for handle in handles { handle.await.unwrap().unwrap(); } - assert_eq!(gate.starts(), vec![lead.conversation_id.clone()]); + assert_eq!(gate.starts(), vec![worker.conversation_id.clone()]); } #[tokio::test] @@ -5887,34 +6455,47 @@ async fn stopped_session_rejects_late_attach_completion() { .await .unwrap(); gate.wait_for_starts(1).await; - let lead_conversation_id = created.assistants[0].conversation_id.clone(); svc.stop_session("user1", &created.id).await.unwrap(); - - // Leader-only warmup: the replacement session cold-starts the lead only - // (its build is not gated), so it completes without touching the - // dynamically-added worker, which stays dormant in the new session. - svc.ensure_session("user1", &created.id) - .await - .expect("replacement leader-only ensure completes"); - - // The worker attach in the old session issued exactly one kill (before its - // gated build). Record it so we can assert the fenced late completion adds - // no further kill. - let worker_kills_before_release = task_manager - .snapshot() - .kill - .iter() - .filter(|(conversation_id, _)| conversation_id == &added.conversation_id) - .count(); - - // Release the old (stopped) session's still-in-flight worker attach and let - // it run to completion. The generation fence must reject it: it must never - // publish Ready, and its stale cleanup must be skipped because a different - // session is now published (so it cannot kill the new session's runtime). + let svc_for_replacement = Arc::clone(&svc); + let replacement_team_id = created.id.clone(); + let replacement = + tokio::spawn(async move { svc_for_replacement.ensure_session("user1", &replacement_team_id).await }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let replacement_kill_started = task_manager + .snapshot() + .kill + .iter() + .filter(|(conversation_id, _)| conversation_id == &added.conversation_id) + .count(); + if replacement_kill_started >= 2 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("replacement bootstrap must begin replacing the same member before old attach release"); + assert!(!replacement.is_finished()); gate.release(1); - for _ in 0..200 { - tokio::task::yield_now().await; - } + replacement.await.unwrap().unwrap(); + + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let cleanup_kills = task_manager + .snapshot() + .kill + .iter() + .filter(|(conversation_id, _)| conversation_id == &added.conversation_id) + .count(); + if cleanup_kills >= 2 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("late completion must clean up its partial task"); let added_events = recorder .events_by_name("team.agentRuntimeStatusChanged") @@ -5926,22 +6507,12 @@ async fn stopped_session_rejects_late_attach_completion() { .iter() .filter(|event| event.data.get("status").and_then(serde_json::Value::as_str) == Some("ready")) .count(), - 0, - "a fenced late attach must not publish Ready; leader-only keeps the worker dormant in the new session" - ); - let worker_kills_after_release = task_manager - .snapshot() - .kill - .iter() - .filter(|(conversation_id, _)| conversation_id == &added.conversation_id) - .count(); - assert_eq!( - worker_kills_after_release, worker_kills_before_release, - "the fenced late completion must not kill the worker again once a different session is published" + 1, + "only the replacement session may publish Ready for the member" ); assert!( - task_manager.get_task(&lead_conversation_id).is_some(), - "stale worker cleanup must not kill the replacement session's live lead runtime" + task_manager.get_task(&added.conversation_id).is_some(), + "stale cleanup must not kill the replacement session runtime" ); assert!(svc.get_session_scheduler(&created.id).is_some()); } @@ -5968,39 +6539,27 @@ async fn d9_ensure_session_rollbacks_when_build_fails() { reset_runtime_state(&svc, &tm, &created.id).await; let result = svc.ensure_session("user1", &created.id).await; - assert!( - result.is_err(), - "ensure_session should propagate the leader build error" - ); + assert!(result.is_err(), "ensure_session should propagate build error"); - // Leader-only warmup: only the lead attach is attempted; its failure fails - // the whole session and no teammate is ever built or killed. - let lead = created.assistants.iter().find(|a| a.role == "lead").unwrap(); - let worker = created.assistants.iter().find(|a| a.role == "teammate").unwrap(); + // Serial rebuild stops at the first failing agent, and no session is + // inserted after the failure. let calls = tm.snapshot(); assert_eq!( - calls.build, - vec![lead.conversation_id.clone()], - "only the lead build is attempted" - ); - assert!( - !calls.build.contains(&worker.conversation_id) - && calls - .kill - .iter() - .all(|(conversation_id, _)| conversation_id != &worker.conversation_id), - "the dormant teammate is never built or killed on a leader bootstrap failure" + calls.kill.len(), + 3, + "failed bootstrap cleans the full two-member snapshot" ); + assert_eq!(calls.build.len(), 1); let send_result = svc.send_message("user1", &created.id, "Hello", None).await; assert!( send_result.is_err(), - "session must not be registered after leader build failure" + "session must not be registered after build failure" ); } #[tokio::test] -async fn cold_bootstrap_failure_stops_session_when_leader_attach_fails() { +async fn cold_bootstrap_failure_stops_session_and_cleans_all_successful_runtimes() { use futures_util::FutureExt; let fail_conversation_id: Arc>> = Arc::new(Mutex::new(None)); @@ -6031,48 +6590,53 @@ async fn cold_bootstrap_failure_stops_session_when_leader_attach_fails() { ) .await .unwrap(); - // Leader-only warmup: only the lead is attached at cold start, so only a - // lead failure can fail the session (teammate failures are isolated and - // deferred to lazy wakeup, spec 5.1). Fail the lead's build. - let lead = created + let failed_agent = created .assistants .iter() - .find(|assistant| assistant.role == "lead") - .expect("lead"); - *fail_conversation_id.lock().unwrap() = Some(lead.conversation_id.clone()); + .find(|assistant| assistant.name == "Worker 3") + .expect("failed agent") + .conversation_id + .clone(); + *fail_conversation_id.lock().unwrap() = Some(failed_agent); let result = svc.ensure_session("user1", &created.id).await; - assert!( - result.is_err(), - "ensure_session should propagate the leader build error" - ); + assert!(result.is_err(), "ensure_session should propagate build error"); let error = result.unwrap_err().to_string(); assert!( - error.contains(&lead.slot_id), - "leader attach failure should surface as a member-runtime failure for the lead slot: {error}" + error.contains("Worker 3") + && error.contains("backend=acp") + && error.contains("model=worker-3") + && error.contains("role=teammate"), + "rebuild error should identify the failing agent by name, backend, model, and role: {error}" ); let calls = tm.snapshot(); assert_eq!( - calls.build, - vec![lead.conversation_id.clone()], - "only the lead is built at cold start" + calls.build.len(), + 4, + "serial rebuild should stop only after the failing attempted agent" + ); + assert_eq!( + calls.kill.len(), + 11, + "cleanup is idempotent after partial-success cleanup" ); - for teammate in created.assistants.iter().filter(|a| a.role != "lead") { + for agent in &created.assistants { assert!( - !calls.build.contains(&teammate.conversation_id) - && calls - .kill - .iter() - .all(|(conversation_id, _)| conversation_id != &teammate.conversation_id), - "dormant teammate {} must never be built or killed on leader bootstrap failure", - teammate.conversation_id + calls + .kill + .iter() + .filter(|(conversation_id, _)| conversation_id == &agent.conversation_id) + .count() + >= 2, + "bootstrap failure must issue final cleanup for {}", + agent.conversation_id ); } assert_eq!(tm.active_count(), 0); assert!( svc.get_session_scheduler(&created.id).is_none(), - "session must not be registered after leader bootstrap failure" + "session must not be registered after partial rebuild failure" ); let team_session_failed = recorder @@ -6085,7 +6649,7 @@ async fn cold_bootstrap_failure_stops_session_when_leader_attach_fails() { }); assert!( team_session_failed.is_some(), - "leader bootstrap failure must emit a team-level failed/attaching_agents terminal event" + "partial rebuild failure must emit a team-level failed/attaching_agents terminal event" ); } @@ -6125,7 +6689,7 @@ async fn ensure_session_serializes_manual_add_until_rebuild_completes() { let svc_for_add = Arc::clone(&svc); let add_team_id = created.id.clone(); - let add_handle = tokio::spawn(async move { + let mut add_handle = tokio::spawn(async move { svc_for_add .add_agent( "user1", @@ -6141,10 +6705,11 @@ async fn ensure_session_serializes_manual_add_until_rebuild_completes() { .await }); - // Leader-only warmup publishes the session before the (blocking) lead attach - // and releases the membership guard, so a manual add now runs concurrently - // with leader warmup instead of being serialized behind it. The invariant - // that must still hold is a consistent final roster. + tokio::select! { + result = &mut add_handle => panic!("add_agent completed while ensure_session was rebuilding: {result:?}"), + _ = tokio::time::sleep(std::time::Duration::from_millis(200)) => {} + } + release_build.notify_waiters(); ensure_handle.await.unwrap().unwrap(); add_handle.await.unwrap().unwrap(); @@ -6186,14 +6751,17 @@ async fn ensure_session_serializes_manual_remove_until_rebuild_completes() { let svc_for_remove = Arc::clone(&svc); let remove_team_id = created.id.clone(); let remove_slot = worker_slot.clone(); - let remove_handle = tokio::spawn(async move { + let mut remove_handle = tokio::spawn(async move { svc_for_remove .remove_agent("user1", &remove_team_id, &remove_slot) .await }); - // Leader-only warmup runs the manual remove concurrently with leader warmup - // (see the add variant); assert the final roster is consistent. + tokio::select! { + result = &mut remove_handle => panic!("remove_agent completed while ensure_session was rebuilding: {result:?}"), + _ = tokio::time::sleep(std::time::Duration::from_millis(200)) => {} + } + release_build.notify_waiters(); ensure_handle.await.unwrap().unwrap(); remove_handle.await.unwrap().unwrap(); @@ -6237,14 +6805,17 @@ async fn ensure_session_serializes_manual_rename_until_rebuild_completes() { let svc_for_rename = Arc::clone(&svc); let rename_team_id = created.id.clone(); let rename_slot = worker_slot.clone(); - let rename_handle = tokio::spawn(async move { + let mut rename_handle = tokio::spawn(async move { svc_for_rename .rename_agent("user1", &rename_team_id, &rename_slot, "Senior Worker") .await }); - // Leader-only warmup runs the manual rename concurrently with leader warmup - // (see the add variant); assert the rename is reflected in the final roster. + tokio::select! { + result = &mut rename_handle => panic!("rename_agent completed while ensure_session was rebuilding: {result:?}"), + _ = tokio::time::sleep(std::time::Duration::from_millis(200)) => {} + } + release_build.notify_waiters(); ensure_handle.await.unwrap().unwrap(); rename_handle.await.unwrap().unwrap(); @@ -6359,14 +6930,9 @@ async fn d115_remove_team_kills_every_agent_process() { .unwrap(); reset_runtime_state(&svc, &tm, &created.id).await; - // Leader-only warmup: only the lead is live after ensure_session; the - // teammate stays dormant (spec 5.1). + // Bring two agents online — after ensure_session, active_count == 2. svc.ensure_session("user1", &created.id).await.unwrap(); - assert_eq!( - tm.active_count(), - 1, - "leader-only warmup registers only the lead runtime" - ); + assert_eq!(tm.active_count(), 2, "ensure_session must register 2 live agents"); let before_kill = tm.snapshot().kill.len(); @@ -6393,426 +6959,468 @@ async fn d115_remove_team_kills_every_agent_process() { } // =========================================================================== -// Task A7: per-member attach route/service — directed retry/wakeup of a single -// member runtime (dormant or failed). auth + CSRF are enforced by the shared -// team router middleware layer (same as add_agent/remove_agent/send_message); -// the service-level behavior is asserted here. +// Test: Ad-hoc team HTTP routes // =========================================================================== -#[tokio::test] -async fn attach_agent_runtime_wakes_dormant_teammate() { - let (svc, _team_repo, _task_manager, recorder) = setup_with_factory_and_recording_broadcaster(success_factory()); - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Directed attach".into(), - agents: two_agent_input(), - workspace: None, - }, - ) - .await +fn ad_hoc_router_for_user(svc: Arc, user_id: &str) -> axum::Router { + use aionui_auth::CurrentUser; + use aionui_team::{TeamRouterState, team_routes}; + + let state = TeamRouterState { + service: svc, + active_leases: Arc::new(aionui_ai_agent::ActiveLeaseRegistry::new()), + }; + team_routes(state).layer(axum::Extension(CurrentUser { + id: user_id.into(), + username: user_id.into(), + })) +} + +async fn send_route_request( + router: &mut axum::Router, + method: &str, + uri: &str, + body: Option, +) -> axum::response::Response { + use axum::body::Body; + use tower::ServiceExt; + + let request_body = body.map(|value| Body::from(serde_json::to_vec(&value).unwrap())); + let request = axum::http::Request::builder() + .method(method) + .uri(uri) + .header("content-type", "application/json") + .body(request_body.unwrap_or_else(|| Body::from(Vec::new()))) .unwrap(); - let worker = created - .assistants - .iter() - .find(|assistant| assistant.role == "teammate") - .expect("teammate") - .clone(); - svc.ensure_session("user1", &created.id).await.unwrap(); + router.clone().oneshot(request).await.unwrap() +} - // The teammate is dormant after leader-only warmup; a directed attach wakes it. - svc.attach_agent_runtime("user1", &created.id, &worker.slot_id) - .await - .expect("directed attach should succeed"); +async fn route_response_json(response: axum::response::Response) -> serde_json::Value { + use http_body_util::BodyExt; - tokio::time::timeout(std::time::Duration::from_secs(2), async { - loop { - let ready = recorder - .events_by_name("team.agentRuntimeStatusChanged") - .iter() - .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(worker.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("ready") - }); - if ready { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - }) - .await - .expect("directed attach must bring the dormant teammate to ready"); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + serde_json::from_slice(&bytes).unwrap() } #[tokio::test] -async fn attach_agent_runtime_rejects_cross_user() { - let (svc, _tm) = setup_with_factory(success_factory()); - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Directed attach isolation".into(), - agents: two_agent_input(), - workspace: None, - }, - ) - .await - .unwrap(); - let worker = created - .assistants - .iter() - .find(|assistant| assistant.role == "teammate") - .expect("teammate") - .clone(); - svc.ensure_session("user1", &created.id).await.unwrap(); +async fn route_create_ad_hoc_team_from_conversation_returns_201_with_payload() { + let (svc, _team_repo, _tm, conv_repo) = setup_ad_hoc_team_service(success_factory()); + seed_conversation(&conv_repo, "user1", "conv-route-create", "assistant-lead"); + let mut router = ad_hoc_router_for_user(svc, "user1"); + + let response = send_route_request( + &mut router, + "POST", + "/api/teams/from-conversation", + Some(serde_json::json!({ + "conversation_id": "conv-route-create", + "user_id": "user1", + "target_assistant_id": "assistant-target", + "name": "HTTP Ad-hoc", + "workspace_mode": "shared" + })), + ) + .await; - let error = svc - .attach_agent_runtime("intruder", &created.id, &worker.slot_id) - .await - .expect_err("cross-user directed attach must be rejected"); - assert!( - matches!(error, TeamError::Forbidden(_)), - "expected Forbidden, got {error:?}" - ); + assert_eq!(response.status(), axum::http::StatusCode::CREATED); + let json = route_response_json(response).await; + assert!(json["success"].as_bool().unwrap()); + let data = json["data"].as_object().unwrap(); + assert_eq!(data["origin_conversation_id"], "conv-route-create"); + assert!(data["created"].as_bool().unwrap()); + assert!(!data["leader_slot_id"].as_str().unwrap().is_empty()); + assert!(data["target_slot_id"].as_str().is_some()); } #[tokio::test] -async fn attach_agent_runtime_rejects_unknown_slot() { - let (svc, _tm) = setup_with_factory(success_factory()); - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Directed attach unknown slot".into(), - agents: two_agent_input(), - workspace: None, - }, - ) - .await - .unwrap(); - svc.ensure_session("user1", &created.id).await.unwrap(); +async fn route_get_ad_hoc_team_by_conversation_returns_association() { + let (svc, _team_repo, _tm, conv_repo) = setup_ad_hoc_team_service(success_factory()); + seed_conversation(&conv_repo, "user1", "conv-route-get", "assistant-lead"); + let mut router = ad_hoc_router_for_user(svc.clone(), "user1"); + + send_route_request( + &mut router, + "POST", + "/api/teams/from-conversation", + Some(serde_json::json!({ + "conversation_id": "conv-route-get", + "user_id": "user1", + "target_assistant_id": "assistant-target" + })), + ) + .await; - let error = svc - .attach_agent_runtime("user1", &created.id, "slot-does-not-exist") - .await - .expect_err("attaching an unknown slot must be rejected"); - assert!( - matches!(error, TeamError::AgentNotFound(_)), - "expected AgentNotFound, got {error:?}" - ); + let response = send_route_request( + &mut router, + "GET", + "/api/teams/by-conversation?conversation_id=conv-route-get", + None, + ) + .await; + + assert_eq!(response.status(), axum::http::StatusCode::OK); + let json = route_response_json(response).await; + assert!(json["success"].as_bool().unwrap()); + let data = json["data"].as_object().unwrap(); + assert_eq!(data["origin_conversation_id"], "conv-route-get"); + assert!(data["team"].is_object()); } -// The full-screen warmup overlay is driven by session-level status and must -// reflect the leader only (spec 5.4/5.5): once the team is Ready (leader ready), -// waking a dormant teammate is an inline, per-column event and must NOT flip the -// session back to `starting` — otherwise the overlay resurfaces on every lazy -// wakeup / add-member. #[tokio::test] -async fn waking_dormant_teammate_does_not_resurface_session_starting() { - let (svc, _team_repo, _task_manager, recorder) = setup_with_factory_and_recording_broadcaster(success_factory()); - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Lazy wakeup overlay".into(), - agents: two_agent_input(), - workspace: None, - }, - ) - .await - .unwrap(); - let worker = created - .assistants - .iter() - .find(|assistant| assistant.role == "teammate") - .expect("teammate") - .clone(); - svc.ensure_session("user1", &created.id).await.unwrap(); +async fn route_get_ad_hoc_team_by_conversation_returns_empty_for_unassociated() { + let svc = setup(); + let mut router = ad_hoc_router_for_user(svc, "user1"); - // The team is Ready with the teammate dormant. Drop the bootstrap events so - // only the wakeup's session-status broadcasts remain. - recorder.clear(); + let response = send_route_request( + &mut router, + "GET", + "/api/teams/by-conversation?conversation_id=conv-route-unassociated", + None, + ) + .await; - svc.attach_agent_runtime("user1", &created.id, &worker.slot_id) - .await - .expect("directed attach should succeed"); + assert_eq!(response.status(), axum::http::StatusCode::OK); + let json = route_response_json(response).await; + assert!(json["success"].as_bool().unwrap()); + let data = json["data"].as_object().unwrap(); + assert!(data.get("team").is_none()); + assert_eq!(data.get("team_id").unwrap().as_str().unwrap(), ""); +} - // Wait until the teammate attach has fully completed (ready broadcast). - tokio::time::timeout(std::time::Duration::from_secs(2), async { - loop { - let ready = recorder - .events_by_name("team.agentRuntimeStatusChanged") - .iter() - .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(worker.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("ready") - }); - if ready { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - }) - .await - .expect("teammate attach must complete"); +#[tokio::test] +async fn route_create_ad_hoc_team_from_conversation_rejects_cross_user() { + let (svc, _team_repo, _tm, conv_repo) = setup_ad_hoc_team_service(success_factory()); + seed_conversation(&conv_repo, "user1", "conv-route-cross", "assistant-lead"); + let mut router = ad_hoc_router_for_user(svc, "user-attacker"); + + let response = send_route_request( + &mut router, + "POST", + "/api/teams/from-conversation", + Some(serde_json::json!({ + "conversation_id": "conv-route-cross", + "user_id": "user-attacker" + })), + ) + .await; - let starting = recorder - .events_by_name("team.sessionStatusChanged") - .into_iter() - .find(|event| { - event.data.get("team_id").and_then(serde_json::Value::as_str) == Some(created.id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("starting") - }); - assert!( - starting.is_none(), - "waking a dormant teammate must not resurface the warmup overlay (session `starting`); \ - teammate lifecycle is inline via agentRuntimeStatusChanged, got {starting:?}" - ); + assert_eq!(response.status(), axum::http::StatusCode::FORBIDDEN); + let json = route_response_json(response).await; + assert!(!json["success"].as_bool().unwrap()); } -// A teammate's lazy-attach FAILURE is inline (spec 5.4②): the per-member -// runtime status goes `failed` and the send box gates that column, but the -// session-level status must stay Ready (leader still ready = team usable). A -// teammate failure must never raise the full-screen failure card. +// ── team_id marker in conversation extra ───────────────────────────── + #[tokio::test] -async fn failed_teammate_wakeup_does_not_flip_session_to_failed() { - use futures_util::FutureExt; +async fn teammate_conversation_extra_has_team_id_marker_for_sidebar_filtering() { + let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::empty()); + let (svc, _, conv_repo) = + setup_with_factory_and_metadata_and_conversation_repo(success_factory(), agent_metadata_repo); - let fail_next = Arc::new(AtomicBool::new(false)); - let factory_fail_next = Arc::clone(&fail_next); - let factory: AgentFactory = Arc::new(move |opts: BuildTaskOptions| { - let should_fail = factory_fail_next.swap(false, Ordering::SeqCst); - async move { - if should_fail { - return Err(AgentError::internal("simulated teammate lazy attach failure")); - } - Ok(aionui_ai_agent::AgentInstance::Mock(Arc::new( - mock_agent::MockAgent::new(opts.context.conversation.conversation_id, opts.context.workspace.path), - ))) - } - .boxed() - }); - let (svc, _team_repo, _task_manager, recorder) = setup_with_factory_and_recording_broadcaster(factory); let created = svc .create_team( "user1", CreateTeamRequest { - name: "Teammate failure stays inline".into(), + name: "FilterMarker".into(), agents: two_agent_input(), workspace: None, }, ) .await .unwrap(); - let worker = created - .assistants - .iter() - .find(|assistant| assistant.role == "teammate") - .expect("teammate") - .clone(); - svc.ensure_session("user1", &created.id).await.unwrap(); - // Leader is Ready; drop bootstrap events. Now arm a failure and wake the - // dormant teammate — the failure must stay inline. - recorder.clear(); - fail_next.store(true, Ordering::SeqCst); - svc.send_message_to_agent("user1", &created.id, &worker.slot_id, "please do X", None) - .await - .expect("human delivery acks immediately even though the lazy attach will fail"); + let got = svc.get_team("user1", &created.id).await.unwrap(); + let leader = got.assistants.iter().find(|a| a.role == "lead").unwrap(); + let teammate = got.assistants.iter().find(|a| a.role == "teammate").unwrap(); - tokio::time::timeout(std::time::Duration::from_secs(2), async { - loop { - let failed = recorder - .events_by_name("team.agentRuntimeStatusChanged") - .iter() - .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(worker.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); - if failed { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - }) - .await - .expect("failed lazy attach must broadcast a failed runtime status for the teammate"); + let leader_extra = conv_repo.get_extra(&leader.conversation_id).unwrap(); + let teammate_extra = conv_repo.get_extra(&teammate.conversation_id).unwrap(); - let session_failed = recorder - .events_by_name("team.sessionStatusChanged") - .into_iter() - .find(|event| { - event.data.get("team_id").and_then(serde_json::Value::as_str) == Some(created.id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); + // Leader must have teamId (session binding) but NOT team_id (sidebar visible) + assert_eq!( + leader_extra.get("teamId").and_then(serde_json::Value::as_str), + Some(created.id.as_str()), + "leader must carry teamId for session binding" + ); assert!( - session_failed.is_none(), - "a teammate lazy-attach failure must stay inline and never flip session status to `failed` \ - (leader still ready = team usable, spec 5.4②), got {session_failed:?}" + leader_extra.get("team_id").is_none(), + "leader must NOT carry team_id so it stays visible in sidebar" + ); + + // Teammate must have BOTH teamId and team_id (hidden from sidebar) + assert_eq!( + teammate_extra.get("teamId").and_then(serde_json::Value::as_str), + Some(created.id.as_str()), + "teammate must carry teamId for session binding" + ); + assert_eq!( + teammate_extra.get("team_id").and_then(serde_json::Value::as_str), + Some(created.id.as_str()), + "teammate must carry team_id so sidebar can filter it out" ); } #[tokio::test] -async fn lazy_attach_failure_preserves_unread_and_skips_leader_on_human_delivery() { - use futures_util::FutureExt; - - // Fail exactly the next build after it is armed; the lead attaches cleanly - // during cold start, then the teammate's lazy attach fails. - let fail_next = Arc::new(AtomicBool::new(false)); - let factory_fail_next = Arc::clone(&fail_next); - let factory: AgentFactory = Arc::new(move |opts: BuildTaskOptions| { - let should_fail = factory_fail_next.swap(false, Ordering::SeqCst); - async move { - if should_fail { - return Err(AgentError::internal("simulated teammate lazy attach failure")); - } - Ok(aionui_ai_agent::AgentInstance::Mock(Arc::new( - mock_agent::MockAgent::new(opts.context.conversation.conversation_id, opts.context.workspace.path), - ))) - } - .boxed() - }); - let (svc, team_repo, _task_manager, recorder) = setup_with_factory_and_recording_broadcaster(factory); - let created = svc - .create_team( - "user1", - CreateTeamRequest { - name: "Lazy failure preserves unread".into(), - agents: two_agent_input(), - workspace: None, - }, - ) - .await - .unwrap(); - let lead = created.assistants.iter().find(|a| a.role == "lead").unwrap().clone(); - let worker = created - .assistants - .iter() - .find(|a| a.role == "teammate") - .unwrap() - .clone(); - svc.ensure_session("user1", &created.id).await.unwrap(); - - // Arm the failure and deliver to the dormant teammate (human-direct). - fail_next.store(true, Ordering::SeqCst); - svc.send_message_to_agent("user1", &created.id, &worker.slot_id, "please do X", None) - .await - .expect("human delivery acks immediately even though the lazy attach will fail"); +async fn route_create_ad_hoc_team_from_conversation_rejects_unknown_conversation() { + let svc = setup(); + let mut router = ad_hoc_router_for_user(svc, "user1"); + + let response = send_route_request( + &mut router, + "POST", + "/api/teams/from-conversation", + Some(serde_json::json!({ + "conversation_id": "conv-route-unknown", + "user_id": "user1" + })), + ) + .await; - tokio::time::timeout(std::time::Duration::from_secs(2), async { - loop { - let failed = recorder - .events_by_name("team.agentRuntimeStatusChanged") - .iter() - .any(|event| { - event.data.get("slot_id").and_then(serde_json::Value::as_str) == Some(worker.slot_id.as_str()) - && event.data.get("status").and_then(serde_json::Value::as_str) == Some("failed") - }); - if failed { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - }) - .await - .expect("failed lazy attach must broadcast a failed runtime status for the teammate"); + assert_eq!(response.status(), axum::http::StatusCode::FORBIDDEN); + let json = route_response_json(response).await; + assert!(!json["success"].as_bool().unwrap()); +} - // Preserve-unread (spec 5.4b): the delivered message must remain unread so a - // retry re-drains it via reconcile_mailbox. - let worker_unread = team_repo.peek_unread(&created.id, &worker.slot_id).await.unwrap(); - assert!( - worker_unread.iter().any(|message| message.content == "please do X"), - "a failed lazy attach must not mark the pending delivery read" - ); +// =========================================================================== +// Helper: ad-hoc team service setup that exposes RecordingBroadcaster +// =========================================================================== - // Human-direct failure must NOT wake the leader (spec 5.4c): the failure is - // surfaced inline to the user instead. - let lead_unread = team_repo.peek_unread(&created.id, &lead.slot_id).await.unwrap(); - assert!( - !lead_unread - .iter() - .any(|message| message.from_agent_id == worker.slot_id), - "a human-direct lazy attach failure must not notify the leader" +fn setup_ad_hoc_team_service_with_broadcaster( + factory: AgentFactory, +) -> ( + Arc, + Arc, + Arc, + Arc, + Arc, +) { + let agent_metadata_repo: Arc = Arc::new(StubAgentMetadataRepo::with_rows(vec![ + AgentMetadataRow { + id: "assistant-lead".into(), + icon: None, + name: "Lead Assistant".into(), + name_i18n: None, + description: None, + description_i18n: None, + backend: Some("acp".into()), + agent_type: "acp".into(), + agent_source: "builtin".into(), + agent_source_info: None, + enabled: true, + command: None, + args: None, + env: None, + native_skills_dirs: None, + behavior_policy: None, + yolo_id: None, + agent_capabilities: None, + auth_methods: None, + config_options: None, + available_modes: None, + available_models: None, + available_commands: None, + sort_order: 0, + last_check_status: None, + last_check_kind: None, + last_check_error_code: None, + last_check_error_message: None, + last_check_guidance: None, + last_check_latency_ms: None, + last_check_at: None, + last_success_at: None, + last_failure_at: None, + command_override: None, + env_override: None, + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + }, + AgentMetadataRow { + id: "assistant-target".into(), + icon: None, + name: "Target Assistant".into(), + name_i18n: None, + description: None, + description_i18n: None, + backend: Some("acp".into()), + agent_type: "acp".into(), + agent_source: "builtin".into(), + agent_source_info: None, + enabled: true, + command: None, + args: None, + env: None, + native_skills_dirs: None, + behavior_policy: None, + yolo_id: None, + agent_capabilities: None, + auth_methods: None, + config_options: None, + available_modes: None, + available_models: None, + available_commands: None, + sort_order: 0, + last_check_status: None, + last_check_kind: None, + last_check_error_code: None, + last_check_error_message: None, + last_check_guidance: None, + last_check_latency_ms: None, + last_check_at: None, + last_success_at: None, + last_failure_at: None, + command_override: None, + env_override: None, + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + }, + ])); + let assistant_definition_repo: Arc = + Arc::new(ManyAssistantDefinitionRepo::with_definitions(vec![ + AssistantDefinitionRow { + id: "def-lead".into(), + assistant_id: "assistant-lead".into(), + source: "builtin".into(), + owner_type: "system".into(), + source_ref: None, + name: "Lead Assistant".into(), + name_i18n: String::new(), + description: Some("lead assistant".into()), + description_i18n: String::new(), + avatar_type: "icon".into(), + avatar_value: None, + agent_id: "assistant-lead".into(), + rule_resource_type: String::new(), + rule_resource_ref: None, + recommended_prompts: String::new(), + recommended_prompts_i18n: String::new(), + default_model_mode: "default".into(), + default_model_value: None, + default_permission_mode: "default".into(), + default_permission_value: None, + default_thought_level_mode: "default".into(), + default_thought_level_value: None, + default_skills_mode: "default".into(), + default_skill_ids: String::new(), + custom_skill_names: String::new(), + default_disabled_builtin_skill_ids: String::new(), + default_mcps_mode: "default".into(), + default_mcp_ids: String::new(), + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + deleted_at: None, + }, + AssistantDefinitionRow { + id: "def-target".into(), + assistant_id: "assistant-target".into(), + source: "builtin".into(), + owner_type: "system".into(), + source_ref: None, + name: "Target Assistant".into(), + name_i18n: String::new(), + description: Some("target assistant".into()), + description_i18n: String::new(), + avatar_type: "icon".into(), + avatar_value: None, + agent_id: "assistant-target".into(), + rule_resource_type: String::new(), + rule_resource_ref: None, + recommended_prompts: String::new(), + recommended_prompts_i18n: String::new(), + default_model_mode: "default".into(), + default_model_value: None, + default_permission_mode: "default".into(), + default_permission_value: None, + default_thought_level_mode: "default".into(), + default_thought_level_value: None, + default_skills_mode: "default".into(), + default_skill_ids: String::new(), + custom_skill_names: String::new(), + default_disabled_builtin_skill_ids: String::new(), + default_mcps_mode: "default".into(), + default_mcp_ids: String::new(), + created_at: aionui_common::now_ms(), + updated_at: aionui_common::now_ms(), + deleted_at: None, + }, + ])); + let assistant_overlay_repo: Arc = + Arc::new(ManyAssistantOverlayRepo::with_overlays(vec![])); + let team_repo = Arc::new(FullMockTeamRepo::new()); + let team_repo_dyn: Arc = team_repo.clone(); + let conv_repo = Arc::new(MockConversationRepo::new()); + let broadcaster = Arc::new(RecordingBroadcaster::new()); + let broadcaster_dyn: Arc = broadcaster.clone(); + let conversation_ports = Arc::new(FakeConversationPorts::new(conv_repo.clone())); + let conversation_port: Arc = conversation_ports.clone(); + let projection_store: Arc = conversation_ports.clone(); + let task_manager = Arc::new(CountingTaskManager::new(factory)); + let task_manager_dyn: Arc = task_manager.clone(); + let backend_binary_path = Arc::new(std::path::PathBuf::from("/tmp/aioncore-test")); + let provider_repo: Arc = Arc::new(EmptyProviderRepo); + let assistant_catalog: Arc = Arc::new(TestTeamAssistantCatalog { + agent_metadata_repo: agent_metadata_repo.clone(), + assistant_definition_repo: assistant_definition_repo.clone(), + assistant_overlay_repo: assistant_overlay_repo.clone(), + }); + let svc = TeamSessionService::new( + team_repo_dyn, + agent_metadata_repo, + assistant_catalog, + assistant_definition_repo, + assistant_overlay_repo, + provider_repo, + conversation_port, + projection_store, + broadcaster_dyn, + task_manager_dyn, + noop_turn_port(), + noop_cancellation_port(), + backend_binary_path, ); + (svc, team_repo, task_manager, conv_repo, broadcaster) } +// =========================================================================== +// Test: ad-hoc team creation broadcasts team.created + team.listChanged events +// =========================================================================== + #[tokio::test] -async fn agent_triggered_attach_failure_notifies_leader() { - use futures_util::FutureExt; +async fn ad_hoc_team_from_conversation_broadcasts_team_created_event() { + let (svc, _team_repo, _task_manager, conv_repo, broadcaster) = + setup_ad_hoc_team_service_with_broadcaster(success_factory()); + let conversation_id = "conv-adhoc-broadcast"; + seed_conversation(&conv_repo, "user1", conversation_id, "assistant-lead"); - // Positive counterpart to the human-direct/manual-add "must NOT notify" - // tests: an agent-triggered attach failure (here a leader-initiated spawn, - // which flows through the single attach path with - // notify_leader_on_failure=true) MUST wake the leader so it can re-delegate - // the work it just handed out (spec 5.4c). - // - // The lead attaches cleanly during cold start; only the spawned teammate's - // attach is armed to fail. - let fail_next = Arc::new(AtomicBool::new(false)); - let factory_fail_next = Arc::clone(&fail_next); - let factory: AgentFactory = Arc::new(move |opts: BuildTaskOptions| { - let should_fail = factory_fail_next.swap(false, Ordering::SeqCst); - async move { - if should_fail { - return Err(AgentError::internal("simulated spawned teammate attach failure")); - } - Ok(aionui_ai_agent::AgentInstance::Mock(Arc::new( - mock_agent::MockAgent::new(opts.context.conversation.conversation_id, opts.context.workspace.path), - ))) - } - .boxed() - }); - let (svc, team_repo, _task_manager, _conv_repo) = setup_with_factory_metadata_assistants_and_conversation_repo( - factory, - seeded_agent_metadata_repo(), - Arc::new(SingleAssistantDefinitionRepo { - row: word_creator_definition(), - }), - Arc::new(EmptyAssistantOverlayRepo), - ); - let created = svc - .create_team( + let resp = svc + .create_ad_hoc_team_from_conversation( "user1", - CreateTeamRequest { - name: "Agent-triggered failure notifies leader".into(), - agents: two_agent_input(), - workspace: None, + CreateAdHocTeamFromConversationRequest { + conversation_id: conversation_id.to_owned(), + user_id: "user1".to_owned(), + target_assistant_id: Some("assistant-target".to_owned()), + name: Some("Broadcast Test Team".to_owned()), + workspace_mode: Some("shared".to_owned()), }, ) .await - .unwrap(); - let lead_slot_id = created.leader_assistant_id.clone().expect("leader slot"); - svc.ensure_session("user1", &created.id).await.unwrap(); + .expect("create ad-hoc team from conversation"); - // Arm the failure, then have the leader spawn a teammate whose attach fails. - fail_next.store(true, Ordering::SeqCst); - let spawned = svc - .spawn_agent_in_session( - &created.id, - &lead_slot_id, - SpawnAgentRequest { - name: "Writer".into(), - assistant_id: Some("word-creator".into()), - }, - ) - .await - .expect("spawn returns before the background attach completes"); + assert!(resp.created); - // Agent/leader-triggered failure MUST notify the leader: its mailbox gets a - // "failed to start its runtime" message from the failed slot so it can - // re-delegate. - tokio::time::timeout(std::time::Duration::from_secs(2), async { - loop { - let leader_messages = team_repo.get_history(&created.id, &lead_slot_id, None).await.unwrap(); - if leader_messages.iter().any(|message| { - message.from_agent_id == spawned.slot_id && message.content.contains("failed to start its runtime") - }) { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - }) - .await - .expect("an agent-triggered attach failure must notify the leader (notify_leader_on_failure=true)"); + let created_events = broadcaster.events_by_name("team.created"); + assert_eq!( + created_events.len(), + 1, + "team.created must be broadcast exactly once after ad-hoc team creation" + ); + assert_eq!( + created_events[0].data["team_id"], resp.team_id, + "team.created payload must contain the created team_id" + ); + + let list_changed_events = broadcaster.events_by_name("team.listChanged"); + assert!( + list_changed_events.len() >= 1, + "team.listChanged must be broadcast after ad-hoc team creation" + ); }