diff --git a/AGENTS.md b/AGENTS.md index ebd2cb076..1ad635cea 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -53,6 +53,17 @@ The OpenAPI spec at `docs/api-reference/fabro-api.yaml` is the source of truth f 4. `cargo nextest run -p fabro-server` — conformance test catches spec/router drift 5. `cd lib/packages/fabro-api-client && bun run generate` — regenerates TypeScript Axios client +### API type ownership + +- Treat OpenAPI as the source of truth for the wire contract, not as the automatic owner of Rust types. +- Before adding or keeping a generated schema type, search the workspace for an existing hand-written Rust type with the same product meaning. +- If the schema and an existing Rust type have the same semantics and serde shape, reuse the existing type via `lib/crates/fabro-api/build.rs` `with_replacement(...)` instead of generating a parallel API type. +- If two types are close but not identical, prefer proposing changes that align them into one canonical type rather than accepting small drift. It is usually better to iterate the API now than to create permanently split Rust/API types. +- Keep a separate API DTO only when the API is intentionally a projection, summary, or presentation-specific view of internal state. In that case, give it a distinct API-facing name instead of reusing the internal concept name. +- Treat `ApiFoo` aliases and `foo_to_api` / `foo_from_api` adapters as a smell unless they represent a real semantic boundary. They should not exist only to bridge accidental duplicate types. +- If a type is shared across crates and is part of the core product vocabulary, move it to a shared crate first, then make `fabro-api` reuse it. +- For every new `with_replacement(...)`, add a `fabro-api` test that proves type identity and JSON parity with the OpenAPI schema. + ## Architecture Fabro is an AI-powered workflow orchestration platform. Workflows are defined as Graphviz graphs, where each node is a stage (agent, prompt, command, conditional, human, parallel, etc.) executed by the workflow engine. diff --git a/Cargo.lock b/Cargo.lock index 5a1f79cc0..c44d03f91 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1544,6 +1544,7 @@ name = "fabro-api" version = "0.208.0-nightly.1" dependencies = [ "chrono", + "fabro-types", "openapiv3", "prettyplease", "progenitor", diff --git a/apps/fabro-web/app/routes/run-stages.test.ts b/apps/fabro-web/app/routes/run-stages.test.ts new file mode 100644 index 000000000..d10b5b2d2 --- /dev/null +++ b/apps/fabro-web/app/routes/run-stages.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, test } from "bun:test"; + +import { isSafeMarkdownHref } from "./run-stages"; + +describe("isSafeMarkdownHref", () => { + test("rejects protocol-relative URLs", () => { + expect(isSafeMarkdownHref("//attacker.example/pixel.png")).toBe(false); + }); + + test("accepts root-relative, hash, http, https, and mailto URLs", () => { + expect(isSafeMarkdownHref("/runs/run-1")).toBe(true); + expect(isSafeMarkdownHref("#section-1")).toBe(true); + expect(isSafeMarkdownHref("https://fabro.sh")).toBe(true); + expect(isSafeMarkdownHref("http://localhost:3000")).toBe(true); + expect(isSafeMarkdownHref("mailto:test@example.com")).toBe(true); + }); +}); diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index 9b205d1df..5ceb146cf 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/apps/fabro-web/app/routes/run-stages.tsx @@ -1,6 +1,37 @@ import { useEffect, useMemo, useRef, useState } from "react"; import { useParams } from "react-router"; -import { marked } from "marked"; +import { Marked } from "marked"; + +const SAFE_HTTP_URL_RE = /^https?:\/\//i; +const SAFE_MAILTO_URL_RE = /^mailto:/i; + +export function isSafeMarkdownHref(href: string): boolean { + return ( + SAFE_HTTP_URL_RE.test(href) || + SAFE_MAILTO_URL_RE.test(href) || + href.startsWith("#") || + (href.startsWith("/") && !href.startsWith("//")) + ); +} + +const markedSafe = new Marked(); +markedSafe.use({ + async: false, + walkTokens(token) { + if ( + (token.type === "link" || token.type === "image") && + typeof token.href === "string" && + !isSafeMarkdownHref(token.href) + ) { + token.href = ""; + } + }, + renderer: { + html() { + return ""; + }, + }, +}); import { CommandLineIcon, ChatBubbleLeftIcon, PlayIcon } from "@heroicons/react/24/outline"; import { ToolBlock } from "../components/tool-use"; import type { ToolUse } from "../components/tool-use"; @@ -167,7 +198,7 @@ export async function loader({ request, params }: any) { } function Markdown({ content }: { content: string }) { - const html = useMemo(() => marked.parse(content, { async: false }) as string, [content]); + const html = useMemo(() => markedSafe.parse(content, { async: false }) as string, [content]); return (
Result<(String, std::collections::HashMap), String> { let sandbox = self.sandbox.as_ref(); - let cmd_str = command.join(" "); + let cmd_str = command + .iter() + .map(|arg| fabro_sandbox::shell_quote(arg)) + .collect::>() + .join(" "); - // Launch the server detached with setsid so Daytona's exec doesn't block + // Launch the server detached with setsid so Daytona's exec doesn't block. + // shell_quote the inner command for the outer `sh -c` so a single quote + // or metacharacter in any argv element can't break out of the wrapper. + let inner = format!("{cmd_str} > /tmp/mcp_server_stdout.log 2>/tmp/mcp_server_stderr.log"); let launch_script = format!( - "setsid sh -c '{cmd_str} > /tmp/mcp_server_stdout.log 2>/tmp/mcp_server_stderr.log' \ - /dev/null 2>&1 &\necho $!" + "setsid sh -c {quoted} /dev/null 2>&1 &\necho $!", + quoted = fabro_sandbox::shell_quote(&inner) ); let env_ref = if env.is_empty() { None } else { Some(env) }; let launch_result = sandbox diff --git a/lib/crates/fabro-api/Cargo.toml b/lib/crates/fabro-api/Cargo.toml index 98ef6836d..cf3335ea6 100644 --- a/lib/crates/fabro-api/Cargo.toml +++ b/lib/crates/fabro-api/Cargo.toml @@ -15,6 +15,7 @@ wildcard_imports = "warn" [dependencies] chrono = { workspace = true, features = ["serde"] } +fabro-types = { path = "../fabro-types" } progenitor-client = "0.13" regress = "0.10" reqwest.workspace = true diff --git a/lib/crates/fabro-api/build.rs b/lib/crates/fabro-api/build.rs index df667b1d5..168fe8bb9 100644 --- a/lib/crates/fabro-api/build.rs +++ b/lib/crates/fabro-api/build.rs @@ -6,7 +6,7 @@ use std::path::{Path, PathBuf}; use std::{env, fs}; -use progenitor::{GenerationSettings, Generator, InterfaceStyle}; +use progenitor::{GenerationSettings, Generator, InterfaceStyle, TypeImpl}; /// Recursively convert OpenAPI 3.1 `type: "null"` patterns to 3.0 `nullable: /// true`. @@ -160,6 +160,27 @@ fn main() { let mut settings = GenerationSettings::default(); settings.with_interface(InterfaceStyle::Builder); + let replacements: &[(&str, &str, &[TypeImpl])] = &[ + ("RunStatus", "fabro_types::status::RunStatus", &[ + TypeImpl::FromStr, + TypeImpl::Display, + ]), + ("StatusReason", "fabro_types::status::StatusReason", &[]), + ("BlockedReason", "fabro_types::status::BlockedReason", &[]), + ( + "RunControlAction", + "fabro_types::status::RunControlAction", + &[], + ), + ( + "RunStatusRecord", + "fabro_types::status::RunStatusRecord", + &[], + ), + ]; + for (name, path, impls) in replacements { + settings.with_replacement(*name, *path, impls.iter().copied()); + } let mut generator = Generator::new(&settings); let tokens = generator diff --git a/lib/crates/fabro-api/src/lib.rs b/lib/crates/fabro-api/src/lib.rs index ca2774c83..b039d6c12 100644 --- a/lib/crates/fabro-api/src/lib.rs +++ b/lib/crates/fabro-api/src/lib.rs @@ -13,4 +13,11 @@ mod generated { include!(concat!(env!("OUT_DIR"), "/codegen.rs")); } -pub use generated::{Client as ApiClient, types}; +pub mod types { + pub use fabro_types::status::{ + BlockedReason, RunControlAction, RunStatus, RunStatusRecord, StatusReason, + }; + + pub use crate::generated::types::*; +} +pub use generated::Client as ApiClient; diff --git a/lib/crates/fabro-api/tests/status_round_trip.rs b/lib/crates/fabro-api/tests/status_round_trip.rs new file mode 100644 index 000000000..894df6ccb --- /dev/null +++ b/lib/crates/fabro-api/tests/status_round_trip.rs @@ -0,0 +1,116 @@ +use std::any::{TypeId, type_name}; + +use chrono::{TimeZone, Utc}; +use fabro_api::types::{ + BlockedReason as ApiBlockedReason, RunControlAction as ApiRunControlAction, + RunStatus as ApiRunStatus, RunStatusRecord as ApiRunStatusRecord, + StatusReason as ApiStatusReason, +}; +use fabro_types::status::{ + BlockedReason, RunControlAction, RunStatus, RunStatusRecord, StatusReason, +}; +use serde::Serialize; +use serde_json::{Value, json}; + +#[test] +fn status_family_reuses_domain_types() { + assert_same_type::(); + assert_same_type::(); + assert_same_type::(); + assert_same_type::(); + assert_same_type::(); +} + +// The `status_family_reuses_domain_types` assertions above prove each API type +// is the same type as its domain counterpart, so each variant below only needs +// to be asserted once to lock in the OpenAPI string token. + +#[test] +fn run_status_json_tokens_match_openapi() { + assert_string_json(RunStatus::Submitted, "submitted"); + assert_string_json(RunStatus::Queued, "queued"); + assert_string_json(RunStatus::Starting, "starting"); + assert_string_json(RunStatus::Running, "running"); + assert_string_json(RunStatus::Blocked, "blocked"); + assert_string_json(RunStatus::Paused, "paused"); + assert_string_json(RunStatus::Removing, "removing"); + assert_string_json(RunStatus::Succeeded, "succeeded"); + assert_string_json(RunStatus::Failed, "failed"); + assert_string_json(RunStatus::Dead, "dead"); + assert_string_json(RunStatus::Archived, "archived"); +} + +#[test] +fn status_reason_json_tokens_match_openapi() { + assert_string_json(StatusReason::Completed, "completed"); + assert_string_json(StatusReason::PartialSuccess, "partial_success"); + assert_string_json(StatusReason::WorkflowError, "workflow_error"); + assert_string_json(StatusReason::Cancelled, "cancelled"); + assert_string_json(StatusReason::Terminated, "terminated"); + assert_string_json(StatusReason::TransientInfra, "transient_infra"); + assert_string_json(StatusReason::BudgetExhausted, "budget_exhausted"); + assert_string_json(StatusReason::LaunchFailed, "launch_failed"); + assert_string_json(StatusReason::BootstrapFailed, "bootstrap_failed"); + assert_string_json(StatusReason::SandboxInitFailed, "sandbox_init_failed"); + assert_string_json(StatusReason::SandboxInitializing, "sandbox_initializing"); +} + +#[test] +fn blocked_reason_json_tokens_match_openapi() { + assert_string_json(BlockedReason::HumanInputRequired, "human_input_required"); +} + +#[test] +fn run_control_action_json_tokens_match_openapi() { + assert_string_json(RunControlAction::Cancel, "cancel"); + assert_string_json(RunControlAction::Pause, "pause"); + assert_string_json(RunControlAction::Unpause, "unpause"); +} + +#[test] +fn run_status_record_json_matches_openapi_shape() { + let updated_at = Utc + .with_ymd_and_hms(2026, 1, 2, 3, 4, 5) + .single() + .expect("fixed timestamp should be valid"); + let expected = json!({ + "status": "failed", + "status_reason": "cancelled", + "blocked_reason": "human_input_required", + "updated_at": "2026-01-02T03:04:05Z" + }); + + let record = RunStatusRecord { + status: RunStatus::Failed, + status_reason: Some(StatusReason::Cancelled), + blocked_reason: Some(BlockedReason::HumanInputRequired), + updated_at, + }; + assert_eq!(serde_json::to_value(&record).unwrap(), expected); + + let round_trip: RunStatusRecord = serde_json::from_value(expected).unwrap(); + assert_eq!(round_trip.status, RunStatus::Failed); + assert_eq!(round_trip.status_reason, Some(StatusReason::Cancelled)); + assert_eq!( + round_trip.blocked_reason, + Some(BlockedReason::HumanInputRequired) + ); + assert_eq!(round_trip.updated_at, updated_at); +} + +fn assert_same_type() { + assert_eq!( + TypeId::of::(), + TypeId::of::(), + "{} should be the same type as {}", + type_name::(), + type_name::() + ); +} + +fn assert_string_json(value: T, expected: &str) { + assert_eq!( + serde_json::to_value(value).unwrap(), + Value::String(expected.into()) + ); +} diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index d6fae8e7d..814f3832c 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -25,18 +25,17 @@ use bytes::Bytes; pub use fabro_api::types::{ AggregateBilling, AggregateBillingTotals, ApiQuestion, ApiQuestionOption, AppendEventResponse, ArtifactEntry, ArtifactListResponse, BilledTokenCounts as ApiBilledTokenCounts, BillingByModel, - BillingStageRef, BlockedReason as ApiBlockedReason, CompletionContentPart, CompletionMessage, - CompletionMessageRole, CompletionResponse, CompletionToolChoiceMode, CompletionUsage, - CreateCompletionRequest, CreateSecretRequest, DeleteSecretRequest, DiskUsageResponse, - DiskUsageRunRow, DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope, ModelReference, - PaginatedEventList, PaginatedRunList, PaginationMeta, PreflightResponse, PreviewUrlRequest, - PreviewUrlResponse, PruneRunEntry, PruneRunsRequest, PruneRunsResponse, - QuestionType as ApiQuestionType, RenderWorkflowGraphDirection, RenderWorkflowGraphRequest, - RunArtifactEntry, RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals, - RunControlAction as ApiRunControlAction, RunError, RunManifest, RunStage, RunStatus, - RunStatusResponse, SandboxFileEntry, SandboxFileListResponse, SecretType as ApiSecretType, - ServerSettings, SshAccessRequest, SshAccessResponse, StageStatus as ApiStageStatus, - StartRunRequest, StatusReason as ApiStatusReason, SubmitAnswerRequest, SystemFeatures, + BillingStageRef, CompletionContentPart, CompletionMessage, CompletionMessageRole, + CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest, + CreateSecretRequest, DeleteSecretRequest, DiskUsageResponse, DiskUsageRunRow, + DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope, ModelReference, PaginatedEventList, + PaginatedRunList, PaginationMeta, PreflightResponse, PreviewUrlRequest, PreviewUrlResponse, + PruneRunEntry, PruneRunsRequest, PruneRunsResponse, QuestionType as ApiQuestionType, + RenderWorkflowGraphDirection, RenderWorkflowGraphRequest, RunArtifactEntry, + RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals, RunError, RunManifest, + RunStage, RunStatusResponse, SandboxFileEntry, SandboxFileListResponse, + SecretType as ApiSecretType, ServerSettings, SshAccessRequest, SshAccessResponse, + StageStatus as ApiStageStatus, StartRunRequest, SubmitAnswerRequest, SystemFeatures, SystemInfoResponse, SystemRunCounts, WriteBlobResponse, }; use fabro_auth::parse_credential_secret; @@ -86,9 +85,7 @@ use fabro_workflow::records::Checkpoint; use fabro_workflow::run_lookup::{ RunInfo, StatusFilter, filter_runs, scan_runs_with_summaries, scratch_base, }; -use fabro_workflow::run_status::{ - RunStatus as WorkflowRunStatus, StatusReason as WorkflowStatusReason, -}; +use fabro_workflow::run_status::{RunStatus, StatusReason}; use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation}; use object_store::memory::InMemory as MemoryObjectStore; use rand::TryRngCore; @@ -2691,16 +2688,14 @@ fn test_secret_store_path() -> PathBuf { dir.join("secrets.json") } -fn board_column(status: WorkflowRunStatus) -> Option<&'static str> { +fn board_column(status: RunStatus) -> Option<&'static str> { match status { - WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting => { - Some("initializing") - } - WorkflowRunStatus::Running | WorkflowRunStatus::Paused => Some("running"), - WorkflowRunStatus::Blocked => Some("blocked"), - WorkflowRunStatus::Succeeded => Some("succeeded"), - WorkflowRunStatus::Failed | WorkflowRunStatus::Dead => Some("failed"), - WorkflowRunStatus::Removing | WorkflowRunStatus::Archived => None, + RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"), + RunStatus::Running | RunStatus::Paused => Some("running"), + RunStatus::Blocked => Some("blocked"), + RunStatus::Succeeded => Some("succeeded"), + RunStatus::Failed | RunStatus::Dead => Some("failed"), + RunStatus::Removing | RunStatus::Archived => None, } } @@ -2755,9 +2750,9 @@ fn summary_to_api_run_summary(summary: fabro_types::RunSummary) -> serde_json::V "repository": { "name": repository }, "start_time": summary.start_time.map(|time| time.to_rfc3339()), "status": summary.status, - "status_reason": summary.status_reason.map(api_status_reason), - "blocked_reason": summary.blocked_reason.map(api_blocked_reason), - "pending_control": summary.pending_control.map(api_pending_control), + "status_reason": summary.status_reason, + "blocked_reason": summary.blocked_reason, + "pending_control": summary.pending_control, "duration_ms": summary.duration_ms, "elapsed_secs": elapsed_secs(summary.duration_ms), "total_usd_micros": summary.total_usd_micros, @@ -2889,7 +2884,7 @@ async fn list_runs( let include_archived = params.include_archived; let items = runs .into_iter() - .filter(|summary| include_archived || summary.status != WorkflowRunStatus::Archived) + .filter(|summary| include_archived || summary.status != RunStatus::Archived) .map(summary_to_api_run_summary) .collect::>(); let (data, has_more) = paginate_items(items, ¶ms.pagination()); @@ -3351,29 +3346,26 @@ struct LiveWorkerProcess { fn failure_for_incomplete_run( pending_control: Option, terminated_message: String, -) -> (WorkflowError, Option) { +) -> (WorkflowError, Option) { if pending_control == Some(RunControlAction::Cancel) { - ( - WorkflowError::Cancelled, - Some(WorkflowStatusReason::Cancelled), - ) + (WorkflowError::Cancelled, Some(StatusReason::Cancelled)) } else { ( WorkflowError::engine(terminated_message), - Some(WorkflowStatusReason::Terminated), + Some(StatusReason::Terminated), ) } } -fn should_reconcile_run_on_startup(status: WorkflowRunStatus) -> bool { +fn should_reconcile_run_on_startup(status: RunStatus) -> bool { matches!( status, - WorkflowRunStatus::Queued - | WorkflowRunStatus::Starting - | WorkflowRunStatus::Running - | WorkflowRunStatus::Blocked - | WorkflowRunStatus::Paused - | WorkflowRunStatus::Removing + RunStatus::Queued + | RunStatus::Starting + | RunStatus::Running + | RunStatus::Blocked + | RunStatus::Paused + | RunStatus::Removing ) } @@ -3533,7 +3525,7 @@ async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::Cancelled, duration_ms: 0, - reason: Some(WorkflowStatusReason::Cancelled), + reason: Some(StatusReason::Cancelled), git_commit_sha: None, final_patch: None, }, @@ -3587,28 +3579,6 @@ fn managed_run( } } -fn api_status_from_workflow(status: WorkflowRunStatus) -> RunStatus { - match status { - WorkflowRunStatus::Submitted => RunStatus::Submitted, - WorkflowRunStatus::Queued => RunStatus::Queued, - WorkflowRunStatus::Starting => RunStatus::Starting, - WorkflowRunStatus::Running => RunStatus::Running, - WorkflowRunStatus::Blocked => RunStatus::Blocked, - WorkflowRunStatus::Paused => RunStatus::Paused, - WorkflowRunStatus::Removing => RunStatus::Removing, - WorkflowRunStatus::Succeeded => RunStatus::Succeeded, - WorkflowRunStatus::Failed => RunStatus::Failed, - WorkflowRunStatus::Dead => RunStatus::Dead, - WorkflowRunStatus::Archived => RunStatus::Archived, - } -} - -fn api_blocked_reason(reason: BlockedReason) -> ApiBlockedReason { - match reason { - BlockedReason::HumanInputRequired => ApiBlockedReason::HumanInputRequired, - } -} - fn worker_mode_arg(mode: RunExecutionMode) -> &'static str { match mode { RunExecutionMode::Start => "start", @@ -3616,43 +3586,19 @@ fn worker_mode_arg(mode: RunExecutionMode) -> &'static str { } } -fn api_status_reason(reason: WorkflowStatusReason) -> ApiStatusReason { - match reason { - WorkflowStatusReason::Completed => ApiStatusReason::Completed, - WorkflowStatusReason::PartialSuccess => ApiStatusReason::PartialSuccess, - WorkflowStatusReason::WorkflowError => ApiStatusReason::WorkflowError, - WorkflowStatusReason::Cancelled => ApiStatusReason::Cancelled, - WorkflowStatusReason::Terminated => ApiStatusReason::Terminated, - WorkflowStatusReason::TransientInfra => ApiStatusReason::TransientInfra, - WorkflowStatusReason::BudgetExhausted => ApiStatusReason::BudgetExhausted, - WorkflowStatusReason::LaunchFailed => ApiStatusReason::LaunchFailed, - WorkflowStatusReason::BootstrapFailed => ApiStatusReason::BootstrapFailed, - WorkflowStatusReason::SandboxInitFailed => ApiStatusReason::SandboxInitFailed, - WorkflowStatusReason::SandboxInitializing => ApiStatusReason::SandboxInitializing, - } -} - -fn api_pending_control(action: RunControlAction) -> ApiRunControlAction { - match action { - RunControlAction::Cancel => ApiRunControlAction::Cancel, - RunControlAction::Pause => ApiRunControlAction::Pause, - RunControlAction::Unpause => ApiRunControlAction::Unpause, - } -} - async fn load_run_status_metadata( state: &AppState, run_id: RunId, ) -> ( - Option, - Option, - Option, + Option, + Option, + Option, ) { match state.store.runs().find(&run_id).await { Ok(Some(summary)) => ( - summary.status_reason.map(api_status_reason), - summary.blocked_reason.map(api_blocked_reason), - summary.pending_control.map(api_pending_control), + summary.status_reason, + summary.blocked_reason, + summary.pending_control, ), _ => (None, None, None), } @@ -4334,7 +4280,7 @@ async fn start_run( } else if let Some(record) = run_state.status.as_ref() { if !matches!( record.status, - WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting + RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting ) { return ApiError::new( StatusCode::CONFLICT, @@ -4761,7 +4707,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(err.to_string()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4783,7 +4729,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(message.clone()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4813,7 +4759,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(message.clone()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4834,7 +4780,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(message.clone()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4867,7 +4813,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(err.to_string()), duration_ms: 0, - reason: Some(WorkflowStatusReason::Terminated), + reason: Some(StatusReason::Terminated), git_commit_sha: None, final_patch: None, }, @@ -4953,7 +4899,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { let mut runs = state.runs.lock().expect("runs lock poisoned"); if let Some(managed_run) = runs.get_mut(&run_id) { if let Some(status) = final_state.status.as_ref() { - managed_run.status = api_status_from_workflow(status.status); + managed_run.status = status.status; } else if !wait_status.success() { managed_run.status = RunStatus::Failed; } @@ -6272,7 +6218,7 @@ async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option Response { ) .into_response(); }; - let status = api_status_from_workflow(record.status); - let status_reason = record.status_reason.map(api_status_reason); - let blocked_reason = record.blocked_reason.map(api_blocked_reason); + let status = record.status; + let status_reason = record.status_reason; + let blocked_reason = record.blocked_reason; ( StatusCode::OK, Json(RunStatusResponse { @@ -9961,8 +9907,8 @@ level = "debug" let run_store = state.store.open_run_reader(&run_id).await.unwrap(); let status = run_store.state().await.unwrap().status.unwrap(); - assert_eq!(status.status, WorkflowRunStatus::Failed); - assert_eq!(status.status_reason, Some(WorkflowStatusReason::Cancelled)); + assert_eq!(status.status, RunStatus::Failed); + assert_eq!(status.status_reason, Some(StatusReason::Cancelled)); } #[tokio::test] @@ -10118,7 +10064,7 @@ level = "debug" assert_eq!(body["pending_control"], serde_json::Value::Null); let summary = state.store.runs().find(&run_id).await.unwrap().unwrap(); - assert_eq!(summary.status, WorkflowRunStatus::Paused); + assert_eq!(summary.status, RunStatus::Paused); assert_eq!( summary.blocked_reason, Some(BlockedReason::HumanInputRequired) @@ -10204,7 +10150,7 @@ level = "debug" assert_eq!(body["pending_control"], serde_json::Value::Null); let summary = state.store.runs().find(&run_id).await.unwrap().unwrap(); - assert_eq!(summary.status, WorkflowRunStatus::Blocked); + assert_eq!(summary.status, RunStatus::Blocked); assert_eq!( summary.blocked_reason, Some(BlockedReason::HumanInputRequired) @@ -10255,7 +10201,7 @@ level = "debug" .state() .await .unwrap(); - assert_eq!(run_1.status.unwrap().status, WorkflowRunStatus::Submitted); + assert_eq!(run_1.status.unwrap().status, RunStatus::Submitted); let run_2 = state .store @@ -10266,11 +10212,8 @@ level = "debug" .await .unwrap(); let run_2_status = run_2.status.unwrap(); - assert_eq!(run_2_status.status, WorkflowRunStatus::Failed); - assert_eq!( - run_2_status.status_reason, - Some(WorkflowStatusReason::Terminated) - ); + assert_eq!(run_2_status.status, RunStatus::Failed); + assert_eq!(run_2_status.status_reason, Some(StatusReason::Terminated)); let run_3 = state .store @@ -10281,11 +10224,8 @@ level = "debug" .await .unwrap(); let run_3_status = run_3.status.unwrap(); - assert_eq!(run_3_status.status, WorkflowRunStatus::Failed); - assert_eq!( - run_3_status.status_reason, - Some(WorkflowStatusReason::Cancelled) - ); + assert_eq!(run_3_status.status, RunStatus::Failed); + assert_eq!(run_3_status.status_reason, Some(StatusReason::Cancelled)); assert_eq!(run_3.pending_control, None); } @@ -10356,11 +10296,8 @@ level = "debug" .await .unwrap(); let run_status = run_state.status.unwrap(); - assert_eq!(run_status.status, WorkflowRunStatus::Failed); - assert_eq!( - run_status.status_reason, - Some(WorkflowStatusReason::Terminated) - ); + assert_eq!(run_status.status, RunStatus::Failed); + assert_eq!(run_status.status_reason, Some(StatusReason::Terminated)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -10446,8 +10383,8 @@ timeout = "30s" let mut status_record = None; for _ in 0..50 { if let Some(record) = run_store.state().await.unwrap().status { - if record.status == WorkflowRunStatus::Failed - && record.status_reason == Some(WorkflowStatusReason::Cancelled) + if record.status == RunStatus::Failed + && record.status_reason == Some(StatusReason::Cancelled) { status_record = Some(record); break; @@ -10457,11 +10394,8 @@ timeout = "30s" } let status_record = status_record.expect("status record should be persisted"); - assert_eq!(status_record.status, WorkflowRunStatus::Failed); - assert_eq!( - status_record.status_reason, - Some(WorkflowStatusReason::Cancelled) - ); + assert_eq!(status_record.status, RunStatus::Failed); + assert_eq!(status_record.status_reason, Some(StatusReason::Cancelled)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] diff --git a/lib/crates/fabro-types/src/status.rs b/lib/crates/fabro-types/src/status.rs index 00f78b163..60f02f5ae 100644 --- a/lib/crates/fabro-types/src/status.rs +++ b/lib/crates/fabro-types/src/status.rs @@ -172,6 +172,57 @@ pub enum StatusReason { SandboxInitializing, } +impl fmt::Display for StatusReason { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + Self::Completed => "completed", + Self::PartialSuccess => "partial_success", + Self::WorkflowError => "workflow_error", + Self::Cancelled => "cancelled", + Self::Terminated => "terminated", + Self::TransientInfra => "transient_infra", + Self::BudgetExhausted => "budget_exhausted", + Self::LaunchFailed => "launch_failed", + Self::BootstrapFailed => "bootstrap_failed", + Self::SandboxInitFailed => "sandbox_init_failed", + Self::SandboxInitializing => "sandbox_initializing", + }; + f.write_str(s) + } +} + +impl FromStr for StatusReason { + type Err = ParseStatusReasonError; + + fn from_str(s: &str) -> Result { + match s { + "completed" => Ok(Self::Completed), + "partial_success" => Ok(Self::PartialSuccess), + "workflow_error" => Ok(Self::WorkflowError), + "cancelled" => Ok(Self::Cancelled), + "terminated" => Ok(Self::Terminated), + "transient_infra" => Ok(Self::TransientInfra), + "budget_exhausted" => Ok(Self::BudgetExhausted), + "launch_failed" => Ok(Self::LaunchFailed), + "bootstrap_failed" => Ok(Self::BootstrapFailed), + "sandbox_init_failed" => Ok(Self::SandboxInitFailed), + "sandbox_initializing" => Ok(Self::SandboxInitializing), + _ => Err(ParseStatusReasonError(s.to_string())), + } + } +} + +#[derive(Debug, Clone)] +pub struct ParseStatusReasonError(String); + +impl fmt::Display for ParseStatusReasonError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "invalid status reason: {:?}", self.0) + } +} + +impl std::error::Error for ParseStatusReasonError {} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum BlockedReason { @@ -211,7 +262,7 @@ impl RunStatusRecord { mod tests { use std::str::FromStr; - use super::{InvalidTransition, RunStatus}; + use super::{InvalidTransition, RunStatus, StatusReason}; #[test] fn queued_and_blocked_parse_and_format() { @@ -249,6 +300,13 @@ mod tests { assert_eq!(parsed.to_string(), "archived"); } + #[test] + fn status_reason_parses_and_round_trips() { + let parsed = StatusReason::from_str("cancelled").expect("cancelled should parse"); + assert_eq!(parsed, StatusReason::Cancelled); + assert_eq!(parsed.to_string(), "cancelled"); + } + #[test] fn terminal_statuses_can_transition_to_archived() { assert!(RunStatus::Succeeded.can_transition_to(RunStatus::Archived)); diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 947ffd593..d19311e06 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1348,7 +1348,13 @@ fn billed_token_counts_from_llm(usage: &LlmTokenCounts) -> BilledTokenCounts { } fn stage_status_from_string(status: &str) -> StageStatus { - serde_json::from_value(Value::String(status.to_string())).expect("valid stage status") + status.parse().unwrap_or_else(|_| { + tracing::warn!( + status, + "unknown stage status in StageCompleted event; using Fail" + ); + StageStatus::Fail + }) } fn stored_event_fields(event: &Event, scope: Option<&StageScope>) -> StoredEventFields {