Stage-based pairing API and fabro_run_pair MCP tool (#344)
Some checks are pending
Rust / Clippy (push) Waiting to run
Rust / Generated Docs (push) Waiting to run
Rust / Test (Linux) (push) Waiting to run
Rust / Test (macOS) (push) Waiting to run
Rust / Format (push) Waiting to run
TypeScript / Build (push) Waiting to run
TypeScript / Typecheck (push) Waiting to run
TypeScript / Test (push) Waiting to run

## Summary

Run pairing previously required callers to supply an opaque
`agent_session_id` alongside a `stage_id` to start or target a pair
session. This leaked an internal runtime identifier across the public
HTTP API, generated TypeScript client, and would have bled into any MCP
tooling. This PR removes that coupling: the public pair API now
identifies targets by `StageId` alone, the server resolves the live
session internally, and a new `fabro_run_pair` MCP tool exposes the full
pair lifecycle without ever seeing session identifiers.

### What changed

**Public contract simplification** (`fabro-types`, OpenAPI, generated TS
client)
- `PairTarget` is now `{ stage_id, node_label }` — `node_id`, `visit`,
`agent_session_id`, `provider`, and `model` are removed.
- `PairStartRequest` accepts `{ stage_id }` instead of `{ target:
PairTargetSelector }`.
- `PairTargetSelector` and `PairTranscriptModel` types are deleted
entirely.
- `PairMessageRecord.target` (selector) replaced by
`PairMessageRecord.stage_id`.
- `PairTranscriptAssistantMessage.model` field removed.
- `MAX_PAIR_MESSAGE_BYTES` extracted as a public constant shared between
the server handler and the MCP tool.

**Internal session binding** (`SteeringHub`, server projection)
- `ActivePair` now carries `session_id: String` separately from the
public `PairRecord`. This preserves the stale-session protection that
previously relied on `target.agent_session_id`.
- Transcript matching changed from `(session_id AND stage_id)` to
`stage_id` within the already-scoped pair window sequence range —
simpler and sufficient.
- `active_api_targets` deactivation no longer does a per-target
`agent_session_id` check; it relies on the `active_steerable_stages`
lease already doing that guard.

**New `fabro_run_pair` MCP tool** (`fabro-mcp-server`)
- Actions: `status`, `start`, `get`, `message`, `end`, `transcript`.
- Validation happens before any network call; missing `run_id`, missing
`stage_id` for `start`, missing/invalid `pair_id` for other actions, and
overlong message text all return clean tool-level errors.
- `strum::IntoStaticStr` on `RunPairAction` enables the
`parse_pair_id_for_action` helper to embed the action name in error
messages without a `match`.
- MCP result schema and serialized results are covered by leakage
assertions confirming none of the removed fields surface.

**Tests**
- Negative leakage assertions added to pair DTO tests, event round-trip
tests, control-protocol tests, server handler tests, MCP validation
tests, and MCP schema test.
- Tool count updated from 5 → 6 in all CLI MCP integration tests.
- Steering hub test renamed:
`pair_start_rejects_non_selected_or_missing_target` →
`pair_start_rejects_missing_target` (session-mismatch rejection is now
an internal concern).


### Fabro Details

<details>
<summary>Ran 9 stages in 60m 29s for $31.78</summary>

| Stage | Duration | Cost | Retries |
|---|---|---|---|
| start | 0s | – | 0 |
| toolchain | 1s | – | 0 |
| preflight_compile | 2m 9s | – | 0 |
| preflight_lint | 2m 26s | – | 0 |
| implement | 33m 14s | $25.51 | 0 |
| simplify_opus | 15m 7s | $4.18 | 0 |
| simplify_gpt | 3m 46s | $2.08 | 0 |
| verify | 3m 9s | – | 0 |
| fmt | 3s | – | 0 |
| **Total** | **60m 29s** | **$31.78** | **0** |

</details>

<details>
<summary>Ran <code>ImplementPlan.fabro</code> (12 nodes and 15
edges)</summary>

```dot
digraph ImplementPlan {
    graph [
        goal="Implement and simplify",
        model_stylesheet="
            * { model: claude-opus-4-7; }
        "
    ]
    rankdir=LR

    start [shape=Mdiamond, label="Start"]
    exit  [shape=Msquare, label="Exit"]

    toolchain         [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0]
    preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0]
    preflight_lint    [label="Preflight Lint", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1", max_retries=0]
    fix_lints         [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3]
    implement         [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD."]
    simplify_opus     [label="Simplify (Opus)", prompt="@prompts/simplify.md"]
    simplify_gpt      [label="Simplify (GPT-55)", prompt="@prompts/simplify.md", model="gpt-55"]
    verify            [label="Verify", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1 && cargo dev docs refresh 2>&1 && cargo dev docs check 2>&1", goal_gate=true, retry_target="fixup"]
    fixup             [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all clippy lint warnings, test failures, and generated docs errors.", max_visits=3]
    fmt               [label="Format", shape=parallelogram, script="cargo +nightly-2026-04-14 fmt --all 2>&1", max_retries=0]

    start -> toolchain
    toolchain -> preflight_compile [condition="outcome=succeeded"]
    toolchain -> exit
    preflight_compile -> preflight_lint [condition="outcome=succeeded"]
    preflight_compile -> exit
    preflight_lint -> implement [condition="outcome=succeeded"]
    preflight_lint -> fix_lints
    fix_lints -> preflight_lint
    implement -> simplify_opus -> simplify_gpt -> verify
    verify -> fmt   [condition="outcome=succeeded"]
    verify -> fixup
    fixup -> verify
    fmt -> exit
}

```

</details>

⚒️ Generated with [Fabro](https://fabro.sh)

---------

Co-authored-by: Fabro <noreply@fabro.sh>
Co-authored-by: Bryan Helmkamp <bryan@brynary.com>
This commit is contained in:
fabro-sh-0530[bot] 2026-05-21 21:57:55 -04:00 • committed by GitHub
parent 296fbddec9
commit f5ec711a2c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
32 changed files with 939 additions and 313 deletions

2
Cargo.lock generated
View file

@ -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",

View file

@ -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

View file

@ -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", &[]),
(

View file

@ -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,

View file

@ -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::<ApiPairId, PairId>();
assert_same_type::<ApiPairMessageId, PairMessageId>();
assert_same_type::<ApiPairStatus, PairStatus>();
assert_same_type::<ApiPairTargetSelector, PairTargetSelector>();
assert_same_type::<ApiPairTarget, PairTarget>();
assert_same_type::<ApiPairRecord, PairRecord>();
assert_same_type::<ApiRunPairStatusResponse, RunPairStatusResponse>();
@ -34,6 +33,44 @@ fn pair_api_reuses_canonical_types() {
assert_same_type::<ApiRunEventDetailResponse, RunEventDetailResponse>();
}
#[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::<PairRecord>(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"
})
}

