scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix: clean up inference observability

Browse source
This commit is contained in:
Bryan Helmkamp 2026-07-24 22:50:00 -04:00
parent e1d0b1af4f
commit d4f619bc2a
No known key found for this signature in database
36 changed files with 484 additions and 249 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
apps/fabro-web/app/components/stage-inference-indicator.test.tsx
View file

@ -1,4 +1,4 @@
import { describe, expect, test } from "bun:test";
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
import TestRenderer, { act } from "react-test-renderer";
import { LlmOutputKind } from "@qltysh/fabro-api-client";
@ -17,7 +17,7 @@ function makeInference(
requested_model: {
provider: "anthropic",
model_id: "claude-fable-5",
} as StageInferenceProjection["requested_model"],
},
retries: 0,
...overrides,
};
@ -27,17 +27,35 @@ function render(
inference: StageInferenceProjection | null | undefined,
settled = false,
): string {
(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
let renderer!: TestRenderer.ReactTestRenderer;
act(() => {
renderer = TestRenderer.create(
<StageInferenceIndicator inference={inference} settled={settled} />,
);
});
return JSON.stringify(renderer.toJSON());
const output = JSON.stringify(renderer.toJSON());
act(() => renderer.unmount());
return output;
}
describe("StageInferenceIndicator", () => {
const actGlobal = globalThis as {
IS_REACT_ACT_ENVIRONMENT?: boolean;
};
const previousActEnvironment = actGlobal.IS_REACT_ACT_ENVIRONMENT;
beforeEach(() => {
actGlobal.IS_REACT_ACT_ENVIRONMENT = true;
});
afterEach(() => {
if (previousActEnvironment === undefined) {
delete actGlobal.IS_REACT_ACT_ENVIRONMENT;
} else {
actGlobal.IS_REACT_ACT_ENVIRONMENT = previousActEnvironment;
}
});
test("renders nothing without an open bracket", () => {
expect(render(undefined)).toBe("null");
expect(render(null)).toBe("null");
@ -47,6 +65,8 @@ describe("StageInferenceIndicator", () => {
const output = render(makeInference());
expect(output).toContain("Model request");
expect(output).toContain("waiting on claude-fable-5");
expect(output).toContain('"aria-live":"polite"');
expect(output).toContain('"aria-hidden":"true"');
// No completion estimate exists, so none may be shown.
expect(output).not.toContain("%");
});

17
apps/fabro-web/app/components/stage-inference-indicator.tsx
View file

@ -57,13 +57,14 @@ export function StageInferenceIndicator({
? ACTIVITY_LABEL[inference.first_output_kind]
: `waiting on ${inference.requested_model.model_id}`;
const parts = ["Model request", activity];
if (elapsedSecs !== null) parts.push(formatDurationSecs(elapsedSecs));
const statusParts = ["Model request", activity];
// A retry that later succeeds is normal, so this is a count, not a failure.
if (inference.retries > 0) parts.push(`retry ${inference.retries}`);
if (inference.retries > 0) {
statusParts.push(`retry ${inference.retries}`);
}
return (
<p className="pb-2 text-xs text-fg-muted" aria-live="polite">
<p className="pb-2 text-xs text-fg-muted">
<Tooltip
label={`Model request opened ${formatAbsoluteTs(inference.started_at)}`}
>
@ -72,7 +73,13 @@ export function StageInferenceIndicator({
className="size-1.5 animate-pulse rounded-full bg-teal-500"
aria-hidden="true"
/>
{parts.join(" · ")}
<span aria-live="polite">{statusParts.join(" · ")}</span>
{elapsedSecs !== null && (
<span aria-hidden="true">
{" · "}
{formatDurationSecs(elapsedSecs)}
</span>
)}
</span>
</Tooltip>
</p>

12
apps/fabro-web/app/lib/query-keys.test.ts
View file

@ -87,7 +87,6 @@ describe("queryKeys", () => {
test("agent activity events invalidate per-stage resources", () => {
for (const event of [
"stage.prompt",
"agent.message",
"agent.tool.started",
"agent.tool.completed",
"command.started",
@ -98,9 +97,16 @@ describe("queryKeys", () => {
queryKeys.runs.stageContextWindow("run-1", "stage-1"),
]);
}
expect(queryKeysForRunEvent("run-1", "agent.message", "stage-1")).toEqual([
queryKeys.runs.state("run-1"),
queryKeys.runs.stageEvents("run-1", "stage-1"),
queryKeys.runs.stageContextWindow("run-1", "stage-1"),
]);
});
test("agent activity events without a node_id invalidate nothing", () => {
expect(queryKeysForRunEvent("run-1", "agent.message")).toEqual([]);
test("agent message without a node_id still invalidates projected state", () => {
expect(queryKeysForRunEvent("run-1", "agent.message")).toEqual([
queryKeys.runs.state("run-1"),
]);
});
});

3
apps/fabro-web/app/lib/run-actions.test.ts
View file

@ -20,6 +20,7 @@ import {
cancelRun,
deleteRuns,
isTerminalCancelledRun,
isTerminalRunStatus,
isCancellationPending,
isCancellationPendingState,
mapError,
@ -424,6 +425,8 @@ describe("run lifecycle actions", () => {
expect(canArchive("failed")).toBe(true);
expect(canArchive("dead")).toBe(true);
expect(canArchive("archived")).toBe(false);
expect(isTerminalRunStatus("succeeded")).toBe(true);
expect(isTerminalRunStatus("running")).toBe(false);
expect(canUnarchive("archived")).toBe(true);
expect(canUnarchive("failed")).toBe(false);

13
apps/fabro-web/app/lib/run-actions.ts
View file

@ -41,7 +41,7 @@ const CANCELABLE_STATUSES = new Set<RunStatus>([
"blocked",
]);
const ARCHIVABLE_STATUSES = new Set<RunStatus>([
const TERMINAL_RUN_STATUSES = new Set<RunStatus>([
"succeeded",
"failed",
"dead",
@ -143,7 +143,7 @@ export function canApprove(run: Run | null | undefined): boolean {
}
export function canArchive(status: string | null | undefined): boolean {
return !!status && ARCHIVABLE_STATUSES.has(status as RunStatus);
return isTerminalRunStatus(status);
}
export function canUnarchive(status: string | null | undefined): boolean {
@ -152,8 +152,13 @@ export function canUnarchive(status: string | null | undefined): boolean {
export function canRetry(run: Pick<Run, "lifecycle"> | null | undefined): boolean {
if (!run || run.lifecycle.archived) return false;
const status = run.lifecycle.status;
return status.kind === "succeeded" || status.kind === "failed" || status.kind === "dead";
return isTerminalRunStatus(run.lifecycle.status.kind);
}
export function isTerminalRunStatus(
status: string | null | undefined,
): boolean {
return !!status && TERMINAL_RUN_STATUSES.has(status as RunStatus);
}
export function canDelete(status: string | null | undefined): boolean {

30
apps/fabro-web/app/lib/run-events.test.tsx
View file

@ -125,6 +125,36 @@ describe("queryKeysForRunEvent", () => {
queryKeys.runs.stageEvents("run-1", "code@1"),
]);
});
test("every inference projection transition invalidates live run state", () => {
for (const event of [
"agent.llm.started",
"agent.llm.first_output",
"agent.llm.retry",
"agent.error",
]) {
expect(queryKeysForRunEvent("run-1", event, "code@1")).toEqual([
queryKeys.runs.state("run-1"),
queryKeys.runs.stageEvents("run-1", "code@1"),
]);
}
expect(
queryKeysForRunEvent("run-1", "agent.message", "code@1"),
).toEqual([
queryKeys.runs.state("run-1"),
queryKeys.runs.stageEvents("run-1", "code@1"),
queryKeys.runs.stageContextWindow("run-1", "code@1"),
]);
expect(queryKeysForRunEvent("run-1", "agent.session.ended")).toEqual([
queryKeys.runs.state("run-1"),
]);
});
test("watchdog timeout refreshes the stage events that settle inference", () => {
expect(
queryKeysForRunEvent("run-1", "watchdog.timeout", "code@1"),
).toEqual([queryKeys.runs.stageEvents("run-1", "code@1")]);
});
});
describe("subscribeToRunEvents", () => {

23
apps/fabro-web/app/lib/run-events.ts
View file

@ -110,6 +110,14 @@ const AGENT_CONTROL_STATE_EVENTS = new Set([
"agent.steering.injected",
"agent.session.deactivated",
]);
const INFERENCE_EVENTS = new Set([
"agent.llm.started",
"agent.llm.first_output",
"agent.llm.retry",
"agent.message",
"agent.error",
"agent.session.ended",
]);
// Todo / task mutation events refresh `getRunState` consumers (so per-stage
// todo projections update live) and the run events list.
const TODO_EVENTS = new Set([
@ -183,6 +191,21 @@ export function queryKeysForRunEvent(
return keys;
}
if (INFERENCE_EVENTS.has(event)) {
const keys: Key[] = [queryKeys.runs.state(runId)];
if (stageId) {
keys.push(queryKeys.runs.stageEvents(runId, stageId));
if (event === "agent.message") {
keys.push(queryKeys.runs.stageContextWindow(runId, stageId));
}
}
return keys;
}
if (event === "watchdog.timeout") {
return stageId ? [queryKeys.runs.stageEvents(runId, stageId)] : [];
}
if (STAGE_ACTIVITY_EVENTS.has(event)) {
return stageId
? [

12
apps/fabro-web/app/routes/run-stages.test.ts
View file

@ -884,6 +884,18 @@ describe("buildChatItems", () => {
});
describe("buildStageActivity pending tools", () => {
test("records watchdog settlement during the activity pass", () => {
const events: EventEnvelope[] = [
envelope(1, {
event: "watchdog.timeout",
stage_id: "plan@1",
node_id: "plan",
}),
];
expect(buildStageActivity(events, "plan@1").watchdogTimedOut).toBe(true);
});
test("returns started-but-not-completed calls for the stage", () => {
const events: EventEnvelope[] = [
envelope(1, {

22
apps/fabro-web/app/routes/run-stages.tsx
View file

@ -73,6 +73,7 @@ import {
useRunStages,
useRunState,
} from "../lib/queries";
import { isTerminalRunStatus } from "../lib/run-actions";
import {
STAGE_ACTIVITY_EVENT_TYPES,
type StageActivityEventType,
@ -270,6 +271,7 @@ export interface PendingToolCall {
interface StageActivity {
turns: TurnType[];
pendingTools: PendingToolCall[];
watchdogTimedOut: boolean;
}
interface PendingCommand {
@ -285,9 +287,14 @@ export function buildStageActivity(
const pendingTools = new Map<string, PendingTool>();
let pendingCommand: PendingCommand | undefined;
let sawAssistantMessage = false;
let watchdogTimedOut = false;
for (const e of events) {
const eventName = e.event;
if (eventName === "watchdog.timeout") {
watchdogTimedOut = true;
continue;
}
if (
activityEventStageId(e) !== stageId ||
!eventName ||
@ -443,6 +450,7 @@ export function buildStageActivity(
return {
turns,
watchdogTimedOut,
pendingTools: Array.from(pendingTools, ([toolCallId, tool]) => ({
toolCallId,
toolName: tool.toolName,
@ -2096,19 +2104,12 @@ function RunStageActivityStage({
// An open bracket on a run that can no longer advance means we never learned
// how the request ended, not that it is still working. The watchdog stays
// the authority on "actually stuck", so its timeout settles the readout too.
const inferenceSettled = useMemo(
() =>
runSettled ||
(stageEventsQuery.data ?? []).some(
(event) => event.event === "watchdog.timeout",
),
[runSettled, stageEventsQuery.data],
);
const activity = useMemo(
() => buildStageActivity(stageEventsQuery.data ?? [], selectedStageId),
[stageEventsQuery.data, selectedStageId],
);
const { turns } = activity;
const inferenceSettled = runSettled || activity.watchdogTimedOut;
const renderer: StageRenderer = selectStageRenderer(selectedStage.handler);
const debugEvents = useMemo<EventEnvelope[]>(() => {
return (stageEventsQuery.data ?? []).filter(
@ -2436,10 +2437,7 @@ export default function RunStages() {
? runStateQuery.data?.stages[selectedStageId]
: undefined;
const runStatusKind = runQuery.data?.lifecycle.status.kind;
const runSettled =
runStatusKind === "succeeded" ||
runStatusKind === "failed" ||
runStatusKind === "dead";
const runSettled = isTerminalRunStatus(runStatusKind);
if (!id || !selectedStage) {
return (

17
docs/internal/events.md
View file

@ -974,9 +974,10 @@ An inference request is about to be dispatched for this round. Emitted once
per round, after the request is built and compaction has run, immediately
before the stream is opened.
`provider` and `model` are the *requested* target. Failover can re-target
mid-stage, so `agent.message` remains authoritative for what actually
answered. No usage or cost fields: neither exists yet at this point.
`requested_model` is the canonical requested target, including an optional
speed tier. Failover can re-target mid-stage, so `agent.message` remains
authoritative for what actually answered. No usage or cost fields: neither
exists yet at this point.
```json
{
@ -985,8 +986,11 @@ answered. No usage or cost fields: neither exists yet at this point.
"node_id": "code", "node_label": "code",
"session_id": "ses_abc",
"properties": {
"provider": "anthropic",
"model": "claude-fable-5",
"requested_model": {
"provider": "anthropic",
"model_id": "claude-fable-5",
"speed": "fast"
},
"visit": 1
}
}
@ -994,8 +998,7 @@ answered. No usage or cost fields: neither exists yet at this point.
| Property | Type | Description |
|----------|------|-------------|
| `provider` | string | Requested provider |
| `model` | string | Requested model |
| `requested_model` | object | Requested provider, model ID, and optional speed tier |
| `visit` | number | Graph visit |
### `agent.llm.first_output`

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

@ -10786,16 +10786,6 @@ components:
- text
- tool_call
LlmRetryPhase:
description: >
Which retry loop produced the `attempt` index on an `agent.llm.retry`
event: `open` for a stream that failed to open, `consume` for one that
broke or ended without a finish event.
type: string
enum:
- open
- consume
SubAgentProjection:
description: Current projected state for one subagent spawned by an agent stage.
type: object

34
lib/apps/fabro-cli/src/commands/run/run_progress/event.rs
View file

@ -136,6 +136,7 @@ pub(super) enum ProgressEvent {
AssistantMessage {
stage_node_id: String,
model: String,
root_session: bool,
},
ToolCallStarted {
stage_node_id: String,
@ -179,6 +180,9 @@ pub(super) enum ProgressEvent {
delay_ms: u64,
error: String,
},
LlmRequestFinished {
stage_node_id: String,
},
SubagentSpawned {
stage_node_id: String,
agent_id: String,
@ -332,6 +336,7 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
EventBody::AgentMessage(props) => Some(ProgressEvent::AssistantMessage {
stage_node_id: node_id,
model: props.model.model_id.to_string(),
root_session: stored.parent_session_id.is_none(),
}),
EventBody::AgentToolStarted(props) => Some(ProgressEvent::ToolCallStarted {
stage_node_id: node_id,
@ -368,15 +373,19 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
preserved_turn_count: props.preserved_turn_count as u64,
tracked_file_count: props.tracked_file_count as u64,
}),
EventBody::AgentLlmStarted(props) => Some(ProgressEvent::LlmRequestStarted {
stage_node_id: node_id,
model: props.model.clone(),
}),
EventBody::AgentLlmFirstOutput(props) => Some(ProgressEvent::LlmFirstOutput {
stage_node_id: node_id,
kind: props.kind,
}),
EventBody::AgentLlmRetry(props) => {
EventBody::AgentLlmStarted(props) if stored.parent_session_id.is_none() => {
Some(ProgressEvent::LlmRequestStarted {
stage_node_id: node_id,
model: props.requested_model.model_id.to_string(),
})
}
EventBody::AgentLlmFirstOutput(props) if stored.parent_session_id.is_none() => {
Some(ProgressEvent::LlmFirstOutput {
stage_node_id: node_id,
kind: props.kind,
})
}
EventBody::AgentLlmRetry(props) if stored.parent_session_id.is_none() => {
#[allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
@ -391,6 +400,13 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
error: display_value(&props.error).unwrap_or_else(|| "unknown error".to_string()),
})
}
EventBody::AgentError(_) | EventBody::AgentRoundInterrupted(_)
if stored.parent_session_id.is_none() =>
{
Some(ProgressEvent::LlmRequestFinished {
stage_node_id: node_id,
})
}
EventBody::AgentSubSpawned(props) => Some(ProgressEvent::SubagentSpawned {
stage_node_id: node_id,
agent_id: props.agent_id.clone(),

128
lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs
View file

@ -256,7 +256,11 @@ impl ProgressUI {
ProgressEvent::AssistantMessage {
stage_node_id,
model,
root_session,
} => {
if root_session {
self.stage.on_llm_request_finished(&stage_node_id);
}
self.stage
.on_assistant_message(renderer, &stage_node_id, &model);
}
@ -345,6 +349,9 @@ impl ProgressUI {
&error,
);
}
ProgressEvent::LlmRequestFinished { stage_node_id } => {
self.stage.on_llm_request_finished(&stage_node_id);
}
ProgressEvent::SubagentSpawned {
stage_node_id,
agent_id,
@ -521,6 +528,17 @@ mod tests {
}
}
fn child_agent_event(stage: &str, event: AgentEvent) -> Event {
Event::Agent {
stage: stage.into(),
visit: 1,
event,
session_id: Some("ses_child".into()),
parent_session_id: Some("ses_root".into()),
tool_call_id: None,
}
}
fn stage_started(node_id: &str, name: &str) -> Event {
Event::StageStarted {
graph_visit: None,
@ -534,9 +552,9 @@ mod tests {
}
}
fn assistant_message(stage: &str, model: &str) -> Event {
agent_event(stage, AgentEvent::AssistantMessage {
text: "done".into(),
fn assistant_event(model: &str, text: &str) -> AgentEvent {
AgentEvent::AssistantMessage {
text: text.into(),
model: ModelRef {
provider: ProviderId::openai(),
model_id: model.into(),
@ -548,6 +566,24 @@ mod tests {
tool_call_count: 0,
context_window: None,
reasoning: None,
}
}
fn assistant_message(stage: &str, model: &str) -> Event {
agent_event(stage, assistant_event(model, "done"))
}
fn child_assistant_message(stage: &str, model: &str) -> Event {
child_agent_event(stage, assistant_event(model, "child done"))
}
fn llm_request_started(stage: &str, model: &str) -> Event {
agent_event(stage, AgentEvent::LlmRequestStarted {
requested_model: ModelRef {
provider: ProviderId::anthropic(),
model_id: model.into(),
speed: None,
},
})
}
@ -699,13 +735,7 @@ mod tests {
emit(&mut ui, stage_started("s1", "Build"));
assert!(ui.stage.active_stages["s1"].inference_bar.is_none());
emit(
&mut ui,
agent_event("s1", AgentEvent::LlmRequestStarted {
provider: "anthropic".into(),
model: "claude-fable-5".into(),
}),
);
emit(&mut ui, llm_request_started("s1", "claude-fable-5"));
let message = ui.stage.active_stages["s1"]
.inference_bar
.as_ref()
@ -736,6 +766,80 @@ mod tests {
assert!(ui.stage.active_stages["s1"].inference_bar.is_none());
}
#[test]
fn inference_retry_resets_the_live_line_before_verbose_output() {
let mut ui = ProgressUI::new(true, false);
emit(&mut ui, stage_started("s1", "Build"));
emit(&mut ui, llm_request_started("s1", "claude-fable-5"));
emit(
&mut ui,
agent_event("s1", AgentEvent::LlmFirstOutput {
kind: fabro_types::LlmOutputKind::Text,
}),
);
emit(
&mut ui,
agent_event("s1", AgentEvent::LlmRetry {
provider: "anthropic".into(),
model: "claude-fable-5".into(),
attempt: 1,
delay_secs: 0.1,
phase: fabro_types::LlmRetryPhase::Consume,
error: fabro_llm::Error::Configuration {
message: "retry".into(),
source: None,
},
}),
);
let message = ui.stage.active_stages["s1"]
.inference_bar
.as_ref()
.expect("retry keeps the bracket open")
.message();
assert!(message.contains("waiting on claude-fable-5"));
}
#[test]
fn inference_interrupt_clears_the_live_line() {
let mut ui = ProgressUI::new(true, false);
emit(&mut ui, stage_started("s1", "Build"));
emit(&mut ui, llm_request_started("s1", "claude-fable-5"));
emit(
&mut ui,
agent_event("s1", AgentEvent::RoundInterrupted { generation: 1 }),
);
assert!(ui.stage.active_stages["s1"].inference_bar.is_none());
}
#[test]
fn child_session_events_do_not_mutate_the_root_inference_line() {
let mut ui = ProgressUI::new(true, false);
emit(&mut ui, stage_started("s1", "Build"));
emit(&mut ui, llm_request_started("s1", "claude-fable-5"));
emit(
&mut ui,
child_agent_event("s1", AgentEvent::LlmFirstOutput {
kind: fabro_types::LlmOutputKind::ToolCall,
}),
);
emit(&mut ui, child_assistant_message("s1", "child-model"));
let message = ui.stage.active_stages["s1"]
.inference_bar
.as_ref()
.expect("child output must not close the root bracket")
.message();
assert!(message.contains("waiting on claude-fable-5"));
emit(&mut ui, assistant_message("s1", "claude-fable-5"));
assert!(ui.stage.active_stages["s1"].inference_bar.is_none());
}
#[test]
fn handle_json_line_ignores_invalid_json() {
let (mut ui, buffer) = capture_ui(false);
@ -800,7 +904,7 @@ mod tests {
model: "gpt-5-mini".into(),
attempt: 2,
delay_secs: 1.5,
phase: Some(fabro_types::LlmRetryPhase::Open),
phase: fabro_types::LlmRetryPhase::Open,
error: fabro_llm::Error::Configuration {
message: "busy".into(),
source: None,
@ -1159,7 +1263,7 @@ mod tests {
model: "gpt-5-mini".into(),
attempt: 2,
delay_secs: 1.5,
phase: Some(fabro_types::LlmRetryPhase::Open),
phase: fabro_types::LlmRetryPhase::Open,
error: fabro_llm::Error::Configuration {
message: "busy".into(),
source: None,

12
lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs
View file

@ -301,8 +301,6 @@ impl StageDisplay {
stage_node_id: &str,
model: &str,
) {
self.on_llm_request_finished(stage_node_id);
if let Some(counts) = self.stage_counts.get_mut(stage_node_id) {
counts.0 += 1;
}
@ -556,6 +554,16 @@ impl StageDisplay {
delay_ms: u64,
error: &str,
) {
if let Some(bar) = self
.active_stages
.get(stage_node_id)
.and_then(|stage| stage.inference_bar.as_ref())
{
bar.set_message(format!(
"\u{27f3} model request: waiting on {model}\u{2026}"
));
}
if !self.verbose {
return;
}

49
lib/components/fabro-agent/src/session.rs
View file

@ -15,7 +15,7 @@ use fabro_llm::{Error as LlmError, retry};
use fabro_mcp::config::{McpServerSettings, McpTransport};
use fabro_mcp::connection_manager::McpConnectionManager;
use fabro_mcp::http_transport;
use fabro_model::{AgentProfileKind, Catalog, ModelRef, Speed, UsdMicros};
use fabro_model::{AgentProfileKind, Catalog, ModelId, ModelRef, Speed, UsdMicros};
use fabro_types::{
AgentToolSummary, LlmOutputKind, LlmRetryPhase, PermissionLevel, Principal, SessionMessage,
SessionRecord, StageContextWindowProjection, SteeringMessage,
@ -95,9 +95,9 @@ fn record_elapsed(start: &mut Option<Instant>, total: &mut Duration) {
/// Classify a stream event as the first unit of provider output, or `None`
/// when it carries no output.
///
/// `StreamStart` is deliberately excluded: `openai_compatible` never emits it,
/// so latching on it would leave whole providers silent. The start/delta/end
/// events below are emitted by every codec that produces that kind of output.
/// `StreamStart` is deliberately excluded because it proves only that the
/// provider responded, not what kind of output followed. The start/delta/end
/// events below identify the first observed content kind.
fn first_output_kind(event: &StreamEvent) -> Option<LlmOutputKind> {
match event {
StreamEvent::ReasoningStart | StreamEvent::ReasoningDelta { .. } => {
@ -1498,8 +1498,11 @@ impl Session {
let local_context_window = built_request.context_window.clone();
let request = built_request.request;
let requested_provider = self.provider_profile.provider_id().to_string();
let requested_model = self.provider_profile.model().to_string();
let requested_model = ModelRef {
provider: self.provider_profile.provider_id(),
model_id: ModelId::new(self.provider_profile.model()),
speed: self.config.speed,
};
// Open the inference bracket for this round. The request is built
// and compaction has run, so this is the last point before the
@ -1507,15 +1510,14 @@ impl Session {
// response.
self.event_emitter
.emit(self.id.clone(), AgentEvent::LlmRequestStarted {
provider: requested_provider.clone(),
model: requested_model.clone(),
requested_model: requested_model.clone(),
});
// Call LLM (streaming) with retry for transient errors
let retry_emitter = self.event_emitter.clone();
let retry_session_id = self.id.clone();
let retry_provider = requested_provider.clone();
let retry_model = requested_model.clone();
let retry_provider = requested_model.provider.to_string();
let retry_model = requested_model.model_id.to_string();
let retry_policy = RetryPolicy {
max_retries: 3,
on_retry: Some(std::sync::Arc::new(move |err, attempt, delay| {
@ -1525,7 +1527,7 @@ impl Session {
attempt: attempt as usize,
delay_secs: delay.as_secs_f64(),
error: err.clone(),
phase: Some(LlmRetryPhase::Open),
phase: LlmRetryPhase::Open,
});
})),
..Default::default()
@ -1684,12 +1686,12 @@ impl Session {
// callback only ever runs for stream-open failures.
self.event_emitter
.emit(self.id.clone(), AgentEvent::LlmRetry {
provider: requested_provider.clone(),
model: requested_model.clone(),
provider: requested_model.provider.to_string(),
model: requested_model.model_id.to_string(),
attempt: stream_attempt,
delay_secs: delay.as_secs_f64(),
error: err.clone(),
phase: Some(LlmRetryPhase::Consume),
error: err,
phase: LlmRetryPhase::Consume,
});
let delay_outcome = tokio::select! {
@ -1762,15 +1764,15 @@ impl Session {
// with nothing on the durable stream to show for it.
self.event_emitter
.emit(self.id.clone(), AgentEvent::LlmRetry {
provider: requested_provider.clone(),
model: requested_model.clone(),
provider: requested_model.provider.to_string(),
model: requested_model.model_id.to_string(),
attempt: stream_attempt,
delay_secs: 0.0,
error: LlmError::Stream {
message: "Stream ended without a finish event".to_string(),
source: None,
},
phase: Some(LlmRetryPhase::Consume),
phase: LlmRetryPhase::Consume,
});
let cancel_token_for_select = self.cancel_token.clone();
let retry_outcome: Option<Result<StreamEventStream, Error>> = tokio::select! {
@ -4276,8 +4278,11 @@ mod tests {
let mut observed = Vec::new();
while let Ok(event) = rx.try_recv() {
match event.event {
AgentEvent::LlmRequestStarted { provider, model } => {
observed.push(("started".to_string(), format!("{provider}/{model}")));
AgentEvent::LlmRequestStarted { requested_model } => {
observed.push((
"started".to_string(),
format!("{}/{}", requested_model.provider, requested_model.model_id),
));
}
AgentEvent::LlmFirstOutput { kind } => {
observed.push(("first_output".to_string(), kind.to_string()));
@ -4430,7 +4435,7 @@ mod tests {
assert_eq!(assistant_messages, vec!["Recovered".to_string()]);
// The finish-less restart is the one mid-turn path with no error to
// report; without this event it would be invisible downstream.
assert_eq!(consume_retries, vec![(0, Some(LlmRetryPhase::Consume))]);
assert_eq!(consume_retries, vec![(0, LlmRetryPhase::Consume)]);
}
#[tokio::test]
@ -4461,7 +4466,7 @@ mod tests {
observed.push(format!("replace:{text}:{reasoning:?}"));
}
AgentEvent::LlmRetry { phase, .. } => {
observed.push(format!("retry:{}", phase.expect("phase is always set")));
observed.push(format!("retry:{phase}"));
}
AgentEvent::AssistantMessage { text, .. } => {
observed.push(format!("message:{text}"));

20
lib/components/fabro-agent/src/types.rs
View file

@ -242,8 +242,7 @@ pub enum AgentEvent {
/// failover can re-target, so `AssistantMessage` stays authoritative for
/// what actually answered.
LlmRequestStarted {
provider: String,
model: String,
requested_model: ModelRef,
},
/// The provider produced its first output for the current attempt.
/// Edge-triggered: emitted once per stream attempt, re-armed when a
@ -336,8 +335,7 @@ pub enum AgentEvent {
attempt: usize,
delay_secs: f64,
error: LlmError,
#[serde(default, skip_serializing_if = "Option::is_none")]
phase: Option<LlmRetryPhase>,
phase: LlmRetryPhase,
},
SubAgentSpawned {
agent_id: String,
@ -427,8 +425,14 @@ impl AgentEvent {
Self::UserInput { text } => {
debug!(session_id, text_len = text.len(), "User input received");
}
Self::LlmRequestStarted { provider, model } => {
debug!(session_id, provider, model, "LLM request started");
Self::LlmRequestStarted { requested_model } => {
debug!(
session_id,
provider = %requested_model.provider,
model = %requested_model.model_id,
speed = requested_model.speed.map_or("", <&'static str>::from),
"LLM request started"
);
}
Self::LlmFirstOutput { kind } => {
debug!(session_id, kind = %kind, "LLM produced first output");
@ -537,7 +541,7 @@ impl AgentEvent {
model,
attempt,
delay_secs,
phase = phase.map_or("", <&'static str>::from),
phase = %phase,
error = %error,
"LLM request failed, retrying"
);
@ -987,7 +991,7 @@ mod tests {
model: "gpt-4".into(),
attempt: 1,
delay_secs: 2.0,
phase: Some(LlmRetryPhase::Open),
phase: LlmRetryPhase::Open,
error: LlmError::Provider {
kind: ProviderErrorKind::RateLimit,
detail: Box::new(ProviderErrorDetail {

17
lib/components/fabro-llm/src/providers/bedrock/mod.rs
View file

@ -312,7 +312,7 @@ struct EventStreamLoop {
response: fabro_http::Response,
frames: FrameDecoder,
decoder: Box<dyn StreamDecoder>,
pending: VecDeque<StreamEvent>,
pending: VecDeque<Result<StreamEvent, Error>>,
done: bool,
/// `finish()` already drained.
finished: bool,
@ -343,7 +343,7 @@ fn decode_eventstream(
move |mut state| async move {
loop {
if let Some(event) = state.pending.pop_front() {
return Some((Ok(event), state));
return Some((event, state));
}
if state.done {
@ -351,7 +351,9 @@ fn decode_eventstream(
return None;
}
state.finished = true;
state.pending.extend(state.decoder.finish());
state
.pending
.extend(state.decoder.finish().into_iter().map(Ok));
if state.pending.is_empty() {
return None;
}
@ -378,11 +380,14 @@ fn decode_eventstream(
// type the provider happens to open with.
if !state.stream_started {
state.stream_started = true;
state.pending.push_back(StreamEvent::StreamStart);
state.pending.push_back(Ok(StreamEvent::StreamStart));
}
match state.decoder.on_event(raw) {
Ok(events) => state.pending.extend(events),
Err(e) => return Some((Err(e), state)),
Ok(events) => state.pending.extend(events.into_iter().map(Ok)),
Err(error) => {
state.pending.push_back(Err(error));
break;
}
}
}
}

16
lib/components/fabro-llm/src/transport.rs
View file

@ -247,8 +247,8 @@ pub(crate) async fn stream_via_http(
struct StreamLoop {
decoder: Box<dyn StreamDecoder>,
line_reader: LineReader,
/// Events decoded but not yet yielded.
pending: VecDeque<StreamEvent>,
/// Events or decoder errors not yet yielded.
pending: VecDeque<Result<StreamEvent, Error>>,
/// Byte stream exhausted.
done: bool,
/// `finish()` already drained.
@ -278,7 +278,7 @@ fn decode_sse_stream(
move |mut state| async move {
loop {
if let Some(event) = state.pending.pop_front() {
return Some((Ok(event), state));
return Some((event, state));
}
if state.done {
@ -286,7 +286,9 @@ fn decode_sse_stream(
return None;
}
state.finished_emitted = true;
state.pending.extend(state.decoder.finish());
state
.pending
.extend(state.decoder.finish().into_iter().map(Ok));
if state.pending.is_empty() {
return None;
}
@ -305,11 +307,11 @@ fn decode_sse_stream(
// particular opening frame.
if !state.stream_started {
state.stream_started = true;
state.pending.push_back(StreamEvent::StreamStart);
state.pending.push_back(Ok(StreamEvent::StreamStart));
}
match state.decoder.on_event(RawEvent { event, data: &data }) {
Ok(events) => state.pending.extend(events),
Err(e) => return Some((Err(e), state)),
Ok(events) => state.pending.extend(events.into_iter().map(Ok)),
Err(error) => state.pending.push_back(Err(error)),
}
}
Ok(None) => state.done = true,

12
lib/components/fabro-llm/tests/it/support.rs
View file

@ -136,6 +136,18 @@ pub(crate) async fn collect_stream_events(
events
}
/// Pin the transport-level liveness contract independently of snapshots.
pub(crate) fn assert_stream_starts(events: &[serde_json::Value]) {
assert_eq!(
events
.first()
.and_then(|event| event.get("type"))
.and_then(serde_json::Value::as_str),
Some("stream_start"),
"the first decoded provider frame must open with stream_start"
);
}
/// Builds a catalog from inline TOML (same `LlmCatalogSettings` schema as the
/// shipped catalog files).
pub(crate) fn catalog_from_toml(source: &str) -> Arc<Catalog> {

11
lib/components/fabro-llm/tests/it/wire/anthropic.rs
View file

@ -648,19 +648,10 @@ async fn stream_text_happy_path_request() {
#[tokio::test]
async fn stream_text_happy_path_events() {
let (_, events) = stream_text_happy_path_capture().await;
support::assert_stream_starts(&events);
fabro_test::fabro_json_snapshot!(events);
}
/// Every dialect's stream opens with `StreamStart`, emitted by the driving
/// loop rather than the decoder. Asserted outside the snapshot because a
/// snapshot can be re-accepted silently, and this is the one event a liveness
/// consumer needs to be able to rely on from every provider.
#[tokio::test]
async fn stream_opens_with_stream_start() {
let (_, events) = stream_text_happy_path_capture().await;
assert_eq!(events[0]["type"], "stream_start");
}
#[tokio::test]
async fn stream_tool_call_deltas() {
let sse = support::sse_transcript(&[

11
lib/components/fabro-llm/tests/it/wire/gemini.rs
View file

@ -448,19 +448,10 @@ async fn stream_text_happy_path_request() {
#[tokio::test]
async fn stream_text_happy_path_events() {
let (_, events) = stream_text_happy_path_capture().await;
support::assert_stream_starts(&events);
fabro_test::fabro_json_snapshot!(events);
}
/// Every dialect's stream opens with `StreamStart`, emitted by the driving
/// loop rather than the decoder. Asserted outside the snapshot because a
/// snapshot can be re-accepted silently, and this is the one event a liveness
/// consumer needs to be able to rely on from every provider.
#[tokio::test]
async fn stream_opens_with_stream_start() {
let (_, events) = stream_text_happy_path_capture().await;
assert_eq!(events[0]["type"], "stream_start");
}
#[tokio::test]
async fn stream_function_call() {
let sse = support::sse_data_transcript(&[

11
lib/components/fabro-llm/tests/it/wire/openai_compatible.rs
View file

@ -823,19 +823,10 @@ async fn stream_text_happy_path_request() {
#[tokio::test]
async fn stream_text_happy_path_events() {
let (_, events) = stream_text_happy_path_capture().await;
support::assert_stream_starts(&events);
fabro_test::fabro_json_snapshot!(events);
}
/// Every dialect's stream opens with `StreamStart`, emitted by the driving
/// loop rather than the decoder. Asserted outside the snapshot because a
/// snapshot can be re-accepted silently, and this is the one event a liveness
/// consumer needs to be able to rely on from every provider.
#[tokio::test]
async fn stream_opens_with_stream_start() {
let (_, events) = stream_text_happy_path_capture().await;
assert_eq!(events[0]["type"], "stream_start");
}
/// OpenRouter streams report `cost` in the usage chunk; the Finish response
/// carries it as authoritative, with cached tokens in their own bucket.
#[tokio::test]

22
lib/components/fabro-llm/tests/it/wire/openai_responses.rs
View file

@ -542,17 +542,25 @@ async fn stream_text_happy_path_request() {
#[tokio::test]
async fn stream_text_happy_path_events() {
let (_, events) = stream_text_happy_path_capture().await;
support::assert_stream_starts(&events);
fabro_test::fabro_json_snapshot!(events);
}
/// Every dialect's stream opens with `StreamStart`, emitted by the driving
/// loop rather than the decoder. Asserted outside the snapshot because a
/// snapshot can be re-accepted silently, and this is the one event a liveness
/// consumer needs to be able to rely on from every provider.
#[tokio::test]
async fn stream_opens_with_stream_start() {
let (_, events) = stream_text_happy_path_capture().await;
assert_eq!(events[0]["type"], "stream_start");
async fn stream_first_frame_error_still_opens_with_stream_start() {
let sse = support::sse_data_transcript(&[
r#"{"type":"response.failed","response":{"id":"resp_stream","error":{"code":"server_error","message":"boom"}}}"#,
]);
let (_capture, events) = stream_capture(adapter(), &base_request(MODEL), &sse).await;
support::assert_stream_starts(&events);
assert!(
events
.get(1)
.and_then(|event| event.get("stream_item_error"))
.is_some(),
"the decoder error should follow stream_start: {events:?}"
);
}
#[tokio::test]

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

@ -11,15 +11,14 @@ use fabro_types::settings::run::{EnvironmentProvider, RunEnvironmentSettings};
use fabro_types::{
ActivatedSkill, AgentControlState, AskFabro, BilledModelUsage, BilledTokenCounts, Checkpoint,
CheckpointRecord, CommandTermination, Conclusion, EventBody, FailureCategory, FailureSignature,
InterviewQuestionRecord, McpServerProjection, McpServerStatus, ModelId, ModelRef, Outcome,
PendingInterviewRecord, PendingReason, ProviderId, PullRequestLink, RepositoryRef, Run,
RunApproval, RunApprovalState, RunBillingSummary, RunControlAction, RunDiff, RunEvent, RunId,
RunLifecycle, RunLinks, RunModel, RunOrigin, RunProjection, RunSandbox, RunSandboxFailure,
RunSandboxInstance, RunSandboxPlan, RunSandboxRuntime, RunSize, RunSpec, RunStatus,
RunTimestamps, SandboxProviderKind, StageCompletion, StageHandler, StageId,
StageInferenceProjection, StageModelUsage, StageOutcome, StageProjection, StageState,
StartRecord, SubAgentProjection, SubAgentStatus, TodoListKind, TodoListProjection,
TodoProjection, WorkflowRef, first_event_seq,
InterviewQuestionRecord, McpServerProjection, McpServerStatus, Outcome, PendingInterviewRecord,
PendingReason, PullRequestLink, RepositoryRef, Run, RunApproval, RunApprovalState,
RunBillingSummary, RunControlAction, RunDiff, RunEvent, RunId, RunLifecycle, RunLinks,
RunModel, RunOrigin, RunProjection, RunSandbox, RunSandboxFailure, RunSandboxInstance,
RunSandboxPlan, RunSandboxRuntime, RunSize, RunSpec, RunStatus, RunTimestamps,
SandboxProviderKind, StageCompletion, StageHandler, StageId, StageInferenceProjection,
StageModelUsage, StageOutcome, StageProjection, StageState, StartRecord, SubAgentProjection,
SubAgentStatus, TodoListKind, TodoListProjection, TodoProjection, WorkflowRef, first_event_seq,
};
use fabro_util::error::render_compact_with_causes;
@ -1024,11 +1023,7 @@ fn open_inference_bracket(
stage.inference = Some(StageInferenceProjection {
session_id,
started_at: ts,
requested_model: ModelRef {
provider: ProviderId::new(props.provider.clone()),
model_id: ModelId::new(props.model.clone()),
speed: None,
},
requested_model: props.requested_model.clone(),
first_output_at: None,
first_output_kind: None,
retries: 0,
@ -1099,7 +1094,7 @@ fn close_inference_brackets_for_session(state: &mut RunProjection, stored: &RunE
let Some(session_id) = stored.session_id.as_deref() else {
return;
};
for (_, stage) in state.iter_stages_mut() {
for (_, stage) in state.iter_stages_unordered_mut() {
let opened_here = stage
.inference
.as_ref()
@ -6291,7 +6286,9 @@ mod tests {
use fabro_types::run_event::{
AgentErrorProps, AgentLlmFirstOutputProps, AgentLlmRetryProps, AgentLlmStartedProps,
};
use fabro_types::{LlmOutputKind, LlmRetryPhase, StageInferenceProjection};
use fabro_types::{
LlmOutputKind, LlmRetryPhase, ModelRef, Speed, StageInferenceProjection,
};
use super::*;
@ -6330,9 +6327,12 @@ mod tests {
fn started() -> EventBody {
EventBody::AgentLlmStarted(AgentLlmStartedProps {
provider: "anthropic".to_string(),
model: "claude-fable-5".to_string(),
visit: 1,
requested_model: ModelRef {
provider: "anthropic".parse().unwrap(),
model_id: "claude-fable-5".into(),
speed: Some(Speed::Fast),
},
visit: 1,
})
}
@ -6368,6 +6368,7 @@ mod tests {
inference.requested_model.model_id.as_str(),
"claude-fable-5"
);
assert_eq!(inference.requested_model.speed, Some(Speed::Fast));
assert_eq!(inference.first_output_at, None);
assert_eq!(inference.first_output_kind, None);
assert_eq!(inference.retries, 0);

57
lib/components/fabro-store/src/slate/projection_cache.rs
View file

@ -72,12 +72,36 @@ impl RunProjectionCacheState {
let Some(parent_id) = entry.summary.parent_id else {
return;
};
let Some(children) = self.children_by_parent.get_mut(&parent_id) else {
self.remove_parent_link(&parent_id, &entry.run_id);
}
fn remove_parent_link(&mut self, parent_id: &RunId, run_id: &RunId) {
let Some(children) = self.children_by_parent.get_mut(parent_id) else {
return;
};
children.remove(&entry.run_id);
children.remove(run_id);
if children.is_empty() {
self.children_by_parent.remove(&parent_id);
self.children_by_parent.remove(parent_id);
}
}
fn update_parent_index(
&mut self,
run_id: RunId,
previous_parent_id: Option<RunId>,
parent_id: Option<RunId>,
) {
if previous_parent_id == parent_id {
return;
}
if let Some(previous_parent_id) = previous_parent_id {
self.remove_parent_link(&previous_parent_id, &run_id);
}
if let Some(parent_id) = parent_id {
self.children_by_parent
.entry(parent_id)
.or_default()
.insert(run_id);
}
}
@ -204,7 +228,7 @@ impl RunProjectionCache {
event: &EventEnvelope,
) -> Result<CachedRunProjection> {
let mut state = self.state.lock().await;
let Some(entry) = state.entries.get(run_id).cloned() else {
let Some(entry) = state.entries.get(run_id) else {
if event.seq == 1 {
let projection = RunProjection::apply_events(std::slice::from_ref(event))?;
let entry = CachedRunProjection::from_projection(*run_id, projection, event.seq);
@ -217,20 +241,29 @@ impl RunProjectionCache {
)));
};
if event.seq <= entry.last_seq {
return Ok(entry);
let last_seq = entry.last_seq;
if event.seq <= last_seq {
return Ok(entry.clone());
}
if event.seq != entry.last_seq.saturating_add(1) {
if event.seq != last_seq.saturating_add(1) {
return Err(Error::Other(format!(
"projection cache sequence gap for run {run_id}: last_seq={}, event_seq={}",
entry.last_seq, event.seq
last_seq, event.seq
)));
}
let mut projection = (*entry.projection).clone();
projection.apply_event(event)?;
let entry = CachedRunProjection::from_projection(*run_id, projection, event.seq);
state.insert(entry.clone());
let (previous_parent_id, parent_id, entry) = {
let entry = state
.entries
.get_mut(run_id)
.expect("entry was read from the same locked map");
let previous_parent_id = entry.summary.parent_id;
Arc::make_mut(&mut entry.projection).apply_event(event)?;
entry.summary = build_summary(&entry.projection, run_id);
entry.last_seq = event.seq;
(previous_parent_id, entry.summary.parent_id, entry.clone())
};
state.update_parent_index(*run_id, previous_parent_id, parent_id);
Ok(entry)
}

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

@ -704,11 +704,10 @@ fn event_body_from_event(event: &Event) -> EventBody {
tracked_file_count: *tracked_file_count,
visit: *visit,
}),
AgentEvent::LlmRequestStarted { provider, model } => {
AgentEvent::LlmRequestStarted { requested_model } => {
EventBody::AgentLlmStarted(fabro_types::AgentLlmStartedProps {
provider: provider.clone(),
model: model.clone(),
visit: *visit,
requested_model: requested_model.clone(),
visit: *visit,
})
}
AgentEvent::LlmFirstOutput { kind } => {
@ -730,7 +729,7 @@ fn event_body_from_event(event: &Event) -> EventBody {
attempt: *attempt,
delay_secs: *delay_secs,
error: serde_json::to_value(error).expect("LLM SDK error derives Serialize with no custom logic that can fail"),
phase: *phase,
phase: Some(*phase),
visit: *visit,
}),
AgentEvent::SubAgentSpawned {

1
lib/foundation/fabro-api/build.rs
View file

@ -362,7 +362,6 @@ fn main() {
&[],
),
("LlmOutputKind", "fabro_types::LlmOutputKind", &[]),
("LlmRetryPhase", "fabro_types::LlmRetryPhase", &[]),
("PermissionLevel", "fabro_types::PermissionLevel", &[]),
(
"AgentSessionActivatedProps",

2
lib/foundation/fabro-api/src/lib.rs
View file

@ -47,7 +47,7 @@ pub mod types {
DirtyStatus, EventEnvelope, ExecOutputTail, FailureCategory, FailureDetail,
FailureSignature, GitContext, IdpIdentity, IntegrationConnectionKind,
IntegrationConnectionState, IntegrationConnectionStatus, IntegrationProvider,
IntegrationStatus, InterviewOption, InterviewQuestionRecord, LlmOutputKind, LlmRetryPhase,
IntegrationStatus, InterviewOption, InterviewQuestionRecord, LlmOutputKind,
McpServerDraft as CreateMcpServerRequest, McpServerProjection,
McpServerReplace as ReplaceMcpServerRequest, McpServerStatus, McpServerView as McpServer,
McpTransportView, Message, PairId, PairMessageId, PairMessageRecord, PairMessageRequest,

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

@ -7,10 +7,9 @@ use fabro_api::types::{
AgentSkillSummary as ApiAgentSkillSummary, AgentToolCategory as ApiAgentToolCategory,
AgentToolSource as ApiAgentToolSource, AgentToolSummary as ApiAgentToolSummary,
AgentToolsAvailableProps as ApiAgentToolsAvailableProps, LlmOutputKind as ApiLlmOutputKind,
LlmRetryPhase as ApiLlmRetryPhase, McpServerProjection as ApiMcpServerProjection,
McpServerStatus as ApiMcpServerStatus, ParallelBranchResult as ApiParallelBranchResult,
PermissionLevel as ApiPermissionLevel, SkillsProjection as ApiSkillsProjection,
StageContextWindow as ApiStageContextWindow,
McpServerProjection as ApiMcpServerProjection, McpServerStatus as ApiMcpServerStatus,
ParallelBranchResult as ApiParallelBranchResult, PermissionLevel as ApiPermissionLevel,
SkillsProjection as ApiSkillsProjection, StageContextWindow as ApiStageContextWindow,
StageContextWindowBreakdownItem as ApiStageContextWindowBreakdownItem,
StageContextWindowCategory as ApiStageContextWindowCategory,
StageContextWindowCountMethod as ApiStageContextWindowCountMethod,
@ -22,15 +21,16 @@ use fabro_api::types::{
SubAgentProjection as ApiSubAgentProjection, SubAgentStatus as ApiSubAgentStatus,
TodoListProjection as ApiTodoListProjection,
};
use fabro_model::{ModelId, ModelRef, ProviderId, Speed};
use fabro_types::{
ActivatedSkill, AgentControlState, AgentMcpToolSummary, AgentSkillActivationSource,
AgentSkillSummary, AgentToolCategory, AgentToolSource, AgentToolSummary,
AgentToolsAvailableProps, LlmOutputKind, LlmRetryPhase, McpServerProjection, McpServerStatus,
ModelId, ModelRef, ParallelBranchResult, PermissionLevel, ProviderId, SkillsProjection,
StageContextWindow, StageContextWindowBreakdownItem, StageContextWindowCategory,
StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness,
StageContextWindowUnavailableReason, StageContextWindowWarning, StageInferenceProjection,
StageProjection, SubAgentProjection, SubAgentStatus, TodoListKind, TodoListProjection,
AgentToolsAvailableProps, LlmOutputKind, McpServerProjection, McpServerStatus,
ParallelBranchResult, PermissionLevel, SkillsProjection, StageContextWindow,
StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowUnavailableReason,
StageContextWindowWarning, StageInferenceProjection, StageProjection, SubAgentProjection,
SubAgentStatus, TodoListKind, TodoListProjection,
};
use serde_json::json;
@ -69,7 +69,6 @@ fn stage_projection_reuses_nested_agent_state_types() {
assert_same_type::<ApiStageContextWindowWarning, StageContextWindowWarning>();
assert_same_type::<ApiStageInferenceProjection, StageInferenceProjection>();
assert_same_type::<ApiLlmOutputKind, LlmOutputKind>();
assert_same_type::<ApiLlmRetryPhase, LlmRetryPhase>();
}
#[test]
@ -80,7 +79,7 @@ fn stage_inference_projection_matches_openapi_json_shape() {
requested_model: ModelRef {
provider: ProviderId::new("anthropic"),
model_id: ModelId::new("claude-fable-5"),
speed: None,
speed: Some(Speed::Fast),
},
first_output_at: Some("2026-04-29T12:34:07Z".parse().unwrap()),
first_output_kind: Some(LlmOutputKind::Reasoning),
@ -94,7 +93,8 @@ fn stage_inference_projection_matches_openapi_json_shape() {
"started_at": "2026-04-29T12:34:00Z",
"requested_model": {
"provider": "anthropic",
"model_id": "claude-fable-5"
"model_id": "claude-fable-5",
"speed": "fast"
},
"first_output_at": "2026-04-29T12:34:07Z",
"first_output_kind": "reasoning",
@ -117,16 +117,6 @@ fn llm_enums_match_openapi_json_shape() {
let api_kind: ApiLlmOutputKind = serde_json::from_value(value).unwrap();
assert_eq!(api_kind, kind);
}
for (phase, wire) in [
(LlmRetryPhase::Open, "open"),
(LlmRetryPhase::Consume, "consume"),
] {
let value = serde_json::to_value(phase).unwrap();
assert_eq!(value, json!(wire));
let api_phase: ApiLlmRetryPhase = serde_json::from_value(value).unwrap();
assert_eq!(api_phase, phase);
}
}
/// A stage projection written before `inference` existed must still

5
lib/foundation/fabro-types/src/billing.rs
View file

@ -1,7 +1,6 @@
pub use fabro_model::{
AnthropicBillingFacts, AnthropicModelPricing, BilledModelUsage, BilledTokenCounts,
GeminiBillingFacts, GeminiModelPricing, GeminiStoragePricing, GeminiStorageSegment,
ModelBillingFacts, ModelBillingInput, ModelId, ModelPricing, ModelPricingPolicy, ModelRef,
ModelUsage, OpenAiBillingFacts, OpenAiModelPricing, PricePerMTok, ProviderId, Speed,
TokenCounts, UsdMicros,
ModelBillingFacts, ModelBillingInput, ModelPricing, ModelPricingPolicy, ModelRef, ModelUsage,
OpenAiBillingFacts, OpenAiModelPricing, PricePerMTok, Speed, TokenCounts, UsdMicros,
};

5
lib/foundation/fabro-types/src/lib.rs
View file

@ -59,9 +59,8 @@ pub use auth::{IdpIdentity, IdpIdentityError};
pub use billing::{
AnthropicBillingFacts, AnthropicModelPricing, BilledModelUsage, BilledTokenCounts,
GeminiBillingFacts, GeminiModelPricing, GeminiStoragePricing, GeminiStorageSegment,
ModelBillingFacts, ModelBillingInput, ModelId, ModelPricing, ModelPricingPolicy, ModelRef,
ModelUsage, OpenAiBillingFacts, OpenAiModelPricing, PricePerMTok, ProviderId, Speed,
TokenCounts, UsdMicros,
ModelBillingFacts, ModelBillingInput, ModelPricing, ModelPricingPolicy, ModelRef, ModelUsage,
OpenAiBillingFacts, OpenAiModelPricing, PricePerMTok, Speed, TokenCounts, UsdMicros,
};
pub use blob_ref::{format_blob_ref, parse_blob_ref, parse_managed_blob_file_ref};
pub use checkpoint::Checkpoint;

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

@ -340,14 +340,13 @@ pub enum LlmOutputKind {
/// An inference request is about to be dispatched for this round.
///
/// `provider` and `model` are the *requested* target from the session's
/// provider profile. Failover can re-target mid-stage, so `agent.message`
/// remains authoritative for what actually answered.
/// `requested_model` is the requested target from the session's provider
/// profile. Failover can re-target mid-stage, so `agent.message` remains
/// authoritative for what actually answered.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AgentLlmStartedProps {
pub provider: String,
pub model: String,
pub visit: u32,
pub requested_model: ModelRef,
pub visit: u32,
}
/// The provider produced its first output for the current attempt.

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

@ -632,6 +632,16 @@ impl RunProjection {
self.stages.iter()
}
/// Mutable counterpart of [`Self::iter_stages_unordered`].
///
/// Use this only for order-independent mutation. Presentation and
/// serialization callers should use [`Self::iter_stages_mut`] instead.
pub fn iter_stages_unordered_mut(
&mut self,
) -> impl Iterator<Item = (&StageId, &mut StageProjection)> {
self.stages.iter_mut()
}
/// Iterate stages in `first_event_seq` order (the chronological order in
/// which each stage's first lifecycle event was recorded). Internal
/// storage is a `HashMap`, so presentation callers sort through this

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

@ -198,7 +198,6 @@ models/interview-provider-settings.ts
models/interview-question-record.ts
models/link-run-pull-request-request.ts
models/llm-output-kind.ts
models/llm-retry-phase.ts
models/log-destination.ts
models/manifest-args.ts
models/manifest-config.ts

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

@ -168,7 +168,6 @@ export * from './interview-provider-settings';
export * from './interview-question-record';
export * from './link-run-pull-request-request';
export * from './llm-output-kind';
export * from './llm-retry-phase';
export * from './log-destination';
export * from './manifest-args';
export * from './manifest-config';

26
lib/packages/fabro-api-client/src/models/llm-retry-phase.ts generated
View file

@ -1,26 +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.
*/
/**
* Which retry loop produced the `attempt` index on an `agent.llm.retry` event: `open` for a stream that failed to open, `consume` for one that broke or ended without a finish event.
*/
export const LlmRetryPhase = {
OPEN: 'open',
CONSUME: 'consume'
} as const;
export type LlmRetryPhase = typeof LlmRetryPhase[keyof typeof LlmRetryPhase];