From f5ec711a2c5960e9e6b291ec14360bec8a9dba9e Mon Sep 17 00:00:00 2001
From: "fabro-sh-0530[bot]"
<281434857+fabro-sh-0530[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 21:57:55 -0400
Subject: [PATCH] Stage-based pairing API and `fabro_run_pair` MCP tool (#344)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
## 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
Ran 9 stages in 60m 29s for $31.78
| 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** |
Ran ImplementPlan.fabro (12 nodes and 15
edges)
```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
}
```
⚒️ Generated with [Fabro](https://fabro.sh)
---------
Co-authored-by: Fabro
Co-authored-by: Bryan Helmkamp
---
Cargo.lock | 2 +
docs/public/api-reference/fabro-api.yaml | 69 +--
lib/crates/fabro-api/build.rs | 1 -
lib/crates/fabro-api/src/lib.rs | 2 +-
lib/crates/fabro-api/tests/pair_round_trip.rs | 57 +-
.../fabro-api/tests/run_event_round_trip.rs | 25 +-
lib/crates/fabro-cli/tests/it/cmd/mcp.rs | 19 +-
lib/crates/fabro-client/src/client.rs | 13 +-
.../fabro-interview/src/control_protocol.rs | 24 +-
lib/crates/fabro-mcp-server/Cargo.toml | 3 +-
lib/crates/fabro-mcp-server/src/server.rs | 62 ++
lib/crates/fabro-server/src/server.rs | 20 +-
.../fabro-server/src/server/handler/pair.rs | 80 +--
lib/crates/fabro-server/src/server/tests.rs | 9 +-
lib/crates/fabro-tool/Cargo.toml | 1 +
lib/crates/fabro-tool/src/common.rs | 45 +-
lib/crates/fabro-tool/src/fabro_client.rs | 48 +-
lib/crates/fabro-tool/src/lib.rs | 10 +-
lib/crates/fabro-tool/src/pair.rs | 564 ++++++++++++++++++
lib/crates/fabro-types/src/lib.rs | 13 +-
lib/crates/fabro-types/src/pair.rs | 46 +-
lib/crates/fabro-workflow/src/event/events.rs | 7 +-
lib/crates/fabro-workflow/src/steering_hub.rs | 65 +-
.../src/.openapi-generator/FILES | 2 -
.../fabro-api-client/src/models/index.ts | 2 -
.../src/models/pair-message-record.ts | 5 +-
.../src/models/pair-start-request.ts | 5 +-
.../src/models/pair-target-selector.ts | 20 -
.../src/models/pair-target.ts | 5 -
...pair-transcript-assistant-message-model.ts | 21 -
.../pair-transcript-assistant-message.ts | 4 -
.../src/models/pair-transcript-entry.ts | 3 -
32 files changed, 939 insertions(+), 313 deletions(-)
create mode 100644 lib/crates/fabro-tool/src/pair.rs
delete mode 100644 lib/packages/fabro-api-client/src/models/pair-target-selector.ts
delete mode 100644 lib/packages/fabro-api-client/src/models/pair-transcript-assistant-message-model.ts
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