diff --git a/Cargo.lock b/Cargo.lock index d5c0e49db..b8d1227a9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2124,6 +2124,7 @@ dependencies = [ "schemars 1.2.1", "serde", "serde_json", + "strum", "tempfile", "tokio", "toml 0.8.23", @@ -2464,6 +2465,7 @@ dependencies = [ "schemars 1.2.1", "serde", "serde_json", + "strum", "tempfile", "tokio", "toml 0.8.23", diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index e1e3b1498..08afabf83 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -5019,48 +5019,19 @@ components: type: string enum: [active, ended, failed] - PairTargetSelector: - type: object - required: - - stage_id - - agent_session_id - properties: - stage_id: - type: string - example: code@1 - agent_session_id: - type: string - example: ses_01 - PairTarget: type: object + additionalProperties: false required: - stage_id - - node_id - node_label - - visit - - agent_session_id properties: stage_id: type: string example: code@1 - node_id: - type: string - example: code node_label: type: string example: Code - visit: - type: integer - minimum: 1 - example: 1 - agent_session_id: - type: string - example: ses_01 - provider: - type: string - model: - type: string PairRecord: type: object @@ -5107,11 +5078,13 @@ components: PairStartRequest: type: object + additionalProperties: false required: - - target + - stage_id properties: - target: - $ref: "#/components/schemas/PairTargetSelector" + stage_id: + type: string + example: code@1 PairMessageRequest: type: object @@ -5131,7 +5104,7 @@ components: - message_id - pair_id - run_id - - target + - stage_id - text - accepted_at properties: @@ -5143,8 +5116,9 @@ components: $ref: "#/components/schemas/PairId" run_id: type: string - target: - $ref: "#/components/schemas/PairTargetSelector" + stage_id: + type: string + example: code@1 text: type: string accepted_at: @@ -5237,14 +5211,23 @@ components: PairTranscriptAssistantMessage: type: object - required: [kind, seq, event_id, ts, pair_id, target, text, model, tool_call_count] + additionalProperties: false + required: + - kind + - seq + - event_id + - ts + - pair_id + - target + - text + - tool_call_count properties: kind: type: string enum: [assistant_message] seq: type: integer - minimum: 1 + format: uint32 event_id: type: string ts: @@ -5256,16 +5239,6 @@ components: $ref: "#/components/schemas/PairTarget" text: type: string - model: - type: object - required: [provider, model_id] - properties: - provider: - type: string - model_id: - type: string - speed: - type: ["string", "null"] tool_call_count: type: integer minimum: 0 diff --git a/lib/crates/fabro-api/build.rs b/lib/crates/fabro-api/build.rs index e1471cf35..66e1d342e 100644 --- a/lib/crates/fabro-api/build.rs +++ b/lib/crates/fabro-api/build.rs @@ -383,7 +383,6 @@ fn main() { ("PairId", "fabro_types::PairId", &[]), ("PairMessageId", "fabro_types::PairMessageId", &[]), ("PairStatus", "fabro_types::PairStatus", &[]), - ("PairTargetSelector", "fabro_types::PairTargetSelector", &[]), ("PairTarget", "fabro_types::PairTarget", &[]), ("PairRecord", "fabro_types::PairRecord", &[]), ( diff --git a/lib/crates/fabro-api/src/lib.rs b/lib/crates/fabro-api/src/lib.rs index f1ff83b16..4816561e8 100644 --- a/lib/crates/fabro-api/src/lib.rs +++ b/lib/crates/fabro-api/src/lib.rs @@ -36,7 +36,7 @@ pub mod types { DiffSummary, DirtyStatus, EventEnvelope, ExecOutputTail, FailureCategory, FailureDetail, FailureSignature, GitContext, IdpIdentity, InterviewOption, InterviewQuestionRecord, PairId, PairMessageId, PairMessageRecord, PairMessageRequest, PairRecord, PairStartRequest, - PairStatus, PairTarget, PairTargetSelector, PairTranscriptEntry, PairTranscriptResponse, + PairStatus, PairTarget, PairTranscriptEntry, PairTranscriptResponse, PendingInterviewRecord, PreRunPushOutcome, Principal, PullRequest, PullRequestDetails, PullRequestDetailsStatus, PullRequestDetailsUnavailableReason, PullRequestLink, PullRequestMeta, PullRequestResponse, QuestionType, RepositoryRef, Run, diff --git a/lib/crates/fabro-api/tests/pair_round_trip.rs b/lib/crates/fabro-api/tests/pair_round_trip.rs index 2d466f251..edb6e54dc 100644 --- a/lib/crates/fabro-api/tests/pair_round_trip.rs +++ b/lib/crates/fabro-api/tests/pair_round_trip.rs @@ -5,15 +5,15 @@ use fabro_api::types::{ PairMessageRecord as ApiPairMessageRecord, PairMessageRequest as ApiPairMessageRequest, PairRecord as ApiPairRecord, PairStartRequest as ApiPairStartRequest, PairStatus as ApiPairStatus, PairTarget as ApiPairTarget, - PairTargetSelector as ApiPairTargetSelector, PairTranscriptEntry as ApiPairTranscriptEntry, + PairTranscriptEntry as ApiPairTranscriptEntry, PairTranscriptResponse as ApiPairTranscriptResponse, RunEventDetailResponse as ApiRunEventDetailResponse, RunPairStatusResponse as ApiRunPairStatusResponse, }; use fabro_types::{ PairId, PairMessageId, PairMessageRecord, PairMessageRequest, PairRecord, PairStartRequest, - PairStatus, PairTarget, PairTargetSelector, PairTranscriptEntry, PairTranscriptResponse, - RunEventDetailResponse, RunPairStatusResponse, fixtures, + PairStatus, PairTarget, PairTranscriptEntry, PairTranscriptResponse, RunEventDetailResponse, + RunPairStatusResponse, fixtures, }; use serde_json::{Value, json}; @@ -22,7 +22,6 @@ fn pair_api_reuses_canonical_types() { assert_same_type::(); assert_same_type::(); assert_same_type::(); - assert_same_type::(); assert_same_type::(); assert_same_type::(); assert_same_type::(); @@ -34,6 +33,44 @@ fn pair_api_reuses_canonical_types() { assert_same_type::(); } +#[test] +fn pair_target_omits_internal_fields() { + let target = PairTarget { + stage_id: "code@1".parse().unwrap(), + node_label: "Code".to_string(), + }; + let serialized = serde_json::to_value(&target).unwrap(); + assert_eq!( + serialized, + json!({ + "stage_id": "code@1", + "node_label": "Code" + }) + ); + assert!(serialized.get("agent_session_id").is_none()); + assert!(serialized.get("session_id").is_none()); + assert!(serialized.get("node_id").is_none()); + assert!(serialized.get("visit").is_none()); + assert!(serialized.get("provider").is_none()); + assert!(serialized.get("model").is_none()); +} + +#[test] +fn pair_start_request_only_carries_stage_id() { + let request = PairStartRequest { + stage_id: "code@1".parse().unwrap(), + }; + let serialized = serde_json::to_value(&request).unwrap(); + assert_eq!( + serialized, + json!({ + "stage_id": "code@1" + }) + ); + assert!(serialized.get("agent_session_id").is_none()); + assert!(serialized.get("target").is_none()); +} + #[test] fn pair_record_round_trips_json() { assert_round_trip::(json!({ @@ -54,10 +91,7 @@ fn pair_message_record_round_trips_json() { "client_message_id": "client-1", "pair_id": "01HZX6M29F1CD5YYMHT1F5D7WQ", "run_id": fixtures::RUN_1, - "target": { - "stage_id": "code@1", - "agent_session_id": "ses_01" - }, + "stage_id": "code@1", "text": "Can you inspect the failing test?", "accepted_at": "2026-05-18T12:01:00Z" })); @@ -139,12 +173,7 @@ fn run_event_detail_response_round_trips_json() { fn pair_target_json() -> Value { json!({ "stage_id": "code@1", - "node_id": "code", - "node_label": "Code", - "visit": 1, - "agent_session_id": "ses_01", - "provider": "openai", - "model": "gpt-5.3" + "node_label": "Code" }) } diff --git a/lib/crates/fabro-api/tests/run_event_round_trip.rs b/lib/crates/fabro-api/tests/run_event_round_trip.rs index 0ce55ca42..e39f3d8f4 100644 --- a/lib/crates/fabro-api/tests/run_event_round_trip.rs +++ b/lib/crates/fabro-api/tests/run_event_round_trip.rs @@ -78,7 +78,7 @@ fn run_event_round_trips_run_steer() { #[test] fn run_event_round_trips_pair_lifecycle_events() { - assert_run_event_round_trip(json!({ + let value = json!({ "id": "evt_pair_started", "ts": "2026-05-18T12:00:00Z", "run_id": fixtures::RUN_1, @@ -88,15 +88,24 @@ fn run_event_round_trips_pair_lifecycle_events() { "pair_id": "01HZX6M29F1CD5YYMHT1F5D7WQ", "target": { "stage_id": "code@1", - "node_id": "code", - "node_label": "Code", - "visit": 1, - "agent_session_id": "ses_01", - "provider": "openai", - "model": "gpt-5.3" + "node_label": "Code" } } - })); + }); + + let event: RunEvent = serde_json::from_value(value.clone()).unwrap(); + let serialized = serde_json::to_value(&event).unwrap(); + assert_eq!(serialized, value); + + let body = &serialized["properties"]; + let body_text = body.to_string(); + assert!(body_text.contains("stage_id")); + assert!(!body_text.contains("agent_session_id")); + assert!(!body_text.contains("session_id")); + assert!(!body_text.contains("provider")); + assert!(!body_text.contains("model")); + assert!(!body_text.contains("\"node_id\"")); + assert!(!body_text.contains("\"visit\"")); assert_run_event_round_trip(json!({ "id": "evt_pair_ended", diff --git a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs index 7ebe3391d..b24796bf6 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs @@ -424,6 +424,7 @@ async fn stdio_server_initializes_and_lists_run_tools() { "fabro_run_events", "fabro_run_gather", "fabro_run_interact", + "fabro_run_pair", "fabro_run_search", ]); for (name, _, schema) in &tools { @@ -505,7 +506,7 @@ async fn stdio_startup_and_list_tools_is_fast() { let start = std::time::Instant::now(); let client = spawn_mcp_client(&context, &[]).await; let tools = client.list_tools().await.unwrap(); - assert_eq!(tools.len(), 5); + assert_eq!(tools.len(), 6); assert!(start.elapsed() < std::time::Duration::from_secs(2)); client .shutdown() @@ -724,7 +725,7 @@ async fn mcp_missing_default_settings_reports_configure_first_error_and_stays_al error.contains("Cannot reach Fabro server: no settings.toml configured."), "{error}" ); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -1219,7 +1220,7 @@ async fn mcp_gather_rejects_too_many_runs() { .await; assert!(error.contains("run_ids"), "{error}"); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -1259,7 +1260,7 @@ async fn mcp_gather_rejects_invalid_timeout_values_before_auth() { ); assert!(poll_error.contains("poll_interval_seconds"), "{poll_error}"); assert!(!poll_error.contains("fabro auth login"), "{poll_error}"); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -1314,7 +1315,7 @@ async fn mcp_interact_error_does_not_stop_server() { .await; assert!(error.contains("message"), "{error}"); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -1486,7 +1487,7 @@ async fn mcp_create_validation_errors_happen_before_auth_or_network() { assert!(empty.contains("runs"), "{empty}"); assert!(many.contains("runs"), "{many}"); assert!(null.contains("decision"), "{null}"); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -1511,7 +1512,7 @@ async fn mcp_interact_answer_validation_happens_before_auth_or_network() { .await; assert!(error.contains("option, options, text"), "{error}"); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -1754,7 +1755,7 @@ async fn mcp_events_requires_action_specific_inputs_before_auth() { ); assert!(search_error.contains("query"), "{search_error}"); assert!(!search_error.contains("fabro auth login"), "{search_error}"); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() .await @@ -2039,7 +2040,7 @@ async fn mcp_tool_auth_error_mentions_login() { error.contains("Run `fabro auth login` to authenticate."), "{error}" ); - assert_eq!(client.list_tools().await.unwrap().len(), 5); + assert_eq!(client.list_tools().await.unwrap().len(), 6); client .shutdown() diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index d8b3e8570..cb741c731 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -14,9 +14,8 @@ use fabro_model::{Model, ModelTestMode, ProviderId}; use fabro_types::settings::run::MergeStrategy; use fabro_types::{ ArtifactUpload, EventEnvelope, PairId, PairMessageRecord, PairMessageRequest, PairRecord, - PairStartRequest, PairTargetSelector, PairTranscriptResponse, Run, RunBlobId, RunEvent, - RunEventDetailResponse, RunId, RunPairStatusResponse, RunProjection, SessionId, SessionRecord, - StageId, + PairStartRequest, PairTranscriptResponse, Run, RunBlobId, RunEvent, RunEventDetailResponse, + RunId, RunPairStatusResponse, RunProjection, SessionId, SessionRecord, StageId, }; use fabro_util::exit::{ErrorExt, ExitClass}; use futures::future::BoxFuture; @@ -967,12 +966,8 @@ impl Client { convert_type(response.into_inner()) } - pub async fn start_run_pair( - &self, - run_id: &RunId, - target: PairTargetSelector, - ) -> Result { - let body = PairStartRequest { target }; + pub async fn start_run_pair(&self, run_id: &RunId, stage_id: StageId) -> Result { + let body = PairStartRequest { stage_id }; let response = self .send_api(|client| { let body = body.clone(); diff --git a/lib/crates/fabro-interview/src/control_protocol.rs b/lib/crates/fabro-interview/src/control_protocol.rs index e428b7793..cdc820216 100644 --- a/lib/crates/fabro-interview/src/control_protocol.rs +++ b/lib/crates/fabro-interview/src/control_protocol.rs @@ -199,7 +199,7 @@ impl From for Answer { #[cfg(test)] mod tests { - use fabro_types::{PairTarget, Principal, StageId, SystemActorKind, fixtures}; + use fabro_types::{PairTarget, Principal, SystemActorKind, fixtures}; use super::*; @@ -274,18 +274,12 @@ mod tests { #[test] fn pair_start_round_trips_through_json() { - let stage_id = StageId::new("code", 1); let envelope = WorkerControlEnvelope::start_pair( fixtures::RUN_1, "01HZX6M29F1CD5YYMHT1F5D7WQ".parse().unwrap(), PairTarget { - stage_id: stage_id.clone(), - node_id: "code".to_string(), - node_label: "Code".to_string(), - visit: 1, - agent_session_id: "ses_01".to_string(), - provider: Some("openai".to_string()), - model: Some("gpt-5.4".to_string()), + stage_id: "code@1".parse().unwrap(), + node_label: "Code".to_string(), }, Principal::System { system_kind: SystemActorKind::Engine, @@ -294,6 +288,18 @@ mod tests { let json = serde_json::to_string(&envelope).unwrap(); let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); assert_eq!(parsed, envelope); + + let value: serde_json::Value = serde_json::from_str(&json).unwrap(); + let target = &value["target"]; + let target_text = target.to_string(); + assert!(target_text.contains("stage_id")); + assert!(target_text.contains("node_label")); + assert!(!target_text.contains("agent_session_id")); + assert!(!target_text.contains("session_id")); + assert!(!target_text.contains("\"node_id\"")); + assert!(!target_text.contains("\"visit\"")); + assert!(!target_text.contains("provider")); + assert!(!target_text.contains("model")); } #[test] diff --git a/lib/crates/fabro-mcp-server/Cargo.toml b/lib/crates/fabro-mcp-server/Cargo.toml index 118fa1ddc..de3caa633 100644 --- a/lib/crates/fabro-mcp-server/Cargo.toml +++ b/lib/crates/fabro-mcp-server/Cargo.toml @@ -29,9 +29,10 @@ rmcp = { workspace = true, features = ["server", "macros", "schemars", "transpor schemars = "1.2.1" serde.workspace = true serde_json.workspace = true +strum.workspace = true tokio.workspace = true toml.workspace = true [dev-dependencies] httpmock = "0.8" -tempfile = "3" +tempfile = "3" \ No newline at end of file diff --git a/lib/crates/fabro-mcp-server/src/server.rs b/lib/crates/fabro-mcp-server/src/server.rs index 5a02c9a17..4208591e3 100644 --- a/lib/crates/fabro-mcp-server/src/server.rs +++ b/lib/crates/fabro-mcp-server/src/server.rs @@ -138,6 +138,28 @@ impl FabroMcpServer { } } + #[tool( + name = "fabro_run_pair", + description = "Inspect, start, message, end, or read transcript for a live Fabro run pairing session." + )] + async fn fabro_run_pair( + &self, + params: Parameters, + ) -> Result { + let params = match run_tools::ValidatedPairRun::try_from(params.0) { + Ok(params) => params, + Err(err) => return Ok(error_result(&err)), + }; + let backend = match self.backend().await { + Ok(backend) => backend, + Err(err) => return Ok(error_result(&err)), + }; + match run_tools::pair_run(backend, params).await { + Ok(result) => success_result(&result, run_tools::pair_run_text(&result)), + Err(err) => Ok(error_result(&err)), + } + } + #[tool( name = "fabro_run_events", description = "List, inspect, or search stored events for a Fabro workflow run." @@ -196,3 +218,43 @@ fn success_result( fn error_result(err: &run_tools::ToolError) -> CallToolResult { CallToolResult::error(vec![Content::text(err.to_string())]) } + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + use std::sync::Arc; + + use serde_json::Value; + + use super::*; + use crate::FabroMcpServerSettings; + + #[test] + fn fabro_run_pair_tool_is_registered_with_stage_based_schema() { + let settings = FabroMcpServerSettings { + cwd: PathBuf::from("."), + config_path: PathBuf::from("fabro.toml"), + client_factory: Arc::new(|| { + Box::pin(async { panic!("client should not be constructed while listing tools") }) + }), + }; + let server = FabroMcpServer::new(Arc::new(settings)); + let tools = server.tool_router.list_all(); + let tool = tools + .iter() + .find(|tool| tool.name.as_ref() == "fabro_run_pair") + .expect("fabro_run_pair should be registered"); + let schema = Value::Object(tool.input_schema.as_ref().clone()); + let schema_text = schema.to_string(); + + assert!(schema_text.contains("stage_id")); + assert!(!schema_text.contains("agent_session_id")); + assert!(!schema_text.contains("session_id")); + assert!(!schema_text.contains("PairTargetSelector")); + assert!(!schema_text.contains("\"target\"")); + assert!(!schema_text.contains("provider")); + assert!(!schema_text.contains("\"model\"")); + assert!(!schema_text.contains("\"node_id\"")); + assert!(!schema_text.contains("\"visit\"")); + } +} diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 7f50b4587..6fff21ae1 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -2687,19 +2687,11 @@ fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) managed_run .active_api_targets .insert(stage_id.clone(), PairTarget { - stage_id: stage_id.clone(), - node_id: event - .node_id - .clone() - .unwrap_or_else(|| stage_id.node_id().to_string()), - node_label: event + stage_id: stage_id.clone(), + node_label: event .node_label .clone() .unwrap_or_else(|| stage_id.node_id().to_string()), - visit: stage_id.visit(), - agent_session_id: session_id.clone(), - provider: props.provider.clone(), - model: props.model.clone(), }); } } else { @@ -2721,13 +2713,7 @@ fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) .is_some_and(|current| current == session_id) { managed_run.active_steerable_stages.remove(stage_id); - if managed_run - .active_api_targets - .get(stage_id) - .is_some_and(|target| target.agent_session_id == *session_id) - { - managed_run.active_api_targets.remove(stage_id); - } + managed_run.active_api_targets.remove(stage_id); } if managed_run .active_non_steerable_stages diff --git a/lib/crates/fabro-server/src/server/handler/pair.rs b/lib/crates/fabro-server/src/server/handler/pair.rs index 83556c7e3..5d90146b9 100644 --- a/lib/crates/fabro-server/src/server/handler/pair.rs +++ b/lib/crates/fabro-server/src/server/handler/pair.rs @@ -9,12 +9,12 @@ use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use fabro_store::EventEnvelope; use fabro_types::{ - EventBody, PairId, PairMessageId, PairMessageRecord, PairMessageRequest, PairRecord, - PairStartRequest, PairStatus, PairTarget, PairTranscriptAssistantMessage, - PairTranscriptDetailRef, PairTranscriptEntry, PairTranscriptError, PairTranscriptMeta, - PairTranscriptModel, PairTranscriptResponse, PairTranscriptSystemMessage, + EventBody, MAX_PAIR_MESSAGE_BYTES, PairId, PairMessageId, PairMessageRecord, + PairMessageRequest, PairRecord, PairStartRequest, PairStatus, PairTarget, + PairTranscriptAssistantMessage, PairTranscriptDetailRef, PairTranscriptEntry, + PairTranscriptError, PairTranscriptMeta, PairTranscriptResponse, PairTranscriptSystemMessage, PairTranscriptToolCall, PairTranscriptToolStatus, PairTranscriptUserMessage, - PairTranscriptWarning, Principal, RunId, + PairTranscriptWarning, Principal, RunId, StageId, }; use fabro_workflow::run_status::RunStatus; use tokio::time::timeout; @@ -95,7 +95,7 @@ async fn start_pair( } } - let (target, transport) = match pair_target_and_transport(state.as_ref(), &id, &req.target) { + let (target, transport) = match pair_target_and_transport(state.as_ref(), &id, &req.stage_id) { Ok(value) => value, Err(response) => return response, }; @@ -208,9 +208,11 @@ async fn send_pair_message( if text.is_empty() { return ApiError::bad_request("Pair message text must not be empty.").into_response(); } - if text.len() > 8192 { - return ApiError::bad_request("Pair message text must be at most 8192 bytes.") - .into_response(); + if text.len() > MAX_PAIR_MESSAGE_BYTES { + return ApiError::bad_request(format!( + "Pair message text must be at most {MAX_PAIR_MESSAGE_BYTES} bytes." + )) + .into_response(); } let pair_window = match pair_window_by_id(state.as_ref(), &id, pair_id).await { Ok(pair) => pair, @@ -325,11 +327,6 @@ fn transcript_entry_from_event( pair_id: pair.pair_id, target: pair.target.clone(), text: props.text.clone(), - model: PairTranscriptModel { - provider: props.model.provider.to_string(), - model_id: props.model.model_id.clone(), - speed: props.model.speed.map(|speed| speed.to_string()), - }, tool_call_count: props.tool_call_count, }), ), @@ -407,8 +404,7 @@ fn transcript_entry_from_event( } fn event_matches_pair_target(pair: &PairRecord, event: &fabro_types::RunEvent) -> bool { - event.session_id.as_deref() == Some(pair.target.agent_session_id.as_str()) - && event.stage_id.as_ref() == Some(&pair.target.stage_id) + event.stage_id.as_ref() == Some(&pair.target.stage_id) } fn compact_summary(tool_name: &str, value: &serde_json::Value, is_error: bool) -> String { @@ -457,25 +453,19 @@ fn live_pair_targets(state: &AppState, id: &RunId) -> Vec { fn pair_target_and_transport( state: &AppState, id: &RunId, - selector: &fabro_types::PairTargetSelector, + stage_id: &StageId, ) -> Result<(PairTarget, Option), Response> { let runs = state.runs.lock().expect("runs lock poisoned"); let Some(run) = runs.get(id) else { return Err(ApiError::not_found("Run not found.").into_response()); }; reject_unpairable_status(run.status)?; - let Some(target) = run.active_api_targets.get(&selector.stage_id) else { + let Some(target) = run.active_api_targets.get(stage_id) else { return Err(pair_conflict( "Requested pair target is not active.", "pair_target_not_active", )); }; - if target.agent_session_id != selector.agent_session_id { - return Err(pair_conflict( - "Requested pair target is not active.", - "pair_target_not_active", - )); - } Ok((target.clone(), run.answer_transport.clone())) } @@ -630,7 +620,7 @@ async fn wait_for_pair_message_record( client_message_id: props.client_message_id.clone(), pair_id: props.pair_id, run_id: *id, - target: pair.record.target.selector(), + stage_id: pair.record.target.stage_id.clone(), text: props.text.clone(), accepted_at: envelope.event.ts, }); @@ -904,7 +894,7 @@ mod tests { use crate::test_support::{build_test_router, test_app_state}; #[test] - fn transcript_projection_includes_matching_assistant_messages() { + fn transcript_projection_matches_by_stage_id() { let pair = PairRecord { pair_id: "01HZX6M29F1CD5YYMHT1F5D7WQ".parse().unwrap(), run_id: fixtures::RUN_1, @@ -913,13 +903,8 @@ mod tests { ended_at: None, failure_reason: None, target: PairTarget { - stage_id: StageId::new("code", 1), - node_id: "code".to_string(), - node_label: "Code".to_string(), - visit: 1, - agent_session_id: "ses_01".to_string(), - provider: Some("openai".to_string()), - model: Some("gpt-5.4".to_string()), + stage_id: StageId::new("code", 1), + node_label: "Code".to_string(), }, }; @@ -945,7 +930,7 @@ mod tests { .unwrap(); assert!(matches!( - entry, + &entry, PairTranscriptEntry::AssistantMessage(PairTranscriptAssistantMessage { text, .. @@ -957,10 +942,10 @@ mod tests { &pair, &envelope( 8, - Some("other"), - Some(StageId::new("code", 1)), + Some("ses_01"), + Some(StageId::new("other", 1)), EventBody::AgentMessage(AgentMessageProps { - text: "wrong session".to_string(), + text: "wrong stage".to_string(), model: ModelRef { provider: ProviderId::new("openai"), model_id: "gpt-5.4".to_string(), @@ -974,6 +959,12 @@ mod tests { ) .is_none() ); + + let serialized = serde_json::to_value(&entry).unwrap(); + let serialized_text = serialized.to_string(); + assert!(!serialized_text.contains("agent_session_id")); + assert!(!serialized_text.contains("provider")); + assert!(!serialized_text.contains("\"model\"")); } #[tokio::test] @@ -983,13 +974,8 @@ mod tests { let run_id = RunId::new(); let pair_id = PairId::new(); let target = PairTarget { - stage_id: StageId::new("code", 1), - node_id: "code".to_string(), - node_label: "Code".to_string(), - visit: 1, - agent_session_id: "ses_01".to_string(), - provider: Some("openai".to_string()), - model: Some("gpt-5.4".to_string()), + stage_id: StageId::new("code", 1), + node_label: "Code".to_string(), }; let run_store = state .store_ref() @@ -1013,9 +999,9 @@ mod tests { &run_store, &run_id, &workflow_event::Event::AgentPairUserMessage { - node_id: target.node_id.clone(), - visit: target.visit, - session_id: target.agent_session_id.clone(), + node_id: target.stage_id.node_id().to_string(), + visit: target.stage_id.visit(), + session_id: "ses_01".to_string(), pair_id, message_id: PairMessageId::new(), client_message_id: None, diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 3736b02ca..51ce36cbc 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -2187,13 +2187,8 @@ async fn subprocess_answer_transport_pair_commands_enqueue_control_messages() { system_kind: SystemActorKind::Engine, }; let target = PairTarget { - stage_id: StageId::new("agent", 1), - node_id: "agent".to_string(), - node_label: "Agent".to_string(), - visit: 1, - agent_session_id: "ses_01".to_string(), - provider: Some("openai".to_string()), - model: Some("gpt-5.4".to_string()), + stage_id: StageId::new("agent", 1), + node_label: "Agent".to_string(), }; transport diff --git a/lib/crates/fabro-tool/Cargo.toml b/lib/crates/fabro-tool/Cargo.toml index 2457bacc8..08b6df7cf 100644 --- a/lib/crates/fabro-tool/Cargo.toml +++ b/lib/crates/fabro-tool/Cargo.toml @@ -24,6 +24,7 @@ futures.workspace = true schemars = "1.2.1" serde.workspace = true serde_json.workspace = true +strum.workspace = true tokio.workspace = true toml.workspace = true diff --git a/lib/crates/fabro-tool/src/common.rs b/lib/crates/fabro-tool/src/common.rs index 6043c9a7e..3ce4e8a92 100644 --- a/lib/crates/fabro-tool/src/common.rs +++ b/lib/crates/fabro-tool/src/common.rs @@ -5,7 +5,10 @@ use std::sync::LazyLock; use async_trait::async_trait; use chrono::{DateTime, NaiveDate, Utc}; use fabro_api::types; -use fabro_types::{Run, RunId, RunStatus}; +use fabro_types::{ + PairId, PairMessageRecord, PairMessageRequest, PairRecord, PairTranscriptResponse, Run, RunId, + RunPairStatusResponse, RunStatus, StageId, +}; use fabro_util::exit::{self, ExitClass}; use schemars::JsonSchema; use serde::Serialize; @@ -85,6 +88,45 @@ pub trait FabroToolBackend: Send + Sync { question_id: &str, body: types::SubmitAnswerRequest, ) -> anyhow::Result<()>; + + async fn get_run_pair_status(&self, _run_id: &RunId) -> anyhow::Result { + Err(ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into()) + } + + async fn start_run_pair( + &self, + _run_id: &RunId, + _stage_id: StageId, + ) -> anyhow::Result { + Err(ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into()) + } + + async fn get_run_pair(&self, _run_id: &RunId, _pair_id: &PairId) -> anyhow::Result { + Err(ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into()) + } + + async fn end_run_pair(&self, _run_id: &RunId, _pair_id: &PairId) -> anyhow::Result { + Err(ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into()) + } + + async fn send_run_pair_message( + &self, + _run_id: &RunId, + _pair_id: &PairId, + _request: PairMessageRequest, + ) -> anyhow::Result { + Err(ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into()) + } + + async fn get_run_pair_transcript( + &self, + _run_id: &RunId, + _pair_id: &PairId, + _since_seq: Option, + _limit: Option, + ) -> anyhow::Result { + Err(ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into()) + } } pub trait RunManifestBuilder: Send + Sync { @@ -127,6 +169,7 @@ pub const FABRO_RUN_SEARCH_TOOL_NAME: &str = "fabro_run_search"; pub const FABRO_RUN_INTERACT_TOOL_NAME: &str = "fabro_run_interact"; pub const FABRO_RUN_GATHER_TOOL_NAME: &str = "fabro_run_gather"; pub const FABRO_RUN_EVENTS_TOOL_NAME: &str = "fabro_run_events"; +pub const FABRO_RUN_PAIR_TOOL_NAME: &str = "fabro_run_pair"; static TOOL_DEFINITIONS: LazyLock> = LazyLock::new(|| { vec![ diff --git a/lib/crates/fabro-tool/src/fabro_client.rs b/lib/crates/fabro-tool/src/fabro_client.rs index 7c364de29..887bd17c6 100644 --- a/lib/crates/fabro-tool/src/fabro_client.rs +++ b/lib/crates/fabro-tool/src/fabro_client.rs @@ -3,7 +3,10 @@ use std::sync::Arc; use async_trait::async_trait; use fabro_api::types; -use fabro_types::{EventEnvelope, Run, RunId, RunProjection}; +use fabro_types::{ + EventEnvelope, PairId, PairMessageRecord, PairMessageRequest, PairRecord, + PairTranscriptResponse, Run, RunId, RunPairStatusResponse, RunProjection, StageId, +}; use crate::{FabroToolBackend, RunManifestBuilder, ToolError}; @@ -138,4 +141,47 @@ impl FabroToolBackend for ClientBackend { .submit_run_answer(run_id, question_id, body) .await } + + async fn get_run_pair_status(&self, run_id: &RunId) -> anyhow::Result { + self.client.get_run_pair_status(run_id).await + } + + async fn start_run_pair( + &self, + run_id: &RunId, + stage_id: StageId, + ) -> anyhow::Result { + self.client.start_run_pair(run_id, stage_id).await + } + + async fn get_run_pair(&self, run_id: &RunId, pair_id: &PairId) -> anyhow::Result { + self.client.get_run_pair(run_id, pair_id).await + } + + async fn end_run_pair(&self, run_id: &RunId, pair_id: &PairId) -> anyhow::Result { + self.client.end_run_pair(run_id, pair_id).await + } + + async fn send_run_pair_message( + &self, + run_id: &RunId, + pair_id: &PairId, + request: PairMessageRequest, + ) -> anyhow::Result { + self.client + .send_run_pair_message(run_id, pair_id, request) + .await + } + + async fn get_run_pair_transcript( + &self, + run_id: &RunId, + pair_id: &PairId, + since_seq: Option, + limit: Option, + ) -> anyhow::Result { + self.client + .get_run_pair_transcript(run_id, pair_id, since_seq, limit) + .await + } } diff --git a/lib/crates/fabro-tool/src/lib.rs b/lib/crates/fabro-tool/src/lib.rs index 02b71389c..4b9c720eb 100644 --- a/lib/crates/fabro-tool/src/lib.rs +++ b/lib/crates/fabro-tool/src/lib.rs @@ -11,12 +11,14 @@ pub mod fabro_client; mod gather; mod interact; mod manifest; +mod pair; mod search; pub use common::{ FABRO_RUN_CREATE_TOOL_NAME, FABRO_RUN_EVENTS_TOOL_NAME, FABRO_RUN_GATHER_TOOL_NAME, - FABRO_RUN_INTERACT_TOOL_NAME, FABRO_RUN_SEARCH_TOOL_NAME, FabroToolBackend, RunManifestBuilder, - RunSummaryResult, ToolDefinition, ToolError, ToolResult, tool_definitions, + FABRO_RUN_INTERACT_TOOL_NAME, FABRO_RUN_PAIR_TOOL_NAME, FABRO_RUN_SEARCH_TOOL_NAME, + FabroToolBackend, RunManifestBuilder, RunSummaryResult, ToolDefinition, ToolError, ToolResult, + tool_definitions, }; pub use create::{ CreateRunOptions, CreateRunSpec, CreateRunsResult, CreatedRunResult, FabroRunCreateParams, @@ -35,6 +37,10 @@ pub use interact::{ ValidatedInteractAction, ValidatedInteractRun, interact_run, interact_run_text, }; pub use manifest::json_to_toml_value; +pub use pair::{ + FabroRunPairParams, PairRunResult, RunPairAction, ValidatedPairAction, ValidatedPairRun, + pair_run, pair_run_text, +}; pub use search::{ FabroRunSearchParams, SearchRunSummaryResult, SearchRunsResult, ValidatedSearchRuns, search_runs, search_runs_text, diff --git a/lib/crates/fabro-tool/src/pair.rs b/lib/crates/fabro-tool/src/pair.rs new file mode 100644 index 000000000..e7d878e52 --- /dev/null +++ b/lib/crates/fabro-tool/src/pair.rs @@ -0,0 +1,564 @@ +use fabro_types::{MAX_PAIR_MESSAGE_BYTES, PairId, PairMessageRequest, StageId}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use strum::IntoStaticStr; + +use super::common::{FabroToolBackend, ToolError, ToolResult}; + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, JsonSchema, IntoStaticStr)] +#[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum RunPairAction { + Status, + Start, + Get, + Message, + End, + Transcript, +} + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct FabroRunPairParams { + pub action: RunPairAction, + pub run_id: Option, + pub pair_id: Option, + pub stage_id: Option, + pub text: Option, + pub client_message_id: Option, + pub since_seq: Option, + pub limit: Option, +} + +#[derive(Debug)] +pub struct ValidatedPairRun { + pub run_id: String, + pub action: ValidatedPairAction, +} + +#[derive(Debug)] +pub enum ValidatedPairAction { + Status, + Start { + stage_id: StageId, + }, + Get { + pair_id: PairId, + }, + Message { + pair_id: PairId, + text: String, + client_message_id: Option, + }, + End { + pair_id: PairId, + }, + Transcript { + pair_id: PairId, + since_seq: Option, + limit: Option, + }, +} + +impl ValidatedPairAction { + fn action(&self) -> RunPairAction { + match self { + Self::Status => RunPairAction::Status, + Self::Start { .. } => RunPairAction::Start, + Self::Get { .. } => RunPairAction::Get, + Self::Message { .. } => RunPairAction::Message, + Self::End { .. } => RunPairAction::End, + Self::Transcript { .. } => RunPairAction::Transcript, + } + } +} + +impl TryFrom for ValidatedPairRun { + type Error = ToolError; + + fn try_from(params: FabroRunPairParams) -> Result { + let Some(run_id) = params + .run_id + .as_deref() + .map(str::trim) + .filter(|run_id| !run_id.is_empty()) + else { + return Err(ToolError::message("run_id is required")); + }; + let run_id = run_id.to_string(); + + let action = match params.action { + RunPairAction::Status => ValidatedPairAction::Status, + RunPairAction::Start => { + let Some(stage_id_raw) = params + .stage_id + .as_deref() + .map(str::trim) + .filter(|stage_id| !stage_id.is_empty()) + else { + return Err(ToolError::message("stage_id is required for action start")); + }; + let stage_id = stage_id_raw.parse::().map_err(|err| { + ToolError::message(format!("invalid stage_id for action start: {err}")) + })?; + ValidatedPairAction::Start { stage_id } + } + RunPairAction::Get => { + let pair_id = + parse_pair_id_for_action(params.pair_id.as_deref(), RunPairAction::Get)?; + ValidatedPairAction::Get { pair_id } + } + RunPairAction::Message => { + let pair_id = + parse_pair_id_for_action(params.pair_id.as_deref(), RunPairAction::Message)?; + let Some(text) = params + .text + .as_deref() + .map(str::trim) + .filter(|text| !text.is_empty()) + else { + return Err(ToolError::message("text is required for action message")); + }; + if text.len() > MAX_PAIR_MESSAGE_BYTES { + return Err(ToolError::message(format!( + "text must be at most {MAX_PAIR_MESSAGE_BYTES} bytes for action message" + ))); + } + ValidatedPairAction::Message { + pair_id, + text: text.to_string(), + client_message_id: params + .client_message_id + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string), + } + } + RunPairAction::End => { + let pair_id = + parse_pair_id_for_action(params.pair_id.as_deref(), RunPairAction::End)?; + ValidatedPairAction::End { pair_id } + } + RunPairAction::Transcript => { + let pair_id = + parse_pair_id_for_action(params.pair_id.as_deref(), RunPairAction::Transcript)?; + ValidatedPairAction::Transcript { + pair_id, + since_seq: params.since_seq, + limit: params.limit, + } + } + }; + Ok(Self { run_id, action }) + } +} + +fn parse_pair_id_for_action(raw: Option<&str>, action: RunPairAction) -> ToolResult { + let name: &'static str = action.into(); + let Some(raw) = raw.map(str::trim).filter(|value| !value.is_empty()) else { + return Err(ToolError::message(format!( + "pair_id is required for action {name}" + ))); + }; + raw.parse::() + .map_err(|err| ToolError::message(format!("invalid pair_id for action {name}: {err}"))) +} + +#[derive(Debug, Serialize, JsonSchema)] +pub struct PairRunResult { + pub run_id: String, + pub action: RunPairAction, + pub result: Value, +} + +pub async fn pair_run( + backend: std::sync::Arc, + params: ValidatedPairRun, +) -> ToolResult { + let run_id = backend + .resolve_run(¶ms.run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + .id; + + let action = params.action.action(); + let result = match params.action { + ValidatedPairAction::Status => json!( + backend + .get_run_pair_status(&run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + ), + ValidatedPairAction::Start { stage_id } => json!( + backend + .start_run_pair(&run_id, stage_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + ), + ValidatedPairAction::Get { pair_id } => json!( + backend + .get_run_pair(&run_id, &pair_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + ), + ValidatedPairAction::Message { + pair_id, + text, + client_message_id, + } => json!( + backend + .send_run_pair_message(&run_id, &pair_id, PairMessageRequest { + text, + client_message_id, + },) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + ), + ValidatedPairAction::End { pair_id } => json!( + backend + .end_run_pair(&run_id, &pair_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + ), + ValidatedPairAction::Transcript { + pair_id, + since_seq, + limit, + } => json!( + backend + .get_run_pair_transcript(&run_id, &pair_id, since_seq, limit) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + ), + }; + + Ok(PairRunResult { + run_id: run_id.to_string(), + action, + result, + }) +} + +pub fn pair_run_text(result: &PairRunResult) -> String { + match result.action { + RunPairAction::Status => format!("read pair status for Fabro run {}", result.run_id), + RunPairAction::Start => format!("started pair for Fabro run {}", result.run_id), + RunPairAction::Get => format!("read pair for Fabro run {}", result.run_id), + RunPairAction::Message => format!("sent pair message for Fabro run {}", result.run_id), + RunPairAction::End => format!("ended pair for Fabro run {}", result.run_id), + RunPairAction::Transcript => { + format!("read pair transcript for Fabro run {}", result.run_id) + } + } +} + +#[cfg(test)] +mod tests { + use chrono::Utc; + use fabro_types::{ + PairId, PairMessageId, PairMessageRecord, PairRecord, PairStatus, PairTarget, + PairTranscriptAssistantMessage, PairTranscriptEntry, PairTranscriptMeta, + PairTranscriptResponse, RunId, RunPairStatusResponse, + }; + use serde_json::json; + + use super::*; + + fn run_id() -> RunId { + RunId::new() + } + + fn pair_id() -> PairId { + "01HZX6M29F1CD5YYMHT1F5D7WQ".parse().unwrap() + } + + fn pair_target() -> PairTarget { + PairTarget { + stage_id: "code@1".parse().unwrap(), + node_label: "Code".to_string(), + } + } + + fn assert_no_public_pair_leaks(value: &serde_json::Value) { + let text = value.to_string(); + assert!(!text.contains("agent_session_id"), "{text}"); + assert!(!text.contains("session_id"), "{text}"); + assert!(!text.contains("provider"), "{text}"); + assert!(!text.contains("\"model\""), "{text}"); + assert!(!text.contains("\"node_id\""), "{text}"); + assert!(!text.contains("\"visit\""), "{text}"); + } + + #[test] + fn missing_or_blank_run_id_returns_tool_error() { + for raw in [None, Some(String::new()), Some(" ".to_string())] { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Status, + run_id: raw, + pair_id: None, + stage_id: None, + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str().contains("run_id is required"), + "{}", + err.as_str() + ); + } + } + + #[test] + fn start_requires_stage_id() { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Start, + run_id: Some("run_123".to_string()), + pair_id: None, + stage_id: None, + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str() + .contains("stage_id is required for action start"), + "{}", + err.as_str() + ); + + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Start, + run_id: Some("run_123".to_string()), + pair_id: None, + stage_id: Some("bad-stage-id".to_string()), + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str().contains("invalid stage_id for action start"), + "{}", + err.as_str() + ); + } + + #[test] + fn message_requires_pair_id_and_text() { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Message, + run_id: Some("run_123".to_string()), + pair_id: None, + stage_id: None, + text: Some("hello".to_string()), + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str() + .contains("pair_id is required for action message"), + "{}", + err.as_str() + ); + + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Message, + run_id: Some("run_123".to_string()), + pair_id: Some(pair_id().to_string()), + stage_id: None, + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str().contains("text is required for action message"), + "{}", + err.as_str() + ); + } + + #[test] + fn message_rejects_overlong_text() { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Message, + run_id: Some("run_123".to_string()), + pair_id: Some(pair_id().to_string()), + stage_id: None, + text: Some("a".repeat(MAX_PAIR_MESSAGE_BYTES + 1)), + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str() + .contains("text must be at most 8192 bytes for action message"), + "{}", + err.as_str() + ); + } + + #[test] + fn transcript_requires_pair_id() { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Transcript, + run_id: Some("run_123".to_string()), + pair_id: None, + stage_id: None, + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str() + .contains("pair_id is required for action transcript"), + "{}", + err.as_str() + ); + } + + #[test] + fn end_requires_pair_id() { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::End, + run_id: Some("run_123".to_string()), + pair_id: None, + stage_id: None, + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str().contains("pair_id is required for action end"), + "{}", + err.as_str() + ); + } + + #[test] + fn invalid_pair_id_for_action_is_reported() { + let err = ValidatedPairRun::try_from(FabroRunPairParams { + action: RunPairAction::Get, + run_id: Some("run_123".to_string()), + pair_id: Some("not-a-pair-id".to_string()), + stage_id: None, + text: None, + client_message_id: None, + since_seq: None, + limit: None, + }) + .unwrap_err(); + assert!( + err.as_str().contains("invalid pair_id for action get"), + "{}", + err.as_str() + ); + } + + #[test] + fn pair_run_result_status_does_not_leak_internals() { + let status = RunPairStatusResponse { + run_id: run_id(), + current_pair: Some(PairRecord { + pair_id: pair_id(), + run_id: run_id(), + status: PairStatus::Active, + started_at: Utc::now(), + ended_at: None, + failure_reason: None, + target: pair_target(), + }), + targets: vec![pair_target()], + }; + let result = PairRunResult { + run_id: "run_123".to_string(), + action: RunPairAction::Status, + result: json!(status), + }; + let value = serde_json::to_value(&result).unwrap(); + assert_no_public_pair_leaks(&value); + } + + #[test] + fn pair_run_result_start_does_not_leak_internals() { + let record = PairRecord { + pair_id: pair_id(), + run_id: run_id(), + status: PairStatus::Active, + started_at: Utc::now(), + ended_at: None, + failure_reason: None, + target: pair_target(), + }; + let result = PairRunResult { + run_id: "run_123".to_string(), + action: RunPairAction::Start, + result: json!(record), + }; + let value = serde_json::to_value(&result).unwrap(); + assert_no_public_pair_leaks(&value); + } + + #[test] + fn pair_run_result_message_does_not_leak_internals() { + let record = PairMessageRecord { + message_id: PairMessageId::new(), + client_message_id: Some("c-1".to_string()), + pair_id: pair_id(), + run_id: run_id(), + stage_id: "code@1".parse().unwrap(), + text: "hi".to_string(), + accepted_at: Utc::now(), + }; + let result = PairRunResult { + run_id: "run_123".to_string(), + action: RunPairAction::Message, + result: json!(record), + }; + let value = serde_json::to_value(&result).unwrap(); + assert_no_public_pair_leaks(&value); + } + + #[test] + fn pair_run_result_transcript_does_not_leak_internals() { + let response = PairTranscriptResponse { + data: vec![PairTranscriptEntry::AssistantMessage( + PairTranscriptAssistantMessage { + seq: 7, + event_id: "evt".to_string(), + ts: Utc::now(), + pair_id: pair_id(), + target: pair_target(), + text: "hi".to_string(), + tool_call_count: 0, + }, + )], + meta: PairTranscriptMeta { + next_since_seq: 8, + has_more: false, + }, + }; + let result = PairRunResult { + run_id: "run_123".to_string(), + action: RunPairAction::Transcript, + result: json!(response), + }; + let value = serde_json::to_value(&result).unwrap(); + assert_no_public_pair_leaks(&value); + } +} diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 3f104088f..c711b5acd 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -71,14 +71,13 @@ pub use outcome::{ FailureCategory, FailureDetail, NodeResult, Outcome, OutcomeMeta, StageOutcome, StageState, }; pub use pair::{ - PairId, PairMessageId, PairMessageRecord, PairMessageRequest, PairRecord, PairStartRequest, - PairStatus, PairSystemMessageKind, PairTarget, PairTargetSelector, + MAX_PAIR_MESSAGE_BYTES, PairId, PairMessageId, PairMessageRecord, PairMessageRequest, + PairRecord, PairStartRequest, PairStatus, PairSystemMessageKind, PairTarget, PairTranscriptAssistantMessage, PairTranscriptDetailRef, PairTranscriptEntry, - PairTranscriptError, PairTranscriptMeta, PairTranscriptModel, PairTranscriptResponse, - PairTranscriptSystemMessage, PairTranscriptToolCall, PairTranscriptToolStatus, - PairTranscriptUserMessage, PairTranscriptWarning, RunEventDetailContent, - RunEventDetailContentKind, RunEventDetailEnvelope, RunEventDetailResponse, - RunPairStatusResponse, + PairTranscriptError, PairTranscriptMeta, PairTranscriptResponse, PairTranscriptSystemMessage, + PairTranscriptToolCall, PairTranscriptToolStatus, PairTranscriptUserMessage, + PairTranscriptWarning, RunEventDetailContent, RunEventDetailContentKind, + RunEventDetailEnvelope, RunEventDetailResponse, RunPairStatusResponse, }; pub use principal::{AuthMethod, Principal, SystemActorKind, UserPrincipal}; pub use pull_request::{ diff --git a/lib/crates/fabro-types/src/pair.rs b/lib/crates/fabro-types/src/pair.rs index a47c6eec5..72d8e60d5 100644 --- a/lib/crates/fabro-types/src/pair.rs +++ b/lib/crates/fabro-types/src/pair.rs @@ -9,6 +9,13 @@ use crate::{RunId, StageId}; ulid_id!(PairId); ulid_id!(PairMessageId); +/// Maximum byte length of a pair message text payload. +/// +/// Mirrored in `docs/public/api-reference/fabro-api.yaml` as the +/// `PairMessageRequest.text` `maxLength`. Keep the YAML and this constant in +/// sync. +pub const MAX_PAIR_MESSAGE_BYTES: usize = 8192; + #[derive( Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display, EnumString, IntoStaticStr, )] @@ -26,32 +33,10 @@ impl PairStatus { } } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct PairTargetSelector { - pub stage_id: StageId, - pub agent_session_id: String, -} - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct PairTarget { - pub stage_id: StageId, - pub node_id: String, - pub node_label: String, - pub visit: u32, - pub agent_session_id: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub provider: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub model: Option, -} - -impl PairTarget { - pub fn selector(&self) -> PairTargetSelector { - PairTargetSelector { - stage_id: self.stage_id.clone(), - agent_session_id: self.agent_session_id.clone(), - } - } + pub stage_id: StageId, + pub node_label: String, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -74,7 +59,7 @@ pub struct RunPairStatusResponse { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct PairStartRequest { - pub target: PairTargetSelector, + pub stage_id: StageId, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -91,7 +76,7 @@ pub struct PairMessageRecord { pub client_message_id: Option, pub pair_id: PairId, pub run_id: RunId, - pub target: PairTargetSelector, + pub stage_id: StageId, pub text: String, pub accepted_at: DateTime, } @@ -157,18 +142,9 @@ pub struct PairTranscriptAssistantMessage { pub pair_id: PairId, pub target: PairTarget, pub text: String, - pub model: PairTranscriptModel, pub tool_call_count: usize, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct PairTranscriptModel { - pub provider: String, - pub model_id: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub speed: Option, -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PairTranscriptToolCall { pub seq: u32, diff --git a/lib/crates/fabro-workflow/src/event/events.rs b/lib/crates/fabro-workflow/src/event/events.rs index adcbe216b..c53302e70 100644 --- a/lib/crates/fabro-workflow/src/event/events.rs +++ b/lib/crates/fabro-workflow/src/event/events.rs @@ -792,7 +792,12 @@ impl Event { Self::RunPairStarted { pair_id, target, .. } => { - info!(%pair_id, stage_id = %target.stage_id, session_id = target.agent_session_id, "Run pairing started"); + info!( + %pair_id, + stage_id = %target.stage_id, + node_label = %target.node_label, + "Run pairing started", + ); } Self::RunPairEnded { pair_id, reason, .. diff --git a/lib/crates/fabro-workflow/src/steering_hub.rs b/lib/crates/fabro-workflow/src/steering_hub.rs index d6cf9d6fa..2dc49e455 100644 --- a/lib/crates/fabro-workflow/src/steering_hub.rs +++ b/lib/crates/fabro-workflow/src/steering_hub.rs @@ -91,7 +91,12 @@ struct ActiveEntry { #[derive(Debug, Clone)] struct ActivePair { - record: PairRecord, + record: PairRecord, + /// Snapshot of the agent session id active at `start_pair` time. Used to + /// detect session replacement on subsequent pair commands and on + /// `AgentSessionDeactivated` cleanup; intentionally not exposed in the + /// public `PairRecord`. + session_id: String, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -404,12 +409,13 @@ impl SteeringHub { let Some(entry) = active.get(&target.stage_id) else { return Err(PairControlError::TargetNotActive); }; - if entry.session_id != target.agent_session_id { - return Err(PairControlError::TargetNotActive); - } let Some(pair_handle) = entry.pair_handle.as_ref() else { return Err(PairControlError::TargetNotActive); }; + let session_id = entry.session_id.clone(); + let interrupt_handle = Arc::clone(&entry.handle); + let pair_handle = pair_handle.clone(); + drop(active); let mut active_pair = self.active_pair.lock().expect("active pair lock poisoned"); if active_pair.is_some() { @@ -441,17 +447,18 @@ impl SteeringHub { actor: actor.clone(), }); - entry.handle.interrupt(actor); + interrupt_handle.interrupt(actor); self.emitter.emit(&Event::AgentPairSystemMessage { - node_id: record.target.node_id.clone(), - visit: record.target.visit, - session_id: record.target.agent_session_id.clone(), + node_id: record.target.stage_id.node_id().to_string(), + visit: record.target.stage_id.visit(), + session_id: session_id.clone(), pair_id, kind: PairSystemMessageKind::HumanJoined, text: text.to_string(), }); *active_pair = Some(ActivePair { record: record.clone(), + session_id, }); Ok(record) } @@ -475,12 +482,13 @@ impl SteeringHub { return Err(PairControlError::PairNotActive); } - let active = self.active.read().expect("active lock poisoned"); let target = &pair.record.target; + let session_id = pair.session_id.clone(); + let active = self.active.read().expect("active lock poisoned"); let Some(entry) = active.get(&target.stage_id) else { return Err(PairControlError::TargetNotActive); }; - if entry.session_id != target.agent_session_id { + if entry.session_id != session_id { return Err(PairControlError::TargetNotActive); } let Some(pair_handle) = entry.pair_handle.as_ref() else { @@ -494,9 +502,9 @@ impl SteeringHub { return Err(PairControlError::MessageNotAccepted); } self.emitter.emit(&Event::AgentPairUserMessage { - node_id: target.node_id.clone(), - visit: target.visit, - session_id: target.agent_session_id.clone(), + node_id: target.stage_id.node_id().to_string(), + visit: target.stage_id.visit(), + session_id, pair_id, message_id, client_message_id: client_message_id.clone(), @@ -508,7 +516,7 @@ impl SteeringHub { client_message_id, pair_id, run_id: pair.record.run_id, - target: target.selector(), + stage_id: target.stage_id.clone(), text, accepted_at: Utc::now(), }) @@ -531,13 +539,14 @@ impl SteeringHub { } let target = pair.record.target.clone(); + let session_id = pair.session_id.clone(); let text = human_left_text(); if let Some(entry) = self .active .read() .expect("active lock poisoned") .get(&target.stage_id) - .filter(|entry| entry.session_id == target.agent_session_id) + .filter(|entry| entry.session_id == session_id) { let Some(pair_handle) = entry.pair_handle.as_ref() else { return Err(PairControlError::TargetNotActive); @@ -551,9 +560,9 @@ impl SteeringHub { return Err(PairControlError::MessageNotAccepted); } self.emitter.emit(&Event::AgentPairSystemMessage { - node_id: target.node_id.clone(), - visit: target.visit, - session_id: target.agent_session_id.clone(), + node_id: target.stage_id.node_id().to_string(), + visit: target.stage_id.visit(), + session_id: session_id.clone(), pair_id, kind: PairSystemMessageKind::HumanLeft, text: text.to_string(), @@ -581,7 +590,7 @@ impl SteeringHub { .is_some_and(|pair| { pair.record.status == PairStatus::Active && pair.record.target.stage_id == *stage_id - && pair.record.target.agent_session_id == session_id + && pair.session_id == session_id }) } @@ -598,7 +607,7 @@ impl SteeringHub { }; if pair.record.status != PairStatus::Active || pair.record.target.stage_id != *stage_id - || pair.record.target.agent_session_id != session_id + || pair.session_id != session_id { return false; } @@ -700,15 +709,10 @@ mod tests { (Arc::new(SteeringHub::new(emitter)), events) } - fn pair_target(stage_id: &StageId, session_id: &str) -> PairTarget { + fn pair_target(stage_id: &StageId, _session_id: &str) -> PairTarget { PairTarget { - stage_id: stage_id.clone(), - node_id: stage_id.node_id().to_string(), - node_label: stage_id.node_id().to_string(), - visit: stage_id.visit(), - agent_session_id: session_id.to_string(), - provider: Some("openai".to_string()), - model: Some("gpt-5.4".to_string()), + stage_id: stage_id.clone(), + node_label: stage_id.node_id().to_string(), } } @@ -1010,16 +1014,17 @@ mod tests { } #[test] - fn pair_start_rejects_non_selected_or_missing_target() { + fn pair_start_rejects_missing_target() { let hub = SteeringHub::for_tests(); let stage_id = StageId::new("code", 1); let handle = SessionControlHandle::new(); assert!(hub.attach_pairable_handle(&stage_id, "ses_01", handle.clone())); + let missing_stage = StageId::new("other", 1); let result = hub.start_pair( RunId::new(), PairId::new(), - pair_target(&stage_id, "ses_02"), + pair_target(&missing_stage, "ses_01"), None, ); assert_eq!(result.unwrap_err(), PairControlError::TargetNotActive); diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index 3afce3e32..e50867659 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -199,9 +199,7 @@ models/pair-message-request.ts models/pair-record.ts models/pair-start-request.ts models/pair-status.ts -models/pair-target-selector.ts models/pair-target.ts -models/pair-transcript-assistant-message-model.ts models/pair-transcript-assistant-message.ts models/pair-transcript-detail-ref.ts models/pair-transcript-entry.ts diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 6bf39c541..ad36f4211 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -176,9 +176,7 @@ export * from './pair-record'; export * from './pair-start-request'; export * from './pair-status'; export * from './pair-target'; -export * from './pair-target-selector'; export * from './pair-transcript-assistant-message'; -export * from './pair-transcript-assistant-message-model'; export * from './pair-transcript-detail-ref'; export * from './pair-transcript-entry'; export * from './pair-transcript-error'; diff --git a/lib/packages/fabro-api-client/src/models/pair-message-record.ts b/lib/packages/fabro-api-client/src/models/pair-message-record.ts index 5f7407669..d8d413209 100644 --- a/lib/packages/fabro-api-client/src/models/pair-message-record.ts +++ b/lib/packages/fabro-api-client/src/models/pair-message-record.ts @@ -13,9 +13,6 @@ */ -// May contain unused imports in some cases -// @ts-ignore -import type { PairTargetSelector } from './pair-target-selector'; export interface PairMessageRecord { /** @@ -28,7 +25,7 @@ export interface PairMessageRecord { */ 'pair_id': string; 'run_id': string; - 'target': PairTargetSelector; + 'stage_id': string; 'text': string; 'accepted_at': string; } diff --git a/lib/packages/fabro-api-client/src/models/pair-start-request.ts b/lib/packages/fabro-api-client/src/models/pair-start-request.ts index 677571c21..73853b374 100644 --- a/lib/packages/fabro-api-client/src/models/pair-start-request.ts +++ b/lib/packages/fabro-api-client/src/models/pair-start-request.ts @@ -13,10 +13,7 @@ */ -// May contain unused imports in some cases -// @ts-ignore -import type { PairTargetSelector } from './pair-target-selector'; export interface PairStartRequest { - 'target': PairTargetSelector; + 'stage_id': string; } diff --git a/lib/packages/fabro-api-client/src/models/pair-target-selector.ts b/lib/packages/fabro-api-client/src/models/pair-target-selector.ts deleted file mode 100644 index 966bce5dd..000000000 --- a/lib/packages/fabro-api-client/src/models/pair-target-selector.ts +++ /dev/null @@ -1,20 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -export interface PairTargetSelector { - 'stage_id': string; - 'agent_session_id': string; -} diff --git a/lib/packages/fabro-api-client/src/models/pair-target.ts b/lib/packages/fabro-api-client/src/models/pair-target.ts index a6f271b9e..40513b26d 100644 --- a/lib/packages/fabro-api-client/src/models/pair-target.ts +++ b/lib/packages/fabro-api-client/src/models/pair-target.ts @@ -16,10 +16,5 @@ export interface PairTarget { 'stage_id': string; - 'node_id': string; 'node_label': string; - 'visit': number; - 'agent_session_id': string; - 'provider'?: string; - 'model'?: string; } diff --git a/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message-model.ts b/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message-model.ts deleted file mode 100644 index 4418424d8..000000000 --- a/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message-model.ts +++ /dev/null @@ -1,21 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -export interface PairTranscriptAssistantMessageModel { - 'provider': string; - 'model_id': string; - 'speed'?: string | null; -} diff --git a/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message.ts b/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message.ts index 725199548..b3d603fa4 100644 --- a/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message.ts +++ b/lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message.ts @@ -16,9 +16,6 @@ // May contain unused imports in some cases // @ts-ignore import type { PairTarget } from './pair-target'; -// May contain unused imports in some cases -// @ts-ignore -import type { PairTranscriptAssistantMessageModel } from './pair-transcript-assistant-message-model'; export interface PairTranscriptAssistantMessage { 'kind': PairTranscriptAssistantMessageKindEnum; @@ -31,7 +28,6 @@ export interface PairTranscriptAssistantMessage { 'pair_id': string; 'target': PairTarget; 'text': string; - 'model': PairTranscriptAssistantMessageModel; 'tool_call_count': number; } diff --git a/lib/packages/fabro-api-client/src/models/pair-transcript-entry.ts b/lib/packages/fabro-api-client/src/models/pair-transcript-entry.ts index b55e10c18..a6e3b8aa2 100644 --- a/lib/packages/fabro-api-client/src/models/pair-transcript-entry.ts +++ b/lib/packages/fabro-api-client/src/models/pair-transcript-entry.ts @@ -21,9 +21,6 @@ import type { PairTarget } from './pair-target'; import type { PairTranscriptAssistantMessage } from './pair-transcript-assistant-message'; // May contain unused imports in some cases // @ts-ignore -import type { PairTranscriptAssistantMessageModel } from './pair-transcript-assistant-message-model'; -// May contain unused imports in some cases -// @ts-ignore import type { PairTranscriptDetailRef } from './pair-transcript-detail-ref'; // May contain unused imports in some cases // @ts-ignore