View file

@ -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",

View file

@ -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()

View file

@ -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<PairRecord> {
let body = PairStartRequest { target };
pub async fn start_run_pair(&self, run_id: &RunId, stage_id: StageId) -> Result<PairRecord> {
let body = PairStartRequest { stage_id };
let response = self
.send_api(|client| {
let body = body.clone();

View file

@ -199,7 +199,7 @@ impl From<WorkerControlAnswer> 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]

View file

@ -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"

View file

@ -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<run_tools::FabroRunPairParams>,
) -> Result<CallToolResult, ErrorData> {
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<T: Serialize>(
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\""));
}
}

View file

@ -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

View file

@ -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<PairTarget> {
fn pair_target_and_transport(
state: &AppState,
id: &RunId,
selector: &fabro_types::PairTargetSelector,
stage_id: &StageId,
) -> Result<(PairTarget, Option<super::super::RunAnswerTransport>), 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,

View file

@ -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

View file

@ -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

View file

@ -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<RunPairStatusResponse> {
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<PairRecord> {
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<PairRecord> {
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<PairRecord> {
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<PairMessageRecord> {
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<u32>,
_limit: Option<u32>,
) -> anyhow::Result<PairTranscriptResponse> {
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<Vec<ToolDefinition>> = LazyLock::new(|| {
vec![

View file

@ -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<RunPairStatusResponse> {
self.client.get_run_pair_status(run_id).await
}
async fn start_run_pair(
&self,
run_id: &RunId,
stage_id: StageId,
) -> anyhow::Result<PairRecord> {
self.client.start_run_pair(run_id, stage_id).await
}
async fn get_run_pair(&self, run_id: &RunId, pair_id: &PairId) -> anyhow::Result<PairRecord> {
self.client.get_run_pair(run_id, pair_id).await
}
async fn end_run_pair(&self, run_id: &RunId, pair_id: &PairId) -> anyhow::Result<PairRecord> {
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<PairMessageRecord> {
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<u32>,
limit: Option<u32>,
) -> anyhow::Result<PairTranscriptResponse> {
self.client
.get_run_pair_transcript(run_id, pair_id, since_seq, limit)
.await
}
}

View file

@ -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,

View file

@ -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<String>,
pub pair_id: Option<String>,
pub stage_id: Option<String>,
pub text: Option<String>,
pub client_message_id: Option<String>,
pub since_seq: Option<u32>,
pub limit: Option<u32>,
}
#[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<String>,
},
End {
pair_id: PairId,
},
Transcript {
pair_id: PairId,
since_seq: Option<u32>,
limit: Option<u32>,
},
}
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<FabroRunPairParams> for ValidatedPairRun {
type Error = ToolError;
fn try_from(params: FabroRunPairParams) -> Result<Self, Self::Error> {
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::<StageId>().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<PairId> {
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::<PairId>()
.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<dyn FabroToolBackend>,
params: ValidatedPairRun,
) -> ToolResult<PairRunResult> {
let run_id = backend
.resolve_run(&params.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);
}
}

View file

@ -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::{

View file

@ -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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub model: Option<String>,
}
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<String>,
pub pair_id: PairId,
pub run_id: RunId,
pub target: PairTargetSelector,
pub stage_id: StageId,
pub text: String,
pub accepted_at: DateTime<Utc>,
}
@ -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<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PairTranscriptToolCall {
pub seq: u32,

View file

@ -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, ..

View file

@ -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);

View file

@ -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

View file

@ -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';

View file

@ -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;
}

View file

@ -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;
}

View file

@ -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;
}

View file

@ -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;
}

View file

@ -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;
}

View file

@ -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;
}

View file

@ -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