scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-13 23:14:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Merge pull request #857 from fabro-sh/pebble-mcp-events

Browse source 
Adopt pebble's MCP disconnect event, startup timing, and tool timeout kind
This commit is contained in:
Bryan Helmkamp 2026-09-12 13:31:48 -06:00 • committed by GitHub
commit 204c2b86f9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 584 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Cargo.lock generated
View file

@ -5875,7 +5875,7 @@ dependencies = [
[[package]]
name = "pebble-agent"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/pebble?rev=222d17f17d7384545d3782f3b315210ad2cb0cfe#222d17f17d7384545d3782f3b315210ad2cb0cfe"
source = "git+https://github.com/lithoscomputer/pebble?rev=4c0063327394cd0f1e9fee2c24541b829b93a8f7#4c0063327394cd0f1e9fee2c24541b829b93a8f7"
dependencies = [
"async-trait",
"futures-util",
@ -5892,7 +5892,7 @@ dependencies = [
[[package]]
name = "pebble-cli-core"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/pebble?rev=222d17f17d7384545d3782f3b315210ad2cb0cfe#222d17f17d7384545d3782f3b315210ad2cb0cfe"
source = "git+https://github.com/lithoscomputer/pebble?rev=4c0063327394cd0f1e9fee2c24541b829b93a8f7#4c0063327394cd0f1e9fee2c24541b829b93a8f7"
dependencies = [
"anyhow",
"async-trait",
@ -5921,7 +5921,7 @@ dependencies = [
[[package]]
name = "pebble-coding-agent"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/pebble?rev=222d17f17d7384545d3782f3b315210ad2cb0cfe#222d17f17d7384545d3782f3b315210ad2cb0cfe"
source = "git+https://github.com/lithoscomputer/pebble?rev=4c0063327394cd0f1e9fee2c24541b829b93a8f7#4c0063327394cd0f1e9fee2c24541b829b93a8f7"
dependencies = [
"async-trait",
"futures-util",

6
Cargo.toml
View file

@ -123,9 +123,9 @@ sandbox-driver-testing = { git = "https://github.com/lithoscomputer/sandbox-driv
# sandbox-driver revision, so move the two pins together. Pebble pins the
# same lithos-llm rev as fabro, and its lockfile policy is that every shared
# crate resolves to the version lithos-llm locks.
pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "222d17f17d7384545d3782f3b315210ad2cb0cfe" }
pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "222d17f17d7384545d3782f3b315210ad2cb0cfe", features = ["mcp", "search-providers"] }
pebble-cli-core = { git = "https://github.com/lithoscomputer/pebble", rev = "222d17f17d7384545d3782f3b315210ad2cb0cfe" }
pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "4c0063327394cd0f1e9fee2c24541b829b93a8f7" }
pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "4c0063327394cd0f1e9fee2c24541b829b93a8f7", features = ["mcp", "search-providers"] }
pebble-cli-core = { git = "https://github.com/lithoscomputer/pebble", rev = "4c0063327394cd0f1e9fee2c24541b829b93a8f7" }
sentry = { version = "0.35", default-features = false, features = ["backtrace", "contexts", "ureq", "rustls"] }
fork = "0.2"
exec = "0.3"

20
apps/fabro-web/app/components/stage-insights-sidebar.test.tsx
View file

@ -215,6 +215,26 @@ describe("StageInsightsSidebar", () => {
expect(dom).toContain("Failed");
});
test("renders a disconnected mcp server as disconnected, still counted as used", () => {
const dom = render(
makeStage({
mcp_servers: [
{
server_name: "github",
tool_count: 4,
status: { kind: "disconnected", error: "transport closed" },
invoked: true,
},
],
}),
null,
);
expect(dom).toContain("1/1");
expect(dom).toContain("github");
expect(dom).toContain("Disconnected");
expect(dom).not.toContain("Failed");
});
test("shows skill activated/available ratio with source label", () => {
const dom = render(
makeStage({

54
apps/fabro-web/app/components/stage-insights-sidebar.tsx
View file

@ -7,6 +7,7 @@ import {
import {
ArrowPathIcon,
CheckCircleIcon,
ExclamationTriangleIcon,
XCircleIcon,
} from "@heroicons/react/24/solid";
import {
@ -554,27 +555,15 @@ function McpSection({ servers }: { servers: McpServerProjection[] }) {
<ul className="space-y-1">
{servers.map((server) => {
// Dim unused servers so the eye lands on the invoked ones first;
// failed servers stay coral regardless.
// failed and disconnected servers keep their tone regardless.
const nameClass = server.status.kind === "ready" && !server.invoked
? "min-w-0 flex-1 truncate text-xs text-fg-muted"
: "min-w-0 flex-1 truncate text-xs text-fg-2";
return (
<li key={server.server_name} className="flex items-center gap-1.5">
{server.status.kind === "ready" ? (
<CheckCircleIcon className="size-3.5 shrink-0 text-mint" aria-label="Ready" />
) : (
<XCircleIcon className="size-3.5 shrink-0 text-coral" aria-label="Failed" />
)}
<McpStatusIcon status={server.status} />
<span className={nameClass}>{server.server_name}</span>
{server.status.kind === "ready" ? (
<span className="font-mono text-[10px] tabular-nums text-fg-muted">
{server.invoked
? "used"
: `${server.tool_count} ${server.tool_count === 1 ? "tool" : "tools"}`}
</span>
) : (
<span className="text-[10px] uppercase tracking-wider text-coral">Failed</span>
)}
<McpStatusBadge server={server} />
</li>
);
})}
@ -582,6 +571,41 @@ function McpSection({ servers }: { servers: McpServerProjection[] }) {
);
}
function McpStatusIcon({ status }: { status: McpServerProjection["status"] }) {
switch (status.kind) {
case "ready":
return <CheckCircleIcon className="size-3.5 shrink-0 text-mint" aria-label="Ready" />;
case "disconnected":
return (
<ExclamationTriangleIcon
className="size-3.5 shrink-0 text-amber"
aria-label="Disconnected"
/>
);
case "failed":
return <XCircleIcon className="size-3.5 shrink-0 text-coral" aria-label="Failed" />;
}
}
function McpStatusBadge({ server }: { server: McpServerProjection }) {
switch (server.status.kind) {
case "ready":
return (
<span className="font-mono text-[10px] tabular-nums text-fg-muted">
{server.invoked
? "used"
: `${server.tool_count} ${server.tool_count === 1 ? "tool" : "tools"}`}
</span>
);
case "disconnected":
return (
<span className="text-[10px] uppercase tracking-wider text-amber">Disconnected</span>
);
case "failed":
return <span className="text-[10px] uppercase tracking-wider text-coral">Failed</span>;
}
}
// ---------- helpers ----------
type IconType = (props: { className?: string }) => ReactNode;

37
docs/internal/events.md
View file

@ -1421,6 +1421,7 @@ Emitted when a sub-agent is spawned.
"original_name": "list_issues"
}
],
"startup_ms": 842,
"visit": 1
}
}
@ -1431,6 +1432,7 @@ Emitted when a sub-agent is spawned.
| `server_name` | string | MCP server name |
| `tool_count` | number | Number of tools available |
| `tools` | array | Names-only tool summaries for the ready server, sorted by qualified `name`. Each entry has `name` (Fabro-qualified `mcp__{server}__{tool}` identifier) and `original_name` (server-provided tool name). Descriptions and input schemas are intentionally omitted. The field is omitted from serialized JSON for legacy parity when empty. |
| `startup_ms` | number | Whole milliseconds from the server's launch to its tools being listed. Events written before the field existed read as `0`. |
| `visit` | number | Stage visit count when the server became ready |
### `agent.mcp.failed`
@ -1443,7 +1445,9 @@ Emitted when a sub-agent is spawned.
"session_id": "ses_abc",
"properties": {
"server_name": "filesystem",
"error": "Connection refused"
"error": "Connection refused",
"startup_ms": 4,
"visit": 1
}
}
```
@ -1452,6 +1456,37 @@ Emitted when a sub-agent is spawned.
|----------|------|-------------|
| `server_name` | string | MCP server name |
| `error` | string | Error message |
| `startup_ms` | number | Whole milliseconds from the server's launch to the failure. Events written before the field existed read as `0`. |
| `visit` | number | Stage visit count when the server failed |
### `agent.mcp.disconnected`
An MCP server that was ready lost its connection during the stage. Pebble
publishes the disconnect once per server, from whichever session's tool call
first observed the closed connection, so the event can originate in a
sub-agent. Every later call to that server's tools fails until the session
ends. The stage projection moves the server's status from `ready` to
`disconnected`; its `tool_count` and `invoked` flag are kept.
```json
{
"id": "...", "ts": "...", "run_id": "...",
"event": "agent.mcp.disconnected",
"node_id": "code", "node_label": "code",
"session_id": "ses_abc",
"properties": {
"server_name": "github",
"error": "transport closed",
"visit": 1
}
}
```
| Property | Type | Description |
|----------|------|-------------|
| `server_name` | string | MCP server name |
| `error` | string | What closed the connection, as the client observed it |
| `visit` | number | Stage visit count when the disconnect was observed |
### `agent.memory.loaded`

1
docs/internal/fabro-event-schema-v2-concrete-shape.md
View file

@ -386,6 +386,7 @@ V2 keeps the current durable family surface broadly intact.
- `agent.sub.closed`
- `agent.mcp.ready`
- `agent.mcp.failed`
- `agent.mcp.disconnected`
- `agent.failover`
### Git

16
docs/public/api-reference/fabro-api.yaml
View file

@ -11528,11 +11528,13 @@ components:
oneOf:
- $ref: "#/components/schemas/McpServerStatusReady"
- $ref: "#/components/schemas/McpServerStatusFailed"
- $ref: "#/components/schemas/McpServerStatusDisconnected"
discriminator:
propertyName: kind
mapping:
ready: "#/components/schemas/McpServerStatusReady"
failed: "#/components/schemas/McpServerStatusFailed"
disconnected: "#/components/schemas/McpServerStatusDisconnected"
McpServerStatusReady:
type: object
@ -11560,6 +11562,20 @@ components:
error:
type: string
McpServerStatusDisconnected:
description: The server was ready and then its connection closed during the stage; its tools fail until the session ends.
type: object
required:
- kind
- error
properties:
kind:
type: string
enum: [disconnected]
error:
type: string
description: What closed the connection, as the client observed it.
AgentMcpToolSummary:
description: Summary of one tool exposed by an MCP server.
type: object

220
lib/components/fabro-store/src/run_state.rs
View file

@ -729,6 +729,13 @@ impl RunProjectionReducer for RunProjection {
invoked: false,
});
}
EventBody::AgentMcpDisconnected(props) => {
let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
else {
return Ok(());
};
mark_mcp_server_disconnected(stage, &props.server_name, &props.error);
}
_ => {}
}
@ -1110,6 +1117,30 @@ fn upsert_mcp_server(stage: &mut StageProjection, mut server: McpServerProjectio
}
}
/// Move a server the stage saw come up to `Disconnected`. Its tool count and
/// sticky `invoked` flag stay: the tools existed and may have been used, they
/// only fail from here on. A disconnect for a server the stage never saw come
/// up is still recorded, without tools.
fn mark_mcp_server_disconnected(stage: &mut StageProjection, server_name: &str, error: &str) {
let status = McpServerStatus::Disconnected {
error: error.to_string(),
};
if let Some(existing) = stage
.mcp_servers
.iter_mut()
.find(|existing| existing.server_name == server_name)
{
existing.status = status;
} else {
stage.mcp_servers.push(McpServerProjection {
server_name: server_name.to_string(),
tool_count: 0,
status,
invoked: false,
});
}
}
/// Extract the `<server>` segment from an `mcp__<server>__<tool>` qualified
/// tool name. Returns `None` for non-MCP tools or malformed names.
fn mcp_server_from_tool_name(tool_name: &str) -> Option<&str> {
@ -1796,12 +1827,13 @@ mod tests {
use fabro_types::run_event::run::RunFailedProps;
use fabro_types::run_event::{
AgentAcpCancelledProps, AgentAcpCompletedProps, AgentAcpStartedProps,
AgentAcpTimedOutProps, AgentEventProps, AgentMcpFailedProps, AgentMcpReadyProps,
AgentMcpToolSummary, AgentSessionActivatedProps, AgentSessionDeactivatedProps,
AgentToolsAvailableProps, CheckpointCompletedProps, InterviewCompletedProps,
InterviewOption, InterviewStartedProps, ParallelBranchCompletedProps,
ParallelBranchStartedProps, RunCompletedProps, RunControlEffectProps, StageCompletedProps,
StageFailedProps, StagePromptProps, StageRetryingProps, StageStartedProps,
AgentAcpTimedOutProps, AgentEventProps, AgentMcpDisconnectedProps, AgentMcpFailedProps,
AgentMcpReadyProps, AgentMcpToolSummary, AgentSessionActivatedProps,
AgentSessionDeactivatedProps, AgentToolsAvailableProps, CheckpointCompletedProps,
InterviewCompletedProps, InterviewOption, InterviewStartedProps,
ParallelBranchCompletedProps, ParallelBranchStartedProps, RunCompletedProps,
RunControlEffectProps, StageCompletedProps, StageFailedProps, StagePromptProps,
StageRetryingProps, StageStartedProps,
};
use fabro_types::settings::run::DockerfileSource;
use fabro_types::{
@ -7324,6 +7356,7 @@ mod tests {
original_name: "write_file".to_string(),
},
],
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7335,6 +7368,7 @@ mod tests {
EventBody::AgentMcpFailed(AgentMcpFailedProps {
server_name: "github".to_string(),
error: "missing token".to_string(),
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7350,6 +7384,7 @@ mod tests {
name: "read_file".to_string(),
original_name: "read_file".to_string(),
}],
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7375,6 +7410,87 @@ mod tests {
assert!(!stage.mcp_servers[1].invoked);
}
#[test]
fn mcp_server_disconnect_keeps_tool_count_and_invoked() {
let mut state = initialized_projection();
let stage_id = stage_id();
state
.apply_event(&test_stage_event(
1,
EventBody::AgentMcpReady(AgentMcpReadyProps {
server_name: "github".to_string(),
tool_count: 1,
tools: vec![AgentMcpToolSummary {
name: "mcp__github__list_issues".to_string(),
original_name: "list_issues".to_string(),
}],
startup_ms: 842,
visit: 1,
}),
stage_id.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event(
2,
agent_body(CodingEvent::ToolCallStarted {
tool_name: "mcp__github__list_issues".to_string(),
tool_call_id: "call_gh".to_string(),
arguments: serde_json::json!({}),
}),
stage_id.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event(
3,
EventBody::AgentMcpDisconnected(AgentMcpDisconnectedProps {
server_name: "github".to_string(),
error: "transport closed".to_string(),
visit: 1,
}),
stage_id.clone(),
))
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.mcp_servers.len(), 1);
let github = &stage.mcp_servers[0];
assert_eq!(github.server_name, "github");
assert_eq!(github.status, McpServerStatus::Disconnected {
error: "transport closed".to_string(),
});
assert_eq!(github.tool_count, 1, "the tools existed; they now fail");
assert!(github.invoked, "the server was used before it dropped");
}
#[test]
fn mcp_server_disconnect_without_a_ready_is_recorded_without_tools() {
let mut state = initialized_projection();
let stage_id = stage_id();
state
.apply_event(&test_stage_event(
1,
EventBody::AgentMcpDisconnected(AgentMcpDisconnectedProps {
server_name: "github".to_string(),
error: "transport closed".to_string(),
visit: 1,
}),
stage_id.clone(),
))
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.mcp_servers.len(), 1);
assert_eq!(stage.mcp_servers[0].tool_count, 0);
assert!(!stage.mcp_servers[0].invoked);
assert_eq!(stage.mcp_servers[0].status, McpServerStatus::Disconnected {
error: "transport closed".to_string(),
});
}
#[test]
fn agent_tool_started_marks_matching_mcp_server_as_invoked() {
let mut state = initialized_projection();
@ -7390,6 +7506,7 @@ mod tests {
name: "read_file".to_string(),
original_name: "read_file".to_string(),
}],
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7402,6 +7519,7 @@ mod tests {
server_name: "other".to_string(),
tool_count: 0,
tools: vec![],
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7462,6 +7580,7 @@ mod tests {
name: "read_file".to_string(),
original_name: "read_file".to_string(),
}],
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7496,6 +7615,7 @@ mod tests {
original_name: "stat".to_string(),
},
],
startup_ms: 0,
visit: 1,
}),
stage_id.clone(),
@ -7851,7 +7971,7 @@ mod tests {
/// other for a retained session that spans two stages, so a stage's live
/// account is the prompt delta pebble reports and the two never drift.
mod session_projection_parity {
use pebble_coding_agent::events::InputSource;
use pebble_coding_agent::events::{InputSource, McpToolSummary};
use pebble_coding_agent::projection::{
SessionActivity, SessionProjection, SubagentStatus as PebbleSubagentStatus,
};
@ -8066,5 +8186,91 @@ mod tests {
assert_eq!(resumed, replayed);
}
/// Pebble folds its own `McpServer*` events; fabro folds the
/// `agent.mcp.*` events the workflow sink mirrors them onto, since
/// the raw pebble event is not stored. The mirrored events are built
/// here the way the sink builds them.
#[test]
fn mcp_servers_agree_across_the_two_folds() {
let code = StageId::new("code", 1);
let tools = vec![McpToolSummary {
name: "mcp__github__list_issues".to_string(),
original_name: "list_issues".to_string(),
}];
let ready = root(CodingEvent::McpServerReady {
server: "github".to_string(),
tools: tools.clone(),
startup_ms: 842,
});
let call = root(CodingEvent::ToolCallStarted {
tool_name: "mcp__github__list_issues".to_string(),
tool_call_id: "call_1".to_string(),
arguments: json!({}),
});
// The child's call is the one that sees the connection close.
let disconnected = child(CodingEvent::McpServerDisconnected {
server: "github".to_string(),
error: "transport closed".to_string(),
});
let mut projection = SessionProjection::new();
projection.apply(&ready);
projection.apply(&call);
// A projection stored between the two carries the disconnect on
// resume like the replayed one.
let stored_bytes = serde_json::to_vec(&projection).unwrap();
projection.apply(&disconnected);
let mut resumed: SessionProjection = serde_json::from_slice(&stored_bytes).unwrap();
resumed.apply(&disconnected);
assert_eq!(resumed, projection);
let mut run = initialized_projection();
run.apply_event(&test_stage_event(
1,
EventBody::AgentMcpReady(AgentMcpReadyProps {
server_name: "github".to_string(),
tool_count: tools.len(),
tools: tools
.iter()
.map(|tool| AgentMcpToolSummary {
name: tool.name.clone(),
original_name: tool.original_name.clone(),
})
.collect(),
startup_ms: 842,
visit: 1,
}),
code.clone(),
))
.unwrap();
run.apply_event(&stored(2, &code, call)).unwrap();
run.apply_event(&test_stage_event(
3,
EventBody::AgentMcpDisconnected(AgentMcpDisconnectedProps {
server_name: "github".to_string(),
error: "transport closed".to_string(),
visit: 1,
}),
code.clone(),
))
.unwrap();
let stage = run.stage(&code).unwrap();
assert_eq!(stage.mcp_servers.len(), projection.mcp_servers.len());
let server = &stage.mcp_servers[0];
let pebble = &projection.mcp_servers["github"];
assert_eq!(server.server_name, "github");
assert_eq!(server.tool_count, pebble.tools.len());
assert_eq!(server.invoked, pebble.invoked);
assert!(server.invoked);
assert_eq!(pebble.error, None, "a disconnect is not a failed start");
assert_eq!(server.status, McpServerStatus::Disconnected {
error: pebble
.disconnected
.clone()
.expect("pebble recorded the disconnect"),
});
}
}
}

14
lib/components/fabro-workflow/src/event/convert.rs
View file

@ -844,19 +844,33 @@ fn event_body_from_event(event: &Event) -> EventBody {
server_name,
tool_count,
tools,
startup_ms,
..
} => EventBody::AgentMcpReady(fabro_types::AgentMcpReadyProps {
server_name: server_name.clone(),
tool_count: *tool_count,
tools: tools.clone(),
startup_ms: *startup_ms,
visit: *visit,
}),
Event::AgentMcpFailed {
visit,
server_name,
error,
startup_ms,
..
} => EventBody::AgentMcpFailed(fabro_types::AgentMcpFailedProps {
server_name: server_name.clone(),
error: error.clone(),
startup_ms: *startup_ms,
visit: *visit,
}),
Event::AgentMcpDisconnected {
visit,
server_name,
error,
..
} => EventBody::AgentMcpDisconnected(fabro_types::AgentMcpDisconnectedProps {
server_name: server_name.clone(),
error: error.clone(),
visit: *visit,

37
lib/components/fabro-workflow/src/event/events.rs
View file

@ -651,6 +651,9 @@ pub enum Event {
server_name: String,
tool_count: usize,
tools: Vec<fabro_types::AgentMcpToolSummary>,
/// Whole milliseconds from launch to the tools being listed.
#[serde(default)]
startup_ms: u64,
},
/// An MCP server configured for a stage failed to start or connect.
AgentMcpFailed {
@ -658,6 +661,17 @@ pub enum Event {
visit: u32,
server_name: String,
error: String,
/// Whole milliseconds from launch to the failure.
#[serde(default)]
startup_ms: u64,
},
/// An MCP server that was ready lost its connection during the stage;
/// its tools fail until the session ends.
AgentMcpDisconnected {
node_id: String,
visit: u32,
server_name: String,
error: String,
},
/// A run-level interrupt was delivered to a concrete steerable agent
/// session/stage.
@ -1534,17 +1548,36 @@ impl Event {
visit,
server_name,
tool_count,
startup_ms,
..
} => {
debug!(node_id, visit, server_name, tool_count, "MCP server ready");
debug!(
node_id,
visit, server_name, tool_count, startup_ms, "MCP server ready"
);
}
Self::AgentMcpFailed {
node_id,
visit,
server_name,
error,
startup_ms,
} => {
warn!(node_id, visit, server_name, error, "MCP server failed");
warn!(
node_id,
visit, server_name, error, startup_ms, "MCP server failed"
);
}
Self::AgentMcpDisconnected {
node_id,
visit,
server_name,
error,
} => {
warn!(
node_id,
visit, server_name, error, "MCP server disconnected"
);
}
Self::AgentInterruptInjected {
node_id,

10
lib/components/fabro-workflow/src/event/names.rs
View file

@ -90,6 +90,7 @@ pub fn event_name(event: &Event) -> Cow<'static, str> {
Event::AgentSessionDeactivated { .. } => "agent.session.deactivated",
Event::AgentMcpReady { .. } => "agent.mcp.ready",
Event::AgentMcpFailed { .. } => "agent.mcp.failed",
Event::AgentMcpDisconnected { .. } => "agent.mcp.disconnected",
Event::AgentInterruptInjected { .. } => "agent.interrupt.injected",
Event::AgentPairUserMessage { .. } => "agent.pair.user_message",
Event::AgentPairSystemMessage { .. } => "agent.pair.system_message",
@ -130,6 +131,15 @@ mod tests {
}),
"parallel.branch.started"
);
assert_eq!(
event_name(&Event::AgentMcpDisconnected {
node_id: "code".to_string(),
visit: 1,
server_name: "github".to_string(),
error: "transport closed".to_string(),
}),
"agent.mcp.disconnected"
);
assert_eq!(
event_name(&Event::Agent {
stage: "code".to_string(),

3
lib/components/fabro-workflow/src/event/stored_fields.rs
View file

@ -134,7 +134,8 @@ fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields {
| Event::AgentAcpTimedOut { node_id, .. } => node_stored_fields(Some(node_id.clone())),
Event::AgentAcpStarted { node_id, visit, .. }
| Event::AgentMcpReady { node_id, visit, .. }
| Event::AgentMcpFailed { node_id, visit, .. } => {
| Event::AgentMcpFailed { node_id, visit, .. }
| Event::AgentMcpDisconnected { node_id, visit, .. } => {
let node_id_str = node_id.clone();
let node_label = default_node_label(Some(&node_id_str), None);
StoredEventFields {

33
lib/components/fabro-workflow/src/handler/llm/pebble.rs
View file

@ -164,9 +164,10 @@ fn classify_agent_error(error: pebble_coding_agent::Error) -> AgentErrorDisposit
/// Pebble's durable event sink for one stage: every agent event becomes a
/// run event in the run's log before the agent goes on. A route failover and
/// an MCP server's outcome are facts the run already has events for, so
/// those are mirrored onto the run's own `agent.failover`, `agent.mcp.ready`,
/// and `agent.mcp.failed` events instead of being stored twice.
/// an MCP server's outcome or disconnect are facts the run already has
/// events for, so those are mirrored onto the run's own `agent.failover`,
/// `agent.mcp.ready`, `agent.mcp.failed`, and `agent.mcp.disconnected`
/// events instead of being stored twice.
struct WorkflowEventSink {
emitter: Arc<Emitter>,
node_id: String,
@ -198,7 +199,11 @@ impl EventSink for WorkflowEventSink {
);
return Ok(());
}
CodingEvent::McpServerReady { server, tools } => {
CodingEvent::McpServerReady {
server,
tools,
startup_ms,
} => {
self.emitter.emit_scoped(
&Event::AgentMcpReady {
node_id: self.node_id.clone(),
@ -212,18 +217,36 @@ impl EventSink for WorkflowEventSink {
original_name: tool.original_name.clone(),
})
.collect(),
startup_ms: *startup_ms,
},
&self.scope,
);
return Ok(());
}
CodingEvent::McpServerFailed { server, error } => {
CodingEvent::McpServerFailed {
server,
error,
startup_ms,
} => {
self.emitter.emit_scoped(
&Event::AgentMcpFailed {
node_id: self.node_id.clone(),
visit: self.scope.visit,
server_name: server.clone(),
error: error.clone(),
startup_ms: *startup_ms,
},
&self.scope,
);
return Ok(());
}
CodingEvent::McpServerDisconnected { server, error } => {
self.emitter.emit_scoped(
&Event::AgentMcpDisconnected {
node_id: self.node_id.clone(),
visit: self.scope.visit,
server_name: server.clone(),
error: error.clone(),
},
&self.scope,
);

25
lib/foundation/fabro-api/tests/stage_projection_round_trip.rs
View file

@ -528,6 +528,31 @@ fn nested_agent_state_types_match_openapi_json_shape() {
let api_mcp: ApiMcpServerProjection = serde_json::from_value(mcp_json).unwrap();
assert_eq!(api_mcp, mcp_server);
assert_eq!(mcp_server.tool_count, 1);
let disconnected = McpServerProjection {
server_name: "filesystem".to_string(),
tool_count: 1,
status: McpServerStatus::Disconnected {
error: "transport closed".to_string(),
},
invoked: true,
};
let disconnected_json = serde_json::to_value(&disconnected).unwrap();
assert_eq!(
disconnected_json,
json!({
"server_name": "filesystem",
"tool_count": 1,
"status": {
"kind": "disconnected",
"error": "transport closed"
},
"invoked": true,
})
);
let api_disconnected: ApiMcpServerProjection =
serde_json::from_value(disconnected_json).unwrap();
assert_eq!(api_disconnected, disconnected);
}
#[test]

26
lib/foundation/fabro-types/src/run_event/agent.rs
View file

@ -79,6 +79,7 @@ pub fn coding_event_name(event: &CodingEvent) -> &'static str {
CodingEvent::RouteFailover { .. } => "agent.route.failover",
CodingEvent::McpServerReady { .. } => "agent.mcp.server.ready",
CodingEvent::McpServerFailed { .. } => "agent.mcp.server.failed",
CodingEvent::McpServerDisconnected { .. } => "agent.mcp.server.disconnected",
CodingEvent::SteeringInjected { .. } => "agent.steering.injected",
CodingEvent::RoundInterrupted { .. } => "agent.round.interrupted",
CodingEvent::CompactionStarted { .. } => "agent.compaction.started",
@ -126,6 +127,7 @@ pub const CODING_EVENT_NAMES: &[&str] = &[
"agent.route.failover",
"agent.mcp.server.ready",
"agent.mcp.server.failed",
"agent.mcp.server.disconnected",
"agent.steering.injected",
"agent.round.interrupted",
"agent.compaction.started",
@ -238,6 +240,10 @@ pub struct AgentMcpReadyProps {
pub tool_count: usize,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub tools: Vec<AgentMcpToolSummary>,
/// Whole milliseconds from the server's launch to its tools being
/// listed. Events written before the field existed read as `0`.
#[serde(default)]
pub startup_ms: u64,
pub visit: u32,
}
@ -251,6 +257,22 @@ pub struct AgentMcpToolSummary {
pub struct AgentMcpFailedProps {
pub server_name: String,
pub error: String,
/// Whole milliseconds from the server's launch to the failure. Events
/// written before the field existed read as `0`.
#[serde(default)]
pub startup_ms: u64,
pub visit: u32,
}
/// An MCP server that was ready lost its connection during the stage; every
/// later call to its tools fails until the session ends. Pebble reports the
/// disconnect once per server, from whichever session's tool call first
/// observed the closed connection.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AgentMcpDisconnectedProps {
pub server_name: String,
/// What closed the connection, as the client observed it.
pub error: String,
pub visit: u32,
}
@ -313,6 +335,10 @@ mod tests {
CodingEvent::SessionEnded,
CodingEvent::ProcessingEnd,
CodingEvent::LoopDetected,
CodingEvent::McpServerDisconnected {
server: "github".to_string(),
error: "transport closed".to_string(),
},
CodingEvent::AssistantMessage {
text: String::new(),
model: "gpt-5.4".to_string(),

64
lib/foundation/fabro-types/src/run_event/mod.rs
View file

@ -221,6 +221,8 @@ pub enum EventBody {
AgentMcpReady(AgentMcpReadyProps),
#[serde(rename = "agent.mcp.failed")]
AgentMcpFailed(AgentMcpFailedProps),
#[serde(rename = "agent.mcp.disconnected")]
AgentMcpDisconnected(AgentMcpDisconnectedProps),
#[serde(rename = "subgraph.started")]
SubgraphStarted(SubgraphStartedProps),
#[serde(rename = "subgraph.completed")]
@ -508,6 +510,7 @@ impl EventBody {
Self::AgentSteerDropped(_) => "agent.steer.dropped",
Self::AgentMcpReady(_) => "agent.mcp.ready",
Self::AgentMcpFailed(_) => "agent.mcp.failed",
Self::AgentMcpDisconnected(_) => "agent.mcp.disconnected",
Self::SubgraphStarted(_) => "subgraph.started",
Self::SubgraphCompleted(_) => "subgraph.completed",
Self::SandboxInitializing(_) => "sandbox.initializing",
@ -640,6 +643,7 @@ fn is_known_event_name(event: &str) -> bool {
| "agent.steer.dropped"
| "agent.mcp.ready"
| "agent.mcp.failed"
| "agent.mcp.disconnected"
| "subgraph.started"
| "subgraph.completed"
| "sandbox.initializing"
@ -2258,6 +2262,7 @@ mod tests {
name: "mcp__github__create_issue".to_string(),
original_name: "create_issue".to_string(),
}],
startup_ms: 0,
visit: 1,
});
let value = serde_json::to_value(&body).unwrap();
@ -2272,12 +2277,71 @@ mod tests {
);
}
#[test]
fn agent_mcp_ready_and_failed_carry_startup_ms_and_default_it_when_absent() {
let ready = EventBody::AgentMcpReady(AgentMcpReadyProps {
server_name: "github".to_string(),
tool_count: 0,
tools: Vec::new(),
startup_ms: 842,
visit: 1,
});
let value = serde_json::to_value(&ready).unwrap();
assert_eq!(value["properties"]["startup_ms"], 842);
let failed = EventBody::AgentMcpFailed(AgentMcpFailedProps {
server_name: "filesystem".to_string(),
error: "could not launch `npx`".to_string(),
startup_ms: 4,
visit: 1,
});
let value = serde_json::to_value(&failed).unwrap();
assert_eq!(value["properties"]["startup_ms"], 4);
// Events written before pebble reported startup time.
let legacy: EventBody = serde_json::from_value(json!({
"event": "agent.mcp.failed",
"properties": {
"server_name": "filesystem",
"error": "Connection refused",
"visit": 1
}
}))
.unwrap();
match legacy {
EventBody::AgentMcpFailed(props) => assert_eq!(props.startup_ms, 0),
other => panic!("unexpected body: {other:?}"),
}
}
#[test]
fn agent_mcp_disconnected_round_trips() {
let body = EventBody::AgentMcpDisconnected(AgentMcpDisconnectedProps {
server_name: "github".to_string(),
error: "transport closed".to_string(),
visit: 1,
});
let value = serde_json::to_value(&body).unwrap();
assert_eq!(value["event"], "agent.mcp.disconnected");
assert_eq!(
value["properties"],
json!({
"server_name": "github",
"error": "transport closed",
"visit": 1
})
);
let parsed: EventBody = serde_json::from_value(value).unwrap();
assert_eq!(parsed, body);
}
#[test]
fn agent_mcp_ready_omits_tools_when_empty() {
let body = EventBody::AgentMcpReady(AgentMcpReadyProps {
server_name: "github".to_string(),
tool_count: 0,
tools: Vec::new(),
startup_ms: 0,
visit: 1,
});
let value = serde_json::to_value(&body).unwrap();

13
lib/foundation/fabro-types/src/run_projection.rs
View file

@ -450,8 +450,17 @@ pub struct McpServerProjection {
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum McpServerStatus {
Ready { tools: Vec<AgentMcpToolSummary> },
Failed { error: String },
Ready {
tools: Vec<AgentMcpToolSummary>,
},
Failed {
error: String,
},
/// The server was ready and then its connection closed during the
/// stage; its tools fail until the session ends.
Disconnected {
error: String,
},
}
/// Convert a 1-based event sequence number into the `NonZeroU32` form used for

1
lib/packages/fabro-api-client/src/.openapi-generator/FILES generated
View file

@ -215,6 +215,7 @@ models/mcp-server-list-meta.ts
models/mcp-server-list-response.ts
models/mcp-server-projection.ts
models/mcp-server-settings.ts
models/mcp-server-status-disconnected.ts
models/mcp-server-status-failed.ts
models/mcp-server-status-ready.ts
models/mcp-server-status.ts

1
lib/packages/fabro-api-client/src/models/index.ts generated
View file

@ -187,6 +187,7 @@ export * from './mcp-server-list-response';
export * from './mcp-server-projection';
export * from './mcp-server-settings';
export * from './mcp-server-status';
export * from './mcp-server-status-disconnected';
export * from './mcp-server-status-failed';
export * from './mcp-server-status-ready';
export * from './mcp-transport';

32
lib/packages/fabro-api-client/src/models/mcp-server-status-disconnected.ts generated Normal file
View file

@ -0,0 +1,32 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* The server was ready and then its connection closed during the stage; its tools fail until the session ends.
*/
export interface McpServerStatusDisconnected {
'kind': McpServerStatusDisconnectedKindEnum;
/**
* What closed the connection, as the client observed it.
*/
'error': string;
}
export const McpServerStatusDisconnectedKindEnum = {
DISCONNECTED: 'disconnected'
} as const;
export type McpServerStatusDisconnectedKindEnum = typeof McpServerStatusDisconnectedKindEnum[keyof typeof McpServerStatusDisconnectedKindEnum];

5
lib/packages/fabro-api-client/src/models/mcp-server-status.ts generated
View file

@ -18,6 +18,9 @@
import type { AgentMcpToolSummary } from './agent-mcp-tool-summary';
// May contain unused imports in some cases
// @ts-ignore
import type { McpServerStatusDisconnected } from './mcp-server-status-disconnected';
// May contain unused imports in some cases
// @ts-ignore
import type { McpServerStatusFailed } from './mcp-server-status-failed';
// May contain unused imports in some cases
// @ts-ignore
@ -27,4 +30,4 @@ import type { McpServerStatusReady } from './mcp-server-status-ready';
* @type McpServerStatus
* Projected MCP server readiness status.
*/
export type McpServerStatus = { kind: 'failed' } & McpServerStatusFailed | { kind: 'ready' } & McpServerStatusReady;
export type McpServerStatus = { kind: 'disconnected' } & McpServerStatusDisconnected | { kind: 'failed' } & McpServerStatusFailed | { kind: 'ready' } & McpServerStatusReady;