diff --git a/apps/fabro-web/app/components/stage-inference-indicator.test.tsx b/apps/fabro-web/app/components/stage-inference-indicator.test.tsx index b2764af67..45da56ae4 100644 --- a/apps/fabro-web/app/components/stage-inference-indicator.test.tsx +++ b/apps/fabro-web/app/components/stage-inference-indicator.test.tsx @@ -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( , ); }); - 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("%"); }); diff --git a/apps/fabro-web/app/components/stage-inference-indicator.tsx b/apps/fabro-web/app/components/stage-inference-indicator.tsx index d18cdcee1..a329f3fa0 100644 --- a/apps/fabro-web/app/components/stage-inference-indicator.tsx +++ b/apps/fabro-web/app/components/stage-inference-indicator.tsx @@ -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 ( -

+

@@ -72,7 +73,13 @@ export function StageInferenceIndicator({ className="size-1.5 animate-pulse rounded-full bg-teal-500" aria-hidden="true" /> - {parts.join(" · ")} + {statusParts.join(" · ")} + {elapsedSecs !== null && ( + + )}

diff --git a/apps/fabro-web/app/lib/query-keys.test.ts b/apps/fabro-web/app/lib/query-keys.test.ts index 6faa4e158..ac441b998 100644 --- a/apps/fabro-web/app/lib/query-keys.test.ts +++ b/apps/fabro-web/app/lib/query-keys.test.ts @@ -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"), + ]); }); }); diff --git a/apps/fabro-web/app/lib/run-actions.test.ts b/apps/fabro-web/app/lib/run-actions.test.ts index ab67c0e90..0547f9398 100644 --- a/apps/fabro-web/app/lib/run-actions.test.ts +++ b/apps/fabro-web/app/lib/run-actions.test.ts @@ -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); diff --git a/apps/fabro-web/app/lib/run-actions.ts b/apps/fabro-web/app/lib/run-actions.ts index 46fc1efdd..a45034088 100644 --- a/apps/fabro-web/app/lib/run-actions.ts +++ b/apps/fabro-web/app/lib/run-actions.ts @@ -41,7 +41,7 @@ const CANCELABLE_STATUSES = new Set([ "blocked", ]); -const ARCHIVABLE_STATUSES = new Set([ +const TERMINAL_RUN_STATUSES = new Set([ "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 | 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 { diff --git a/apps/fabro-web/app/lib/run-events.test.tsx b/apps/fabro-web/app/lib/run-events.test.tsx index 55af2aefb..11544b14d 100644 --- a/apps/fabro-web/app/lib/run-events.test.tsx +++ b/apps/fabro-web/app/lib/run-events.test.tsx @@ -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", () => { diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index 32f8edfb5..d0d4f4b4d 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/apps/fabro-web/app/lib/run-events.ts @@ -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 ? [ diff --git a/apps/fabro-web/app/routes/run-stages.test.ts b/apps/fabro-web/app/routes/run-stages.test.ts index 51684a895..a7024b3ef 100644 --- a/apps/fabro-web/app/routes/run-stages.test.ts +++ b/apps/fabro-web/app/routes/run-stages.test.ts @@ -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, { diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index b9257dc27..d45416939 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/apps/fabro-web/app/routes/run-stages.tsx @@ -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(); 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(() => { 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 ( diff --git a/docs/internal/events.md b/docs/internal/events.md index 480b976b8..18045cf36 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -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` diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 7e00d48aa..d5f66471d 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -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 diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs index bc070ce19..cb063d7ce 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs @@ -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 { 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 { 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 { 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(), diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index 2f923f637..c28d72b20 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -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, diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs index 8283ff674..b54d8d6a7 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs @@ -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; } diff --git a/lib/components/fabro-agent/src/session.rs b/lib/components/fabro-agent/src/session.rs index fff7fc05b..f022504a1 100644 --- a/lib/components/fabro-agent/src/session.rs +++ b/lib/components/fabro-agent/src/session.rs @@ -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, 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 { 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> = 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}")); diff --git a/lib/components/fabro-agent/src/types.rs b/lib/components/fabro-agent/src/types.rs index 78c095cc8..ad01385b8 100644 --- a/lib/components/fabro-agent/src/types.rs +++ b/lib/components/fabro-agent/src/types.rs @@ -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, + 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 { diff --git a/lib/components/fabro-llm/src/providers/bedrock/mod.rs b/lib/components/fabro-llm/src/providers/bedrock/mod.rs index 55eda72db..6f10dadb2 100644 --- a/lib/components/fabro-llm/src/providers/bedrock/mod.rs +++ b/lib/components/fabro-llm/src/providers/bedrock/mod.rs @@ -312,7 +312,7 @@ struct EventStreamLoop { response: fabro_http::Response, frames: FrameDecoder, decoder: Box, - pending: VecDeque, + pending: VecDeque>, 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; + } } } } diff --git a/lib/components/fabro-llm/src/transport.rs b/lib/components/fabro-llm/src/transport.rs index ed03d2ef4..8fc62a32f 100644 --- a/lib/components/fabro-llm/src/transport.rs +++ b/lib/components/fabro-llm/src/transport.rs @@ -247,8 +247,8 @@ pub(crate) async fn stream_via_http( struct StreamLoop { decoder: Box, line_reader: LineReader, - /// Events decoded but not yet yielded. - pending: VecDeque, + /// Events or decoder errors not yet yielded. + pending: VecDeque>, /// 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, diff --git a/lib/components/fabro-llm/tests/it/support.rs b/lib/components/fabro-llm/tests/it/support.rs index c2d30c58c..3c1e3a62c 100644 --- a/lib/components/fabro-llm/tests/it/support.rs +++ b/lib/components/fabro-llm/tests/it/support.rs @@ -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 { diff --git a/lib/components/fabro-llm/tests/it/wire/anthropic.rs b/lib/components/fabro-llm/tests/it/wire/anthropic.rs index bd4d3ff00..ba73edaef 100644 --- a/lib/components/fabro-llm/tests/it/wire/anthropic.rs +++ b/lib/components/fabro-llm/tests/it/wire/anthropic.rs @@ -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(&[ diff --git a/lib/components/fabro-llm/tests/it/wire/gemini.rs b/lib/components/fabro-llm/tests/it/wire/gemini.rs index 5cc18fa20..2d1cc2e5e 100644 --- a/lib/components/fabro-llm/tests/it/wire/gemini.rs +++ b/lib/components/fabro-llm/tests/it/wire/gemini.rs @@ -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(&[ diff --git a/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs b/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs index 494806657..1294e2e8a 100644 --- a/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs +++ b/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs @@ -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] diff --git a/lib/components/fabro-llm/tests/it/wire/openai_responses.rs b/lib/components/fabro-llm/tests/it/wire/openai_responses.rs index a9d24a173..079a05455 100644 --- a/lib/components/fabro-llm/tests/it/wire/openai_responses.rs +++ b/lib/components/fabro-llm/tests/it/wire/openai_responses.rs @@ -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] diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index f3576ec97..11e4ae571 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -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); diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs index cd9efd482..7cec9a204 100644 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ b/lib/components/fabro-store/src/slate/projection_cache.rs @@ -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, + parent_id: Option, + ) { + 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 { 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) } diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 68743e9ec..d41647a96 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -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 { diff --git a/lib/foundation/fabro-api/build.rs b/lib/foundation/fabro-api/build.rs index 03c84fd81..fc943e99d 100644 --- a/lib/foundation/fabro-api/build.rs +++ b/lib/foundation/fabro-api/build.rs @@ -362,7 +362,6 @@ fn main() { &[], ), ("LlmOutputKind", "fabro_types::LlmOutputKind", &[]), - ("LlmRetryPhase", "fabro_types::LlmRetryPhase", &[]), ("PermissionLevel", "fabro_types::PermissionLevel", &[]), ( "AgentSessionActivatedProps", diff --git a/lib/foundation/fabro-api/src/lib.rs b/lib/foundation/fabro-api/src/lib.rs index 3576f4bf4..4312acb91 100644 --- a/lib/foundation/fabro-api/src/lib.rs +++ b/lib/foundation/fabro-api/src/lib.rs @@ -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, diff --git a/lib/foundation/fabro-api/tests/stage_projection_round_trip.rs b/lib/foundation/fabro-api/tests/stage_projection_round_trip.rs index c71100e61..5fd11f521 100644 --- a/lib/foundation/fabro-api/tests/stage_projection_round_trip.rs +++ b/lib/foundation/fabro-api/tests/stage_projection_round_trip.rs @@ -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::(); assert_same_type::(); assert_same_type::(); - assert_same_type::(); } #[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 diff --git a/lib/foundation/fabro-types/src/billing.rs b/lib/foundation/fabro-types/src/billing.rs index 172352690..d994df7ea 100644 --- a/lib/foundation/fabro-types/src/billing.rs +++ b/lib/foundation/fabro-types/src/billing.rs @@ -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, }; diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index 4dccb4964..68f5644a2 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -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; diff --git a/lib/foundation/fabro-types/src/run_event/agent.rs b/lib/foundation/fabro-types/src/run_event/agent.rs index e932f72df..414b7be22 100644 --- a/lib/foundation/fabro-types/src/run_event/agent.rs +++ b/lib/foundation/fabro-types/src/run_event/agent.rs @@ -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. diff --git a/lib/foundation/fabro-types/src/run_projection.rs b/lib/foundation/fabro-types/src/run_projection.rs index 86e0e738d..027cce2fc 100644 --- a/lib/foundation/fabro-types/src/run_projection.rs +++ b/lib/foundation/fabro-types/src/run_projection.rs @@ -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 { + 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 diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index 2725b36f5..e3a1a7c0e 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -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 diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 01009fbae..5cd29601d 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -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'; diff --git a/lib/packages/fabro-api-client/src/models/llm-retry-phase.ts b/lib/packages/fabro-api-client/src/models/llm-retry-phase.ts deleted file mode 100644 index b1fcdb7aa..000000000 --- a/lib/packages/fabro-api-client/src/models/llm-retry-phase.ts +++ /dev/null @@ -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];