scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-16 23:43:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1
fabro/lib/apps/fabro-server/src/server.rs
Bryan Helmkamp 80bc51c40e
Open sandbox provider identity to plugin kinds 
SandboxProviderKind is now a validated string newtype instead of a
closed enum. The bundled kinds (local, docker, daytona) keep their
constants and a BundledProvider enum for the code paths that still
dispatch on them; any other well-formed sandbox-driver kind name is
accepted and names a plugin executable. EnvironmentProvider is gone:
environment settings carry SandboxProviderKind directly, and
is_clone_based is replaced by a workspace policy where local runs in a
designated directory and every other provider clones.

Server sandbox policy is keyed by kind. [server.sandbox.providers.<kind>]
accepts the bundled kinds with `enabled` and any plugin kind with its
launch settings (path, sha256, dev, args, env, inherit_env); bundled
kinds reject the plugin keys and a kind with no entry is disabled. The
OpenAPI schema, generated Rust and TypeScript clients, web settings
pages, and docs follow. The environments table drops its provider CHECK
enumeration in favour of the kind name rules so a plugin environment
can be stored.

Bundled-only code paths (run start, preflight, reconnect, terminal,
details) now fail with an explicit message for a plugin kind until the
driver construction function lands in the next step.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-09 15:17:32 -06:00

4593 lines
161 KiB
Rust
Raw Blame History

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::process::Stdio;
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, LazyLock, Mutex, RwLock};
use std::time::{Duration, Instant};
use anyhow::Context as _;
use axum::body::Body;
#[cfg(test)]
use axum::body::to_bytes;
use axum::extract::{self as axum_extract, DefaultBodyLimit, Path, Query, State};
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::middleware::{self, Next};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use axum_extra::extract::cookie::Key;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use bytes::Bytes;
use chrono::{DateTime, Utc};
pub use fabro_api::types::{
AggregateBilling, AggregateBillingTotals, ApiQuestion, AppendEventResponse, ArtifactEntry,
ArtifactListResponse, BatchDeleteRunsRequest, BatchDeleteRunsResponse, BatchDeleteRunsResult,
BatchDeleteRunsResultOutcome, BatchDeleteRunsSummary, BatchRunLifecycleRequest,
BatchRunLifecycleResponse, BatchRunLifecycleResult, BatchRunLifecycleResultOutcome,
BatchRunLifecycleSummary, BillingByModel, BillingStageRef, CloseRunPullRequestResponse,
CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest,
CreateRunPullRequestRequest, CreateSecretRequest, CreateVariableRequest, DeleteRunResponse,
DeleteRunSandbox, DeleteSecretRequest, DenyRunRequest, DiskUsageResponse, DiskUsageRunRow,
DiskUsageSummaryRow, ErrorResponseEntry, ForkRequest, ForkResponse, IntegrationConnectionKind,
IntegrationConnectionState, IntegrationConnectionStatus, IntegrationProvider,
IntegrationStatus, LinkRunPullRequestRequest, MergeRunPullRequestRequest,
MergeRunPullRequestResponse, ModelReference, PaginatedEventList, PaginatedRunList,
PaginationMeta, PreflightResponse, PreviewUrlRequest, PreviewUrlResponse, Provider,
ProviderCredentialTestRequest, ProviderCredentialTestResponse, ProviderList, PruneRunEntry,
PruneRunsRequest, PruneRunsResponse, RenderWorkflowGraphDirection, RenderWorkflowGraphRequest,
RewindRequest, RewindResponse, Run, RunArtifactEntry, RunArtifactListResponse, RunBilling,
RunBillingStage, RunBillingTotals, RunError, RunManifest, RunStage, SandboxDetails,
SandboxFileEntry, SandboxFileListResponse, SandboxService, SandboxServiceListResponse,
SshAccessRequest, SshAccessResponse, StageHandler, StageState, StartRunRequest,
SubmitAnswerRequest, SystemCpuResourceScope, SystemCpuResources, SystemDiskResourceScope,
SystemDiskResources, SystemInfoResponse, SystemIntegrationStatus, SystemIntegrationsResponse,
SystemMemoryResourceScope, SystemMemoryResources, SystemRepairRunIssue,
SystemRepairRunsResponse, SystemResourcesResponse, SystemRunCounts, TimelineEntryResponse,
UpdateVariableRequest, VariableListResponse, VncPreviewResponse, WriteBlobResponse,
};
use fabro_auth::{CredentialSource, SqlVaultCredentialSource, auth_issue_message};
use fabro_automation::{self, AutomationStore};
use fabro_config::daemon::ServerDaemon;
use fabro_config::{RunLayer, Storage, WorkflowSettingsBuilder};
use fabro_db::DbPool;
use fabro_environment::EnvironmentStore;
use fabro_interview::{
Answer, AnswerSubmission, ControlInterviewer, Interviewer, Question, WorkerControlEnvelope,
};
use fabro_llm::client::Client as LlmClient;
use fabro_llm::generate::{GenerateParams, generate_object};
use fabro_llm::model_test::run_model_test;
use fabro_llm::types::{
FinishReason, Message as LlmMessage, Request as LlmRequest, ToolChoice, ToolDefinition,
};
use fabro_mcp_store::McpServerStore;
use fabro_model::catalog::LlmCatalogSettings;
use fabro_model::{BilledTokenCounts, Catalog, ModelRef, ModelTestMode, ProviderId};
use fabro_redact::redact_jsonl_line;
use fabro_sandbox::daytona::{self, DaytonaSandbox};
use fabro_sandbox::details::sandbox_details;
use fabro_sandbox::reconnect::reconnect_for_run;
use fabro_sandbox::{
DaytonaSandboxProvider, DockerSandboxProvider, LocalSandboxProvider, Sandbox, SandboxProvider,
SandboxProviderRegistry,
};
use fabro_slack::client::{PostedMessage as SlackPostedMessage, SlackClient};
use fabro_slack::config::{
SlackCredentialResolution,
resolve_credentials_status_with_lookup as resolve_slack_credentials_status_with_lookup,
};
use fabro_slack::payload::SlackAnswerSubmission;
use fabro_slack::threads::ThreadRegistry;
use fabro_slack::{blocks as slack_blocks, connection as slack_connection};
use fabro_static::EnvVars;
use fabro_store::{
ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, Database, EventEnvelope,
EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore,
StageArtifactEntry, StageId,
};
#[cfg(test)]
use fabro_types::BlockedReason;
use fabro_types::settings::RunNamespace;
use fabro_types::settings::run::{NotificationRouteSettings, RunMode};
use fabro_types::settings::server::{
GithubIntegrationSettings, GithubIntegrationStrategy, LogDestination,
};
use fabro_types::{
AgentBackend, AskFabro, AskFabroUnavailableReason, BlobHash, EventBody,
InterviewQuestionRecord, PairId, PairMessageId, PairTarget, PendingReason, Principal,
PullRequestLink, QuestionType, RunControlAction, RunEvent, RunId, RunRunnableSource,
RunStatusKind, SandboxProviderKind, ServerSettings, SessionCapability,
};
use fabro_util::error::{
SharedError, collect_causes, render_compact_with_causes, render_with_causes,
};
use fabro_util::version::FABRO_VERSION;
use fabro_variable::{Error as VariableError, VariableStore};
use fabro_vault::{SecretStore, SecretStoreError, SecretType, Vault};
use fabro_workflow::artifact_upload::ArtifactSink;
#[cfg(test)]
use fabro_workflow::command_log::command_log_path;
use fabro_workflow::event::{self as workflow_event, Emitter};
use fabro_workflow::handler::HandlerRegistry;
use fabro_workflow::pipeline::Persisted;
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::{FailureReason, RunStatus, SuccessReason};
use fabro_workflow::{Error as WorkflowError, operations, pull_request};
use futures_util::future::join_all;
use sha2::{Digest, Sha256};
use tempfile::NamedTempFile;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWriteExt, BufReader};
use tokio::process::Command;
use tokio::runtime::Builder as TokioRuntimeBuilder;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{
Mutex as AsyncMutex, Notify, RwLock as AsyncRwLock, Semaphore, broadcast, mpsc, oneshot,
};
use tokio::task::spawn_blocking;
use tokio::time::{sleep, timeout};
use tokio_stream::StreamExt;
use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream};
use tokio_util::sync::CancellationToken;
use tower::{ServiceExt, service_fn};
use tower_http::compression::predicate::{DefaultPredicate, NotForContentType, Predicate};
use tower_http::compression::{CompressionLayer, CompressionLevel};
use tracing::{Instrument, debug, error, info, warn};
use ulid::Ulid;
use crate::auth::{self, GithubEndpoints, auth_translation_middleware, demo_routing_middleware};
use crate::automation_materializer::{
AutomationRunMaterializeInput, AutomationRunMaterialized, AutomationRunMaterializer,
ProductionAutomationRunMaterializer, RunMaterializeError,
};
use crate::canonical_origin::{canonical_origin_from_effective_web_url, effective_web_url};
use crate::error::ApiError;
use crate::git_checkout::GitRepoCache;
use crate::github_webhooks::{
WEBHOOK_ROUTE, WEBHOOK_SECRET_ENV, parse_event_metadata, verify_signature,
};
use crate::jwt_auth::{self, AuthMode};
use crate::principal_middleware::{
AuthContextSlot, RequestAuth, RequestAuthContext, RequireRunBlob, RequireRunManagementTarget,
RequireRunScoped, RequireRunStageScoped, RequireStageArtifact, RequireWorkerRunScoped,
RequiredUser, principal_middleware,
};
use crate::request_id::{self, RequestId};
use crate::run_files::{FilesInFlight, new_files_in_flight};
use crate::server_secrets::{LlmClientResult, ServerSecrets};
use crate::spawn_env::apply_render_graph_env;
use crate::worker_control::{LocalWorkerControlBus, WorkerControlBus, WorkerControlBusError};
use crate::worker_runtime::{
LocalWorkerRuntime, WorkerExit, WorkerLaunchSpec, WorkerRef, WorkerRuntime,
};
use crate::worker_token::{WorkerScopeSet, WorkerTokenKeys, issue_worker_token_with_scopes};
use crate::{
canonical_host, demo, diagnostics, run_manifest, security_headers, static_files, web_auth,
};
mod automation_scheduler;
mod handler;
mod pull_request_supervisor;
pub(crate) mod resource_sampler;
mod session_runtime;
pub(crate) use automation_scheduler::spawn_automation_scheduler;
pub(crate) use handler::events::EventListParams;
#[cfg(test)]
pub(in crate::server) use handler::events::filtered_global_events;
pub(crate) use handler::graph::render_graph_bytes;
#[cfg(test)]
pub(in crate::server) use handler::graph::{
RenderSubprocessError, render_dot_subprocess, render_graph_bytes_with_exe_override,
};
#[cfg(test)]
pub(in crate::server) use handler::system::validate_github_slug;
pub(crate) use pull_request_supervisor::spawn_pull_request_creation_supervisor;
use session_runtime::SessionRuntimeManager;
pub(crate) type EnvLookup = Arc<dyn Fn(&str) -> Option<String> + Send + Sync>;
pub fn default_page_limit() -> u32 {
20
}
#[derive(serde::Deserialize)]
pub struct PaginationParams {
#[serde(rename = "page[limit]", default = "default_page_limit")]
pub limit: u32,
#[serde(rename = "page[offset]", default)]
pub offset: u32,
}
pub(crate) fn clamp_page_limit(limit: u32) -> u32 {
limit.clamp(1, 100)
}
pub(crate) fn clamp_page_offset(offset: u32) -> u32 {
offset.min(MAX_PAGE_OFFSET)
}
pub(crate) fn paginate_items<T>(items: Vec<T>, pagination: &PaginationParams) -> (Vec<T>, bool) {
let limit = clamp_page_limit(pagination.limit) as usize;
let offset = clamp_page_offset(pagination.offset) as usize;
let mut data: Vec<_> = items.into_iter().skip(offset).take(limit + 1).collect();
let has_more = data.len() > limit;
data.truncate(limit);
(data, has_more)
}
#[derive(serde::Deserialize)]
pub(crate) struct DfParams {
#[serde(default)]
pub(crate) verbose: bool,
}
/// List response envelope with pagination metadata.
#[derive(serde::Serialize)]
pub struct ListResponse<T: serde::Serialize> {
data: T,
meta: PaginationMeta,
}
impl<T: serde::Serialize> ListResponse<T> {
/// Non-paginated response with `has_more: false`.
pub fn new(data: T) -> Self {
Self {
data,
meta: PaginationMeta {
has_more: false,
total: None,
},
}
}
pub fn paginated(data: T, has_more: bool, total: u64) -> Self {
Self {
data,
meta: PaginationMeta {
has_more,
total: i64::try_from(total).ok(),
},
}
}
}
/// Snapshot of a managed run.
struct ManagedRun {
dot_source: String,
status: RunStatus,
error: Option<String>,
created_at: chrono::DateTime<chrono::Utc>,
// Populated when running:
answer_transport: Option<RunAnswerTransport>,
accepted_questions: HashSet<String>,
/// Stage IDs of currently steerable live agent sessions, keyed to the
/// session id that owns the active lease. Used by the steerability
/// predicate for steer/interrupt controls.
active_steerable_stages: HashMap<StageId, String>,
/// API-mode session targets eligible for live pair control. ACP sessions
/// can be steerable but are intentionally excluded from pairing.
active_api_targets: HashMap<StageId, PairTarget>,
/// Stage IDs of currently running agent sessions that have no live
/// steering capability, keyed to the session id that owns the marker.
active_non_steerable_stages: HashMap<StageId, String>,
event_tx: Option<broadcast::Sender<RunEvent>>,
checkpoint: Option<Checkpoint>,
cancel_tx: Option<oneshot::Sender<()>>,
cancel_token: Option<CancellationToken>,
worker_ref: Option<WorkerRef>,
/// Exact worker currently covered by a cancellation escalation task.
/// Prevents repeated cancel requests from arming duplicate watchdogs.
cancel_escalation_worker: Option<WorkerRef>,
run_dir: Option<std::path::PathBuf>,
execution_mode: RunExecutionMode,
}
impl ManagedRun {
/// True if cancellation should still escalate to `worker_ref`; clears a
/// stale escalation marker as a side effect.
fn escalation_still_current(&mut self, worker_ref: &WorkerRef) -> bool {
let matches_watchdog = self.cancel_escalation_worker.as_ref() == Some(worker_ref);
let still_current = matches_watchdog
&& !self.status.is_terminal()
&& self.worker_ref.as_ref() == Some(worker_ref);
if matches_watchdog && !still_current {
self.cancel_escalation_worker = None;
}
still_current
}
/// Clears the escalation marker if it is still owned by `worker_ref`.
fn clear_escalation_for(&mut self, worker_ref: &WorkerRef) {
if self.cancel_escalation_worker.as_ref() == Some(worker_ref) {
self.cancel_escalation_worker = None;
}
}
}
#[derive(Clone, Copy)]
enum RunExecutionMode {
Start,
Resume,
}
enum ExecutionResult {
Completed(Box<Result<operations::Started, WorkflowError>>),
CancelledBySignal,
}
const WORKER_CANCEL_GRACE: Duration = Duration::from_secs(5);
const TERMINAL_DELETE_WORKER_GRACE: Duration = Duration::from_millis(50);
const WORKER_CONTROL_ENQUEUE_TIMEOUT: Duration = Duration::from_secs(1);
/// Per-model billing totals.
#[derive(Default)]
struct ModelBillingTotals {
stages: i64,
billing: BilledTokenCounts,
}
/// In-memory aggregate billing counters, reset on server restart.
#[derive(Default)]
struct BillingAccumulator {
total_runs: i64,
total_timing: fabro_types::RunTiming,
by_model: HashMap<ModelRef, ModelBillingTotals>,
}
pub(crate) type RegistryFactoryOverride =
dyn Fn(Arc<dyn Interviewer>) -> HandlerRegistry + Send + Sync;
#[derive(Clone)]
enum RunAnswerTransport {
Worker {
run_id: RunId,
bus: Arc<dyn WorkerControlBus>,
},
InProcess {
interviewer: Arc<ControlInterviewer>,
steering_hub: Arc<fabro_workflow::SteeringHub>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AnswerTransportError {
Closed,
Timeout,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PairTransportError {
Closed,
Timeout,
Control(fabro_workflow::PairControlError),
}
impl RunAnswerTransport {
async fn publish_worker_control(
run_id: RunId,
bus: &Arc<dyn WorkerControlBus>,
message: WorkerControlEnvelope,
) -> Result<(), WorkerControlBusError> {
timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, bus.publish(run_id, message))
.await
.map_err(|_| WorkerControlBusError::PublishTimeout)?
.map(|_| ())
}
fn answer_error_from_bus(error: &WorkerControlBusError) -> AnswerTransportError {
match error {
WorkerControlBusError::PublishTimeout => AnswerTransportError::Timeout,
WorkerControlBusError::Closed
| WorkerControlBusError::Unavailable
| WorkerControlBusError::InvalidCursor { .. } => AnswerTransportError::Closed,
}
}
fn pair_error_from_bus(error: &WorkerControlBusError) -> PairTransportError {
match error {
WorkerControlBusError::PublishTimeout => PairTransportError::Timeout,
WorkerControlBusError::Closed
| WorkerControlBusError::Unavailable
| WorkerControlBusError::InvalidCursor { .. } => PairTransportError::Closed,
}
}
async fn submit(
&self,
qid: &str,
submission: AnswerSubmission,
) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::interview_answer(qid.to_string(), submission);
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { interviewer, .. } => interviewer
.submit(qid, submission)
.await
.map_err(|_| AnswerTransportError::Closed),
}
}
async fn cancel_run(&self) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::cancel_run();
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { interviewer, .. } => {
interviewer.cancel_all().await;
Ok(())
}
}
}
/// Forward a steer to the worker (subprocess) or directly into the
/// in-process steering hub.
async fn steer(&self, text: String, actor: Principal) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::steer(text, actor);
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { steering_hub, .. } => {
steering_hub.deliver_steer(text, Some(actor));
Ok(())
}
}
}
async fn interrupt(&self, actor: Principal) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::interrupt(actor);
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { steering_hub, .. } => {
steering_hub.interrupt(Some(&actor));
Ok(())
}
}
}
async fn interrupt_then_steer(
&self,
text: String,
actor: Principal,
) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::interrupt_then_steer(text, actor);
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { steering_hub, .. } => {
steering_hub.interrupt_then_steer(&text, Some(&actor));
Ok(())
}
}
}
async fn start_pair(
&self,
run_id: RunId,
pair_id: PairId,
target: PairTarget,
actor: Principal,
) -> Result<(), PairTransportError> {
match self {
Self::Worker {
run_id: worker_run_id,
bus,
} => {
let message = WorkerControlEnvelope::start_pair(run_id, pair_id, target, actor);
Self::publish_worker_control(*worker_run_id, bus, message)
.await
.map_err(|err| Self::pair_error_from_bus(&err))
}
Self::InProcess { steering_hub, .. } => steering_hub
.start_pair(run_id, pair_id, target, Some(actor))
.map(|_| ())
.map_err(PairTransportError::Control),
}
}
async fn send_pair_message(
&self,
pair_id: PairId,
message_id: PairMessageId,
text: String,
client_message_id: Option<String>,
actor: Principal,
) -> Result<(), PairTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::pair_message(
pair_id,
message_id,
text.clone(),
client_message_id.clone(),
actor,
);
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::pair_error_from_bus(&err))
}
Self::InProcess { steering_hub, .. } => steering_hub
.send_pair_message(pair_id, message_id, text, client_message_id, Some(actor))
.map(|_| ())
.map_err(PairTransportError::Control),
}
}
async fn end_pair(&self, pair_id: PairId, actor: Principal) -> Result<(), PairTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::end_pair(pair_id, actor);
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::pair_error_from_bus(&err))
}
Self::InProcess { steering_hub, .. } => steering_hub
.end_pair(pair_id, Some(actor))
.map(|_| ())
.map_err(PairTransportError::Control),
}
}
async fn pause_run(&self) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::pause_run();
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { .. } => Err(AnswerTransportError::Closed),
}
}
async fn unpause_run(&self) -> Result<(), AnswerTransportError> {
match self {
Self::Worker { run_id, bus } => {
let message = WorkerControlEnvelope::unpause_run();
Self::publish_worker_control(*run_id, bus, message)
.await
.map_err(|err| Self::answer_error_from_bus(&err))
}
Self::InProcess { .. } => Err(AnswerTransportError::Closed),
}
}
}
#[derive(Debug, Clone)]
struct LoadedPendingInterview {
run_id: RunId,
qid: String,
question: InterviewQuestionRecord,
}
#[derive(Debug, Clone)]
struct SlackLifecycleDetails {
kind: slack_blocks::RunLifecycleKind,
started_event_name: Option<String>,
result: Option<String>,
duration_ms: Option<u64>,
}
#[derive(Debug, Clone, Default)]
struct PriorSlackLifecycleEventDetails {
started_event_name: Option<String>,
pull_request: Option<SlackLifecyclePullRequest>,
}
#[derive(Debug, Clone)]
struct SlackLifecyclePullRequest {
number: u64,
title: Option<String>,
url: Option<String>,
}
#[derive(Debug, Clone)]
struct SlackConnectionRuntimeState {
status: IntegrationConnectionState,
last_connected_at: Option<DateTime<Utc>>,
last_error: Option<String>,
}
impl Default for SlackConnectionRuntimeState {
fn default() -> Self {
Self {
status: IntegrationConnectionState::Connecting,
last_connected_at: None,
last_error: None,
}
}
}
fn sanitize_integration_error(error: &str) -> String {
const MAX_ERROR_CHARS: usize = 240;
let sanitized = error.replace(['\r', '\n'], " ");
sanitized.chars().take(MAX_ERROR_CHARS).collect()
}
#[derive(Clone)]
struct SlackService {
client: SlackClient,
app_token: String,
default_channel: Option<String>,
posted_messages: Arc<Mutex<HashMap<(RunId, String), SlackPostedMessage>>>,
thread_registry: Arc<ThreadRegistry>,
connection: Arc<Mutex<SlackConnectionRuntimeState>>,
}
impl SlackService {
fn new(bot_token: String, app_token: String, default_channel: Option<String>) -> Self {
Self {
client: SlackClient::new(bot_token),
app_token,
default_channel,
posted_messages: Arc::new(Mutex::new(HashMap::new())),
thread_registry: Arc::new(ThreadRegistry::new()),
connection: Arc::new(Mutex::new(SlackConnectionRuntimeState::default())),
}
}
fn connection_status(&self) -> IntegrationConnectionStatus {
let state = self
.connection
.lock()
.expect("slack connection state lock poisoned")
.clone();
IntegrationConnectionStatus {
kind: IntegrationConnectionKind::SocketMode,
status: state.status,
last_connected_at: state.last_connected_at,
last_error: state.last_error,
}
}
fn status_sink(&self) -> slack_connection::ConnectionStatusSink {
let connection = Arc::clone(&self.connection);
Arc::new(move |update| {
let mut state = connection
.lock()
.expect("slack connection state lock poisoned");
match update {
slack_connection::ConnectionStatusUpdate::Connecting => {
state.status = IntegrationConnectionState::Connecting;
state.last_error = None;
}
slack_connection::ConnectionStatusUpdate::Connected => {
state.status = IntegrationConnectionState::Connected;
state.last_connected_at = Some(Utc::now());
state.last_error = None;
}
slack_connection::ConnectionStatusUpdate::Error(error) => {
state.status = IntegrationConnectionState::Error;
state.last_error = Some(sanitize_integration_error(&error));
}
}
})
}
async fn handle_event(
&self,
state: &AppState,
envelope: &EventEnvelope,
run_web_url: Option<&str>,
) {
let event = &envelope.event;
match &event.body {
EventBody::InterviewStarted(props) => {
if props.question_id.is_empty() {
return;
}
let Some(default_channel) = self.default_channel.as_deref() else {
return;
};
let key = (event.run_id, props.question_id.clone());
if self
.posted_messages
.lock()
.expect("slack posted messages lock poisoned")
.contains_key(&key)
{
return;
}
let question = runtime_question_from_interview_record(&InterviewQuestionRecord {
id: props.question_id.clone(),
text: props.question.clone(),
stage: props.stage.clone(),
question_type: props.question_type.parse().unwrap_or_default(),
options: props.options.clone(),
allow_freeform: props.allow_freeform,
timeout_seconds: props.timeout_seconds,
context_display: props.context_display.clone(),
review_target: props.review_target.clone(),
});
let blocks = slack_blocks::question_to_blocks(
&event.run_id.to_string(),
&props.question_id,
&question,
run_web_url,
);
if let Ok(posted) = self
.client
.post_message(default_channel, &blocks, None)
.await
{
if question.allow_freeform || question.question_type == QuestionType::Freeform {
self.thread_registry.register(
&posted.ts,
&event.run_id.to_string(),
&props.question_id,
);
}
self.posted_messages
.lock()
.expect("slack posted messages lock poisoned")
.insert(key, posted);
}
}
EventBody::InterviewCompleted(props) => {
self.finish_interview(
event.run_id,
&props.question_id,
&props.question,
&props.answer,
)
.await;
}
EventBody::InterviewTimeout(props) => {
self.finish_interview(
event.run_id,
&props.question_id,
&props.question,
"Timed out",
)
.await;
}
EventBody::InterviewInterrupted(props) => {
self.finish_interview(
event.run_id,
&props.question_id,
&props.question,
"Interrupted",
)
.await;
}
EventBody::RunStarted(_) | EventBody::RunCompleted(_) | EventBody::RunFailed(_) => {
self.handle_lifecycle_event(state, envelope, run_web_url)
.await;
}
_ => {}
}
}
async fn handle_lifecycle_event(
&self,
state: &AppState,
envelope: &EventEnvelope,
run_web_url: Option<&str>,
) {
let event = &envelope.event;
let Some(details) = slack_lifecycle_details(event) else {
return;
};
let event_name = event.body.event_name();
let projection = match state.stores.runs.load_run_projection(&event.run_id).await {
Ok(Some(projection)) => projection,
Ok(None) => {
warn!(
run_id = %event.run_id,
event = event_name,
"Skipping Slack lifecycle notification because run projection is missing"
);
return;
}
Err(err) => {
warn!(
run_id = %event.run_id,
event = event_name,
error = %err,
"Skipping Slack lifecycle notification because run projection could not be loaded"
);
return;
}
};
// Filter routes first; bail out before any further work if none match.
let mut routes: Vec<_> = projection
.spec
.settings
.run
.notifications
.iter()
.filter(|(_, route)| {
route.enabled
&& route.provider.as_deref() == Some("slack")
&& route.events.iter().any(|event| event == event_name)
})
.collect();
if routes.is_empty() {
return;
}
routes.sort_by_key(|(route_name, _)| *route_name);
// Only completed/failed events need to recover prior PR details (a
// run.started event cannot have a prior PullRequestCreated).
let prior = if matches!(details.kind, slack_blocks::RunLifecycleKind::Started) {
PriorSlackLifecycleEventDetails::default()
} else {
load_prior_slack_lifecycle_event_details(state, event.run_id, envelope.seq).await
};
let workflow_label = slack_lifecycle_workflow_label(
projection.as_ref(),
details
.started_event_name
.as_deref()
.or(prior.started_event_name.as_deref()),
event_name,
);
let pull_request = prior.pull_request.or_else(|| {
projection
.pull_request
.as_ref()
.map(slack_lifecycle_pull_request_from_link)
});
let run_id = event.run_id.to_string();
let run_url = run_web_url.or(projection.web_url.as_deref());
let pull_request_blocks =
pull_request
.as_ref()
.map(|pull_request| slack_blocks::RunLifecyclePullRequest {
number: pull_request.number,
title: pull_request.title.as_deref(),
url: pull_request.url.as_deref(),
});
let blocks =
slack_blocks::run_lifecycle_blocks(details.kind, &slack_blocks::RunLifecycleBlocks {
run_id: &run_id,
run_url,
workflow_label: &workflow_label,
result: details.result.as_deref(),
duration_ms: details.duration_ms,
pull_request: pull_request_blocks,
});
let blocks = &blocks;
let posts = routes.into_iter().filter_map(|(route_name, route)| {
let channel =
resolve_slack_lifecycle_route_channel(event.run_id, route_name, route, event_name)?;
Some(async move {
if let Err(err) = self.client.post_message(&channel, blocks, None).await {
warn!(
run_id = %event.run_id,
event = event_name,
notification_route = route_name.as_str(),
error = %err,
"Failed to post Slack lifecycle notification"
);
}
})
});
join_all(posts).await;
}
async fn finish_interview(
&self,
run_id: RunId,
qid: &str,
question_text: &str,
answer_text: &str,
) {
let key = (run_id, qid.to_string());
let posted = self
.posted_messages
.lock()
.expect("slack posted messages lock poisoned")
.remove(&key);
let Some(posted) = posted else {
return;
};
self.thread_registry.remove(&posted.ts);
let blocks = slack_blocks::answered_blocks(question_text, answer_text);
let _ = self
.client
.update_message(&posted.channel_id, &posted.ts, &blocks)
.await;
}
async fn submit_answer(&self, state: Arc<AppState>, submission: SlackAnswerSubmission) {
let Ok(run_id) = RunId::from_str(&submission.run_id) else {
return;
};
let Ok(pending) = load_pending_interview(state.as_ref(), run_id, &submission.qid).await
else {
return;
};
let answer_submission = AnswerSubmission::new(submission.answer, submission.actor);
let _ = submit_pending_interview_answer(state.as_ref(), &pending, answer_submission).await;
}
}
fn slack_lifecycle_details(event: &RunEvent) -> Option<SlackLifecycleDetails> {
match &event.body {
EventBody::RunStarted(props) => Some(SlackLifecycleDetails {
kind: slack_blocks::RunLifecycleKind::Started,
started_event_name: Some(props.name.clone()),
result: None,
duration_ms: None,
}),
EventBody::RunCompleted(props) => Some(SlackLifecycleDetails {
kind: slack_blocks::RunLifecycleKind::Completed,
started_event_name: None,
result: Some(slack_lifecycle_completed_result(
&props.status,
props.reason,
)),
duration_ms: Some(props.timing.wall_time_ms),
}),
EventBody::RunFailed(props) => Some(SlackLifecycleDetails {
kind: slack_blocks::RunLifecycleKind::Failed,
started_event_name: None,
result: Some(slack_lifecycle_failed_result(&props.failure)),
duration_ms: Some(props.timing.wall_time_ms),
}),
_ => None,
}
}
fn slack_lifecycle_completed_result(status: &str, reason: SuccessReason) -> String {
let status = status.trim();
let reason = reason.to_string();
if status.is_empty() || status == reason {
reason
} else {
format!("{status} — {reason}")
}
}
fn slack_lifecycle_failed_result(failure: &fabro_types::RunFailure) -> String {
let reason = failure.reason.to_string();
let message = failure.detail.message.trim();
if message.is_empty() {
reason
} else {
format!("{reason} — {message}")
}
}
async fn load_prior_slack_lifecycle_event_details(
state: &AppState,
run_id: RunId,
before_seq: u32,
) -> PriorSlackLifecycleEventDetails {
let run_store = match state.stores.runs.open_run_reader(&run_id).await {
Ok(run_store) => run_store,
Err(err) => {
warn!(
run_id = %run_id,
error = %err,
"Unable to inspect prior run events for Slack lifecycle notification"
);
return PriorSlackLifecycleEventDetails::default();
}
};
let events = match run_store.list_events().await {
Ok(events) => events,
Err(err) => {
warn!(
run_id = %run_id,
error = %err,
"Unable to load prior run events for Slack lifecycle notification"
);
return PriorSlackLifecycleEventDetails::default();
}
};
let mut details = PriorSlackLifecycleEventDetails::default();
for envelope in events {
if envelope.seq >= before_seq {
break;
}
match envelope.event.body {
EventBody::RunStarted(props) if !props.name.trim().is_empty() => {
details.started_event_name = Some(props.name);
}
EventBody::PullRequestCreated(props) => {
details.pull_request = Some(SlackLifecyclePullRequest {
number: props.pr_number,
title: Some(props.title),
url: Some(props.pr_url),
});
}
_ => {}
}
}
details
}
fn slack_lifecycle_workflow_label(
projection: &fabro_store::RunProjection,
started_event_name: Option<&str>,
event_name: &str,
) -> String {
[
projection.spec.workflow_name(),
projection.spec.workflow_slug(),
projection.spec.graph_name(),
started_event_name,
]
.into_iter()
.flatten()
.map(str::trim)
.find(|value| !value.is_empty())
.unwrap_or(event_name)
.to_string()
}
fn slack_lifecycle_pull_request_from_link(link: &PullRequestLink) -> SlackLifecyclePullRequest {
SlackLifecyclePullRequest {
number: link.number,
title: None,
url: Some(link.html_url()),
}
}
fn resolve_slack_lifecycle_route_channel(
run_id: RunId,
route_name: &str,
route: &NotificationRouteSettings,
event_name: &str,
) -> Option<String> {
let Some(channel) = route
.slack
.as_ref()
.and_then(|slack| slack.channel.as_ref())
else {
warn!(
run_id = %run_id,
notification_route = route_name,
event = event_name,
"Skipping Slack lifecycle notification route without channel"
);
return None;
};
// `{{ vars.* }}` is substituted at run creation, so the channel is literal
// here; anything still unresolved skips the route rather than sending to a
// half-rendered channel name.
let resolved = match channel.resolve_with(&mut fabro_types::settings::ResolveCtx::new()) {
Ok(resolved) => resolved,
Err(err) => {
warn!(
run_id = %run_id,
notification_route = route_name,
event = event_name,
error = %err,
"Skipping Slack lifecycle notification route with unresolved channel"
);
return None;
}
};
if resolved.trim().is_empty() {
warn!(
run_id = %run_id,
notification_route = route_name,
event = event_name,
"Skipping Slack lifecycle notification route with empty channel"
);
return None;
}
Some(resolved)
}
/// Shared application state for the server.
pub struct AppState {
runs: Mutex<HashMap<RunId, ManagedRun>>,
aggregate_billing: Mutex<BillingAccumulator>,
pub(crate) stores: AppStores,
session_runtimes: SessionRuntimeManager,
artifact_store: ArtifactStore,
automation_repo_cache: Arc<GitRepoCache>,
#[cfg(any(test, feature = "test-support"))]
automation_materializer_override: Option<Arc<dyn AutomationRunMaterializer>>,
worker_tokens: WorkerTokenKeys,
started_at: Instant,
resource_sampler: resource_sampler::ResourceSampler,
max_concurrent_runs: usize,
pub(crate) worker_control_bus: Arc<dyn WorkerControlBus>,
pub(crate) worker_runtime: Arc<dyn WorkerRuntime>,
scheduler_notify: Notify,
automation_scheduler_notify: Notify,
pull_request_scheduler_notify: Notify,
pull_request_creation_queue: Mutex<pull_request_supervisor::PendingPullRequestCreationQueue>,
global_event_tx: broadcast::Sender<EventEnvelope>,
/// Per-run coalescing registry for `GET /runs/{id}/files`. Concurrent
/// callers for the same run share one materialization; different runs
/// proceed in parallel. See `crate::run_files` for semantics.
pub(crate) files_in_flight: FilesInFlight,
pull_request_create_locks: KeyedMutex<RunId>,
parent_link_lock: AsyncMutex<()>,
pub(super) server_secrets: ServerSecrets,
pub(crate) llm_source: Arc<dyn CredentialSource>,
manifest_run_defaults: RwLock<Arc<RunLayer>>,
manifest_run_settings: RwLock<std::result::Result<RunNamespace, SharedError>>,
pub(crate) server_settings: RwLock<Arc<ServerSettings>>,
effective_web_url: RwLock<String>,
catalog: RwLock<Arc<Catalog>>,
pub(crate) env_lookup: EnvLookup,
pub(crate) github_api_base_url: String,
active_config_path: PathBuf,
http_client: Option<fabro_http::HttpClient>,
sandbox_provider_registry: SandboxProviderRegistry,
shutdown: CancellationToken,
shutting_down: AtomicBool,
registry_factory_override: Option<Box<RegistryFactoryOverride>>,
slack_service: Option<Arc<SlackService>>,
slack_started: AtomicBool,
github_webhook_secret: Option<String>,
}
pub(crate) struct AppStores {
pub(crate) runs: Arc<Database>,
pub(crate) run_summaries: Arc<RunSummaryStore>,
pub(crate) auth_codes: Arc<AuthCodeStore>,
pub(crate) auth_sessions: Arc<AuthSessionStore>,
pub(crate) automations: Arc<AutomationStore>,
pub(crate) environments: Arc<EnvironmentStore>,
pub(crate) mcp_servers: Arc<McpServerStore>,
pub(crate) vault: Arc<SecretStore>,
pub(crate) variables: Arc<VariableStore>,
}
#[cfg(any(test, feature = "test-support"))]
impl AppState {
/// Access the auth session store so tests can seed CLI sessions against
/// the same SQLite pool the router reads from.
#[must_use]
pub fn test_auth_session_store(&self) -> &Arc<AuthSessionStore> {
&self.stores.auth_sessions
}
/// Access the auth-code store used by this router.
#[must_use]
pub fn test_auth_code_store(&self) -> &Arc<AuthCodeStore> {
&self.stores.auth_codes
}
}
impl AppState {
pub(crate) fn automation_store(&self) -> &AutomationStore {
&self.stores.automations
}
pub(crate) fn environment_store(&self) -> &EnvironmentStore {
&self.stores.environments
}
pub(crate) fn mcp_server_store(&self) -> &McpServerStore {
&self.stores.mcp_servers
}
pub(crate) async fn materialize_automation_run(
&self,
input: AutomationRunMaterializeInput,
) -> Result<AutomationRunMaterialized, RunMaterializeError> {
#[cfg(any(test, feature = "test-support"))]
if let Some(materializer) = self.automation_materializer_override.as_ref() {
return materializer.materialize(input).await;
}
let settings = self.server_settings();
let credentials = self
.github_credentials(&settings.server.integrations.github)
.await
.map_err(|source| RunMaterializeError::LoadCredentials { source })?;
ProductionAutomationRunMaterializer::new(
credentials,
self.github_api_base_url.clone(),
self.http_client.clone(),
Arc::clone(&self.automation_repo_cache),
fabro_workflow_version::WorkflowVersionStore::new(self.store_ref().blobs()),
)
.materialize(input)
.await
}
pub(crate) fn notify_automation_scheduler(&self) {
self.automation_scheduler_notify.notify_one();
}
pub(crate) fn automation_scheduler_notified(
&self,
) -> impl std::future::Future<Output = ()> + '_ {
self.automation_scheduler_notify.notified()
}
pub(crate) fn notify_pull_request_scheduler(&self) {
self.pull_request_scheduler_notify.notify_one();
}
pub(crate) fn pull_request_scheduler_notified(
&self,
) -> impl std::future::Future<Output = ()> + '_ {
self.pull_request_scheduler_notify.notified()
}
}
pub(crate) struct AskFabroReadiness {
default_model: Option<String>,
}
impl AskFabroReadiness {
pub(crate) fn decorate(&self, mut run: fabro_types::Run) -> fabro_types::Run {
run.ask_fabro = self.ask_fabro_for(&run);
run
}
fn ask_fabro_for(&self, run: &fabro_types::Run) -> AskFabro {
let unavailable_reason = if run.sandbox.is_none() {
Some(AskFabroUnavailableReason::NoSandbox)
} else if run
.sandbox
.as_ref()
.and_then(fabro_types::RunSandbox::instance)
.is_none()
{
Some(AskFabroUnavailableReason::SandboxNotReady)
} else if self.default_model.is_none() {
Some(AskFabroUnavailableReason::LlmUnconfigured)
} else {
None
};
AskFabro {
available: unavailable_reason.is_none(),
unavailable_reason,
default_model: self.default_model.clone(),
}
}
}
pub(crate) struct AppStateConfig {
pub(crate) resolved_settings: ResolvedAppStateSettings,
pub(crate) registry_factory_override: Option<Box<RegistryFactoryOverride>>,
pub(crate) max_concurrent_runs: usize,
pub(crate) store: Arc<Database>,
pub(crate) artifact_store: ArtifactStore,
pub(crate) db_pool: DbPool,
pub(crate) preloaded_vault: Vault,
pub(crate) server_secrets: ServerSecrets,
pub(crate) env_lookup: EnvLookup,
pub(crate) github_api_base_url: Option<String>,
pub(crate) active_config_path: PathBuf,
pub(crate) http_client: Option<fabro_http::HttpClient>,
pub(crate) sandbox_provider_registry: Option<SandboxProviderRegistry>,
pub(crate) shutdown: CancellationToken,
#[cfg(test)]
pub(crate) worker_control_bus: Option<Arc<dyn WorkerControlBus>>,
#[cfg(test)]
pub(crate) worker_runtime: Option<Arc<dyn WorkerRuntime>>,
#[cfg(any(test, feature = "test-support"))]
pub(crate) automation_materializer_override: Option<Arc<dyn AutomationRunMaterializer>>,
}
#[derive(Clone)]
pub(crate) struct ResolvedAppStateSettings {
pub(crate) server_settings: ServerSettings,
pub(crate) manifest_run_defaults: RunLayer,
pub(crate) llm_catalog_settings: LlmCatalogSettings,
}
fn accumulate_billing_rollup(
accumulator: &mut BillingAccumulator,
rollup: &fabro_workflow::ProjectionBillingRollup,
) {
accumulator.total_runs += 1;
accumulator.total_timing = accumulator.total_timing.saturating_add(&rollup.timing);
for model in &rollup.by_model {
let entry = accumulator.by_model.entry(model.model.clone()).or_default();
entry.stages += model.stages;
entry.billing.add_counts(&model.billing);
}
}
impl AppState {
pub(crate) fn manifest_run_defaults(&self) -> Arc<RunLayer> {
Arc::clone(
&self
.manifest_run_defaults
.read()
.expect("manifest run defaults lock poisoned"),
)
}
pub(crate) fn server_settings(&self) -> Arc<ServerSettings> {
Arc::clone(
&self
.server_settings
.read()
.expect("server settings lock poisoned"),
)
}
pub(crate) fn catalog(&self) -> Arc<Catalog> {
Arc::clone(&self.catalog.read().expect("catalog lock poisoned"))
}
pub(crate) fn active_config_path(&self) -> &std::path::Path {
&self.active_config_path
}
pub(crate) fn manifest_run_settings(&self) -> std::result::Result<RunNamespace, SharedError> {
self.manifest_run_settings
.read()
.expect("manifest run settings lock poisoned")
.clone()
}
pub(crate) fn refresh_manifest_run_settings_from_catalogs(&self) {
let manifest_run_defaults = self.manifest_run_defaults();
let manifest_run_settings = resolve_manifest_run_settings_with_catalog(
manifest_run_defaults.as_ref(),
&self.stores.environments,
&self.stores.mcp_servers,
);
*self
.manifest_run_settings
.write()
.expect("manifest run settings lock poisoned") = manifest_run_settings;
}
pub(crate) fn refresh_manifest_run_settings_from_environment_catalog(&self) {
self.refresh_manifest_run_settings_from_catalogs();
}
fn http_client(&self) -> Result<fabro_http::HttpClient, fabro_http::HttpClientBuildError> {
match &self.http_client {
Some(client) => Ok(client.clone()),
None => fabro_http::http_client(),
}
}
pub(crate) fn server_storage_dir(&self) -> PathBuf {
PathBuf::from(&self.server_settings().server.storage.root)
}
/// Scratch directory used by the automation materializer when staging
/// per-run manifests. Shared by API-triggered and scheduled fires.
pub(crate) fn automation_temp_root(&self) -> PathBuf {
Storage::new(self.server_storage_dir())
.scratch_dir()
.join("automations")
}
/// Snapshotted at create-time so attach replays surface the same link
/// even if `server.web.url` is later changed. `None` when the UI is
/// turned off or `server.web.url` is unset/invalid.
pub(crate) fn run_web_url(&self, run_id: &fabro_types::RunId) -> Option<String> {
if !self.server_settings().server.web.enabled {
return None;
}
let base = self.canonical_origin().ok()?;
Some(format!("{}/runs/{run_id}", base.trim_end_matches('/')))
}
pub(crate) async fn resolve_llm_client(&self) -> anyhow::Result<LlmClientResult> {
resolve_llm_client_from_source(self.llm_source.as_ref(), self.catalog()).await
}
pub(crate) async fn configured_llm_provider_ids(&self) -> Vec<ProviderId> {
let catalog = self.catalog();
self.llm_source.configured_providers(catalog.as_ref()).await
}
/// Resolve the LLM client once and derive the ready provider IDs from it,
/// logging a warning when resolution fails. Callers that need both values
/// must use this instead of `ready_llm_provider_ids` so the client is not
/// resolved twice.
pub(crate) async fn resolve_llm_client_with_ready_ids(
&self,
) -> (anyhow::Result<LlmClientResult>, Vec<ProviderId>) {
let llm_result = self.resolve_llm_client().await;
if let Err(err) = &llm_result {
warn!(error = ?err, "Failed to resolve LLM client while checking ready providers");
}
let ready_provider_ids = llm_result
.as_ref()
.map(LlmClientResult::provider_ids)
.unwrap_or_default();
(llm_result, ready_provider_ids)
}
pub(crate) async fn ready_llm_provider_ids(&self) -> Vec<ProviderId> {
self.resolve_llm_client_with_ready_ids().await.1
}
pub(crate) async fn decorate_run_summary(&self, run: fabro_types::Run) -> fabro_types::Run {
self.ask_fabro_readiness().await.decorate(run)
}
pub(crate) async fn decorate_run_summaries(
&self,
runs: Vec<fabro_types::Run>,
) -> Vec<fabro_types::Run> {
let readiness = self.ask_fabro_readiness().await;
runs.into_iter()
.map(|run| readiness.decorate(run))
.collect()
}
pub(crate) async fn ask_fabro_readiness(&self) -> AskFabroReadiness {
let provider_ids = self.ready_llm_provider_ids().await;
let default_model = if provider_ids.is_empty() {
None
} else {
Some(
self.catalog()
.default_for_configured_ids(&provider_ids)
.id
.to_string(),
)
};
AskFabroReadiness { default_model }
}
pub(crate) async fn vault_secret(
&self,
name: &str,
) -> Result<Option<String>, SecretStoreError> {
self.stores
.vault
.get(name)
.await
.map(|entry| entry.map(|entry| entry.value))
}
pub(crate) fn config_env_lookup(&self, name: &str) -> Option<String> {
(self.env_lookup)(name)
}
pub(crate) async fn check_daytona_api_key(
&self,
api_key: String,
) -> anyhow::Result<daytona::DaytonaKeyCheck> {
self.check_daytona_api_key_with_timeout(api_key, daytona::DAYTONA_CREDENTIAL_PROBE_TIMEOUT)
.await
}
pub(crate) async fn check_daytona_api_key_with_timeout(
&self,
api_key: String,
probe_timeout: Duration,
) -> anyhow::Result<daytona::DaytonaKeyCheck> {
let base_url = self
.config_env_lookup(EnvVars::DAYTONA_API_URL)
.or_else(|| self.config_env_lookup(EnvVars::DAYTONA_SERVER_URL))
.unwrap_or_else(|| daytona::DEFAULT_DAYTONA_API_URL.to_string());
let org_id = self.config_env_lookup(EnvVars::DAYTONA_ORGANIZATION_ID);
let http_client = fabro_http::http_client().context("failed to build HTTP client")?;
daytona::check_daytona_api_key_with_timeout(
&base_url,
org_id.as_deref(),
api_key,
http_client,
probe_timeout,
)
.await
}
/// Borrow the persistent store so sibling modules can open run readers
/// without cross-module state coupling on the `AppState` field layout.
pub(crate) fn store_ref(&self) -> &Arc<Database> {
&self.stores.runs
}
/// Loads the current projection for `run_id`, with the standard HTTP error
/// mapping: storage failures become 500s and a missing run becomes the
/// canonical 404.
pub(crate) async fn load_run_projection(
&self,
run_id: &RunId,
) -> Result<Arc<fabro_store::RunProjection>, ApiError> {
self.stores
.runs
.load_run_projection(run_id)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?
.ok_or_else(|| ApiError::not_found("Run not found."))
}
pub(crate) fn session_runtimes(&self) -> &SessionRuntimeManager {
&self.session_runtimes
}
pub(crate) fn sandbox_provider_registry(&self) -> &SandboxProviderRegistry {
&self.sandbox_provider_registry
}
pub(crate) fn server_secret(&self, name: &str) -> Option<String> {
self.server_secrets.get(name)
}
pub(crate) fn worker_token_keys(&self) -> &WorkerTokenKeys {
&self.worker_tokens
}
/// Loopback target this server is bound to, derived from the runtime
/// daemon record. Used by in-process Ask Fabro sessions to call the local
/// API over the normal HTTP path (authed with a same-run worker token).
pub(crate) fn self_server_target(&self) -> anyhow::Result<fabro_client::ServerTarget> {
let storage_dir = self.server_storage_dir();
let runtime_directory = Storage::new(&storage_dir).runtime_directory();
let daemon = ServerDaemon::read(&runtime_directory)?.with_context(|| {
format!(
"server record {} is missing",
runtime_directory.record_path().display()
)
})?;
// `Bind::to_target()` already produces the http(s)-URL-or-absolute-
// socket-path form that `ServerTarget`'s FromStr understands.
daemon.bind.to_target().parse()
}
pub(crate) fn effective_web_url(&self) -> String {
self.effective_web_url
.read()
.expect("effective web url lock poisoned")
.clone()
}
pub(crate) fn canonical_origin(&self) -> Result<String, String> {
canonical_origin_from_effective_web_url(&self.effective_web_url())
}
pub(crate) fn session_key(&self) -> Option<Key> {
self.server_secret(EnvVars::SESSION_SECRET)
.and_then(|value| auth::derive_cookie_key(value.as_bytes()).ok())
}
pub(crate) async fn github_credentials(
&self,
settings: &GithubIntegrationSettings,
) -> anyhow::Result<Option<fabro_github::GitHubCredentials>> {
match settings.strategy {
GithubIntegrationStrategy::App => {
let Some(app_id) = settings.app_id.clone() else {
return Ok(None);
};
let raw = self
.vault_secret(EnvVars::GITHUB_APP_PRIVATE_KEY)
.await
.map_err(anyhow::Error::new)?;
let Some(raw) = raw else {
return Ok(None);
};
let private_key_pem = decode_secret_pem(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw)
.map_err(anyhow::Error::msg)?;
Ok(Some(fabro_github::GitHubCredentials::App(
fabro_github::GitHubAppCredentials {
app_id,
private_key_pem,
slug: settings.slug.clone(),
},
)))
}
GithubIntegrationStrategy::Token => {
let token = self
.vault_secret(EnvVars::GITHUB_TOKEN)
.await
.map_err(anyhow::Error::new)?
.as_deref()
.map(str::trim)
.filter(|token| !token.is_empty())
.map(str::to_string);
match token {
Some(token) => {
fabro_github::validate_static_github_token(&token)
.map_err(anyhow::Error::msg)?;
Ok(Some(fabro_github::GitHubCredentials::Pat(token)))
}
None => anyhow::bail!(
"GITHUB_TOKEN not configured -- run fabro install or run fabro secret set GITHUB_TOKEN"
),
}
}
}
}
fn begin_shutdown(&self) {
self.shutting_down.store(true, Ordering::Relaxed);
self.scheduler_notify.notify_waiters();
self.automation_scheduler_notify.notify_waiters();
self.pull_request_scheduler_notify.notify_waiters();
}
pub(crate) fn shutdown_token(&self) -> CancellationToken {
self.shutdown.clone()
}
fn is_shutting_down(&self) -> bool {
self.shutting_down.load(Ordering::Relaxed)
}
pub(crate) fn replace_runtime_settings(
&self,
resolved_settings: ResolvedAppStateSettings,
) -> anyhow::Result<()> {
let ResolvedAppStateSettings {
server_settings,
manifest_run_defaults,
llm_catalog_settings,
} = resolved_settings;
let server_settings = Arc::new(server_settings);
let manifest_run_defaults = Arc::new(manifest_run_defaults);
let effective_web_url =
effective_web_url(&server_settings.server, |name| (self.env_lookup)(name));
let manifest_run_settings = resolve_manifest_run_settings_with_catalog(
manifest_run_defaults.as_ref(),
&self.stores.environments,
&self.stores.mcp_servers,
);
let catalog = Arc::new(
Catalog::from_builtin_with_overrides(&llm_catalog_settings)
.context("building LLM model catalog")?,
);
canonical_origin_from_effective_web_url(&effective_web_url).map_err(anyhow::Error::msg)?;
*self
.manifest_run_defaults
.write()
.expect("manifest run defaults lock poisoned") = manifest_run_defaults;
*self
.manifest_run_settings
.write()
.expect("manifest run settings lock poisoned") = manifest_run_settings;
*self
.server_settings
.write()
.expect("server settings lock poisoned") = server_settings;
*self
.effective_web_url
.write()
.expect("effective web url lock poisoned") = effective_web_url;
*self.catalog.write().expect("catalog lock poisoned") = catalog;
Ok(())
}
}
async fn resolve_llm_client_from_source(
source: &dyn CredentialSource,
catalog: Arc<Catalog>,
) -> anyhow::Result<LlmClientResult> {
let resolved = source
.resolve(catalog.as_ref())
.await
.context("resolving LLM credentials")?;
let report = LlmClient::from_credentials_report(resolved.credentials, catalog).await;
Ok(LlmClientResult {
client: report.client,
auth_issues: resolved.auth_issues,
registration_issues: report.registration_issues,
})
}
fn decode_secret_pem(name: &str, raw: &str) -> Result<String, String> {
if raw.starts_with("-----") {
return Ok(raw.to_string());
}
let pem_bytes = BASE64_STANDARD
.decode(raw)
.map_err(|err| format!("{name} is not valid PEM or base64: {err}"))?;
String::from_utf8(pem_bytes)
.map_err(|err| format!("{name} base64 decoded to invalid UTF-8: {err}"))
}
fn start_optional_slack_service(state: &Arc<AppState>) {
let Some(service) = state.slack_service.clone() else {
return;
};
if state.slack_started.swap(true, Ordering::SeqCst) {
return;
}
let event_state = Arc::clone(state);
let event_service = Arc::clone(&service);
tokio::spawn(async move {
let mut rx = event_state.global_event_tx.subscribe();
loop {
match rx.recv().await {
Ok(envelope) => {
// Resolve the run's web URL once per event so the Slack
// message can deep-link back to Fabro. Returns None when
// the web UI is disabled or `server.web.url` is unset, in
// which case `question_to_blocks` simply omits the link.
let run_web_url = event_state.run_web_url(&envelope.event.run_id);
event_service
.handle_event(event_state.as_ref(), &envelope, run_web_url.as_deref())
.await;
}
Err(RecvError::Lagged(_)) => {}
Err(RecvError::Closed) => break,
}
}
});
let socket_state = Arc::clone(state);
tokio::spawn(async move {
let submit_service = Arc::clone(&service);
let on_submit: Arc<dyn Fn(SlackAnswerSubmission) + Send + Sync> =
Arc::new(move |submission| {
let state = Arc::clone(&socket_state);
let service = Arc::clone(&submit_service);
tokio::spawn(async move {
service.submit_answer(state, submission).await;
});
});
slack_connection::run_with_status(
&service.client,
&service.app_token,
&service.thread_registry,
on_submit,
service.status_sink(),
)
.await;
});
}
/// Build the axum Router with all run endpoints and embedded static assets.
#[allow(
clippy::needless_pass_by_value,
reason = "Public router helper keeps the existing ergonomic API and forwards by reference."
)]
pub fn build_router(state: Arc<AppState>, auth_mode: AuthMode) -> Router {
build_router_with_options(state, &auth_mode, RouterOptions::default())
}
#[derive(Clone, Debug)]
pub struct RouterOptions {
pub web_enabled: bool,
pub static_asset_root: Option<PathBuf>,
pub github_endpoints: Option<Arc<GithubEndpoints>>,
/// Set when serving with the `--watch-web` dev flag. The static-file
/// handler then refuses to fall back to the embedded SPA snapshot and
/// returns a 503 "build in progress" page on miss, so developers see
/// their edits or a clear signal — never stale embedded bytes.
pub watch_web: bool,
}
impl Default for RouterOptions {
fn default() -> Self {
Self {
web_enabled: true,
static_asset_root: None,
github_endpoints: None,
watch_web: false,
}
}
}
fn removed_web_route(path: &str) -> bool {
matches!(path, "/setup/complete") || path.starts_with("/install")
}
/// Build the axum Router with configurable web surface routing.
pub fn build_router_with_options(
state: Arc<AppState>,
auth_mode: &AuthMode,
options: RouterOptions,
) -> Router {
start_optional_slack_service(&state);
let RouterOptions {
web_enabled,
static_asset_root,
github_endpoints,
watch_web,
} = options;
let translation_state = Arc::clone(&state);
let state_for_canonical_host = Arc::clone(&state);
let github_endpoints =
github_endpoints.unwrap_or_else(|| Arc::new(GithubEndpoints::production_defaults()));
let webhook_secret = state.github_webhook_secret.clone();
let principal_layer = middleware::from_fn_with_state(Arc::clone(&state), principal_middleware);
let api_common = if web_enabled {
Router::new()
.route("/openapi.json", get(handler::openapi_spec))
.merge(web_auth::api_routes())
} else {
Router::new().route("/openapi.json", get(handler::openapi_spec))
};
let demo_router = Router::new()
.nest(
"/api/v1",
api_common
.clone()
.merge(handler::demo_routes())
.layer(principal_layer.clone()),
)
.layer(axum::Extension(auth_mode.clone()))
.layer(axum::Extension(Arc::clone(&github_endpoints)))
.with_state(state.clone());
let mut real_router = Router::new().nest(
"/api/v1",
api_common
.merge(handler::real_routes())
.layer(principal_layer),
);
if web_enabled {
real_router = real_router.nest("/auth", web_auth::routes().merge(auth::web_routes()));
}
let real_router = real_router
.layer(axum::Extension(github_endpoints))
.with_state(state);
let dispatch = service_fn(move |req: axum_extract::Request| {
let demo = demo_router.clone();
let real = real_router.clone();
async move {
let demo_active = web_enabled
&& req.uri().path().starts_with("/api/")
&& req.headers().get("x-fabro-demo").is_some_and(|v| v == "1");
if demo_active {
demo.oneshot(req).await
} else {
real.oneshot(req).await
}
}
});
let mut app_router = Router::new()
.route("/health", get(handler::health))
.fallback_service(service_fn(move |req: axum_extract::Request| {
let dispatch = dispatch.clone();
let static_asset_root = static_asset_root.clone();
async move {
let path = req.uri().path().to_string();
let dispatch_path = path.starts_with("/api/")
|| path == "/health"
|| (web_enabled && path.starts_with("/auth/"));
if dispatch_path {
dispatch.oneshot(req).await
} else if web_enabled && removed_web_route(&path) {
Ok::<_, std::convert::Infallible>(StatusCode::NOT_FOUND.into_response())
} else if web_enabled && matches!(req.method(), &Method::GET | &Method::HEAD) {
let headers = req.headers().clone();
Ok::<_, std::convert::Infallible>(
static_files::serve_with_asset_root(
&path,
&headers,
static_asset_root.as_deref(),
watch_web,
)
.await,
)
} else {
Ok::<_, std::convert::Infallible>(StatusCode::NOT_FOUND.into_response())
}
}
}));
app_router = app_router.layer(middleware::from_fn_with_state(
translation_state,
auth_translation_middleware,
));
app_router = app_router.layer(middleware::from_fn(demo_routing_middleware));
app_router = app_router.layer(axum::Extension(auth_mode.clone()));
let mut router = app_router;
if let Some(secret) = webhook_secret {
let secret: Arc<[u8]> = Arc::from(secret.into_bytes().into_boxed_slice());
router = github_webhook_routes(secret).merge(router);
}
router
// Innermost of the outer layers so every response body — static SPA
// assets and JSON API alike — is compressed before the header/log
// middlewares see it.
.layer(compression_layer())
.layer(middleware::from_fn_with_state(
canonical_host::Config {
state: state_for_canonical_host,
web_enabled,
},
canonical_host::redirect_middleware,
))
.layer(middleware::from_fn(security_headers::layer))
.layer(middleware::from_fn(http_log_middleware))
.layer(middleware::from_fn(request_id::layer))
}
/// Response-compression layer shared by the main and install-mode routers.
///
/// The default predicate skips streaming SSE (`text/event-stream`), gRPC,
/// images, ZIP archives, and tiny bodies. The quality is pinned because
/// tower-http's default defers to each codec's own default, and brotli's is
/// quality 11 — seconds of CPU on a multi-megabyte asset. Level 4 keeps both
/// codecs fast at a near-optimal ratio.
pub(crate) fn compression_layer() -> CompressionLayer<impl Predicate> {
CompressionLayer::new()
.quality(CompressionLevel::Precise(4))
.compress_when(DefaultPredicate::new().and(NotForContentType::const_new("application/zip")))
}
async fn http_log_middleware(mut req: axum_extract::Request, next: Next) -> Response {
let path = req.uri().path();
if path.starts_with("/assets/") || path.starts_with("/images/") {
return next.run(req).await;
}
let method = req.method().clone();
let path = req.uri().path().to_string();
let request_id = req
.extensions()
.get::<RequestId>()
.copied()
.map(RequestId::render)
.unwrap_or_default();
let auth_slot = AuthContextSlot::initial();
req.extensions_mut().insert(auth_slot.clone());
let start = std::time::Instant::now();
let response = next.run(req).await;
let status = response.status().as_u16();
let latency_ms = start.elapsed().as_millis();
let auth_context = auth_slot.log_snapshot();
let principal_kind = auth_context
.principal
.as_ref()
.map_or("none", Principal::kind);
let auth_status = auth_context.auth_status.as_str();
macro_rules! emit_http_log {
($level:ident $(, $field:ident = $value:expr)* $(,)?) => {{
if let Some(auth_error_code) = auth_context.auth_error_code {
let auth_error_code = auth_error_code.as_str();
$level!(
%method,
%path,
status,
latency_ms,
request_id = %request_id,
principal_kind,
auth_status,
auth_error_code,
$($field = $value,)*
"HTTP response"
);
} else {
$level!(
%method,
%path,
status,
latency_ms,
request_id = %request_id,
principal_kind,
auth_status,
$($field = $value,)*
"HTTP response"
);
}
}};
}
macro_rules! emit_principal_http_log {
($level:ident) => {{
match &auth_context.principal {
Some(Principal::User(user)) => emit_http_log!(
$level,
user_auth_method = user.auth_method.as_str(),
idp_issuer = user.identity.issuer(),
idp_subject = user.identity.subject(),
login = user.login.as_str(),
),
Some(Principal::Worker { run_id }) => {
emit_http_log!($level, run_id = run_id.to_string().as_str(),)
}
Some(Principal::Webhook { delivery_id }) => {
emit_http_log!($level, delivery_id = delivery_id.as_str(),)
}
Some(Principal::Slack {
team_id, user_id, ..
}) => emit_http_log!(
$level,
team_id = team_id.as_str(),
user_id = user_id.as_str(),
),
None | Some(Principal::Agent { .. } | Principal::System { .. }) => {
emit_http_log!($level)
}
}
}};
}
if status >= 500 {
emit_principal_http_log!(error);
} else {
emit_principal_http_log!(info);
}
response
}
fn github_webhook_routes(secret: Arc<[u8]>) -> Router {
Router::new()
.route(WEBHOOK_ROUTE, post(github_webhook))
.with_state(secret)
}
async fn github_webhook(
State(secret): State<Arc<[u8]>>,
RequestAuth(auth_slot): RequestAuth,
headers: HeaderMap,
body: Bytes,
) -> StatusCode {
let delivery_id = headers
.get("x-github-delivery")
.and_then(|value| value.to_str().ok())
.unwrap_or("unknown");
let Some(signature) = headers
.get("x-hub-signature-256")
.and_then(|value| value.to_str().ok())
else {
auth_slot.replace(RequestAuthContext::invalid());
warn!(delivery = %delivery_id, "Webhook missing X-Hub-Signature-256 header");
return StatusCode::UNAUTHORIZED;
};
if !verify_signature(&secret, &body, signature) {
auth_slot.replace(RequestAuthContext::invalid());
warn!(delivery = %delivery_id, "Webhook HMAC signature mismatch");
return StatusCode::UNAUTHORIZED;
}
auth_slot.replace(RequestAuthContext::authenticated(
Principal::Webhook {
delivery_id: delivery_id.to_string(),
},
None,
));
let event_type = headers
.get("x-github-event")
.and_then(|value| value.to_str().ok())
.unwrap_or("unknown");
if tracing::enabled!(tracing::Level::DEBUG) {
let (repo, action) = parse_event_metadata(&body);
debug!(
event = %event_type,
delivery = %delivery_id,
repo = %repo,
action = %action,
"Webhook received"
);
} else {
info!(
event = %event_type,
delivery = %delivery_id,
"Webhook received"
);
}
StatusCode::OK
}
struct PrunePlan {
run_ids: Vec<RunId>,
rows: Vec<PruneRunEntry>,
total_size_bytes: u64,
}
#[expect(
clippy::disallowed_methods,
reason = "sync helper invoked from async handler via spawn_blocking (see callers at :1301 / :1341)"
)]
fn build_disk_usage_response(
summaries: &[fabro_types::Run],
storage_dir: &std::path::Path,
verbose: bool,
) -> anyhow::Result<DiskUsageResponse> {
let scratch_base_dir = scratch_base(storage_dir);
let logs_base_dir = Storage::new(storage_dir).runtime_directory().logs_dir();
let runs = scan_runs_with_summaries(summaries, &scratch_base_dir)?;
let mut active_count = 0u64;
let mut total_run_size = 0u64;
let mut reclaimable_run_size = 0u64;
let mut run_rows = Vec::new();
for run in &runs {
let size = dir_size(&run.path);
total_run_size += size;
if run.status().is_active() {
active_count += 1;
} else {
reclaimable_run_size += size;
}
if verbose {
run_rows.push(DiskUsageRunRow {
run_id: Some(run.run_id().to_string()),
workflow_name: Some(run.workflow_display_name()),
status: Some(run.status().to_string()),
start_time: Some(run.start_time()),
size_bytes: Some(to_i64(size)),
reclaimable: Some(!run.status().is_active()),
});
}
}
let mut log_count = 0u64;
let mut total_log_size = 0u64;
if let Ok(entries) = std::fs::read_dir(logs_base_dir) {
for entry in entries.flatten() {
let path = entry.path();
if !path.is_file() || path.extension().is_none_or(|ext| ext != "log") {
continue;
}
if let Ok(metadata) = path.metadata() {
log_count += 1;
total_log_size += metadata.len();
}
}
}
// Measure the whole storage tree so the managed total can't drift as new
// subdirectories are added. "other" is the residual (database, artifacts,
// sessions, vaults) — everything that isn't an enumerated run or log file.
let managed_size = dir_size(storage_dir);
let other_size = managed_size.saturating_sub(total_run_size + total_log_size);
Ok(DiskUsageResponse {
summary: vec![
DiskUsageSummaryRow {
type_: Some("runs".to_string()),
count: Some(to_i64(runs.len())),
active: Some(to_i64(active_count)),
size_bytes: Some(to_i64(total_run_size)),
reclaimable_bytes: Some(to_i64(reclaimable_run_size)),
},
DiskUsageSummaryRow {
type_: Some("logs".to_string()),
count: Some(to_i64(log_count)),
active: None,
size_bytes: Some(to_i64(total_log_size)),
reclaimable_bytes: Some(to_i64(total_log_size)),
},
DiskUsageSummaryRow {
type_: Some("other".to_string()),
count: None,
active: None,
size_bytes: Some(to_i64(other_size)),
reclaimable_bytes: Some(0),
},
],
total_size_bytes: Some(to_i64(managed_size)),
total_reclaimable_bytes: Some(to_i64(reclaimable_run_size + total_log_size)),
runs: verbose.then_some(run_rows),
})
}
fn build_prune_plan(
request: &PruneRunsRequest,
summaries: &[fabro_types::Run],
storage_dir: &std::path::Path,
) -> anyhow::Result<PrunePlan> {
let scratch_base_dir = scratch_base(storage_dir);
let runs = scan_runs_with_summaries(summaries, &scratch_base_dir)?;
let label_filters = request
.labels
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect::<Vec<_>>();
let mut filtered = filter_runs(
&runs,
request.before.as_deref(),
request.workflow.as_deref(),
&label_filters,
request.orphans,
StatusFilter::All,
);
let has_explicit_filters =
request.before.is_some() || request.workflow.is_some() || !label_filters.is_empty();
let staleness_threshold = if let Some(duration) = request.older_than.as_deref() {
Some(parse_system_duration(duration)?)
} else if !has_explicit_filters {
Some(chrono::Duration::hours(24))
} else {
None
};
if let Some(threshold) = staleness_threshold {
let cutoff = chrono::Utc::now() - threshold;
filtered.retain(|run| {
run.end_time
.or(run.start_time_dt)
.is_some_and(|time| time < cutoff)
});
}
filtered.retain(|run| !run.status().is_active());
let rows = filtered
.iter()
.map(|run| PruneRunEntry {
run_id: Some(run.run_id().to_string()),
dir_name: Some(run.dir_name.clone()),
workflow_name: Some(run.workflow_display_name()),
size_bytes: Some(to_i64(dir_size(&run.path))),
})
.collect::<Vec<_>>();
let total_size_bytes = rows
.iter()
.map(|row| row.size_bytes.unwrap_or_default())
.sum::<i64>()
.max(0)
.try_into()
.unwrap_or_default();
Ok(PrunePlan {
run_ids: filtered.iter().map(RunInfo::run_id).collect(),
rows,
total_size_bytes,
})
}
fn resolve_manifest_run_settings_with_catalog(
manifest_run_defaults: &RunLayer,
environment_store: &EnvironmentStore,
mcp_server_store: &McpServerStore,
) -> std::result::Result<RunNamespace, SharedError> {
WorkflowSettingsBuilder::new()
.server_manifest_defaults(
manifest_run_defaults.clone(),
(*environment_store.catalog_layer()).clone(),
)
.server_mcp_catalog(mcp_server_store.catalog_settings())
.build()
.map(|settings| settings.run)
.map_err(|err| SharedError::new(anyhow::Error::msg(err.to_string())))
}
fn system_sandbox_provider(
manifest_run_settings: &std::result::Result<RunNamespace, SharedError>,
) -> String {
manifest_run_settings.as_ref().map_or_else(
|_| SandboxProviderKind::default().to_string(),
|settings| settings.environment.provider.to_string(),
)
}
fn parse_system_duration(raw: &str) -> anyhow::Result<chrono::Duration> {
let raw = raw.trim();
anyhow::ensure!(!raw.is_empty(), "empty duration string");
let (num_str, unit) = raw.split_at(raw.len().saturating_sub(1));
let amount = num_str.parse::<u64>()?;
match unit {
"h" => Ok(chrono::Duration::hours(
i64::try_from(amount).unwrap_or(i64::MAX),
)),
"d" => Ok(chrono::Duration::days(
i64::try_from(amount).unwrap_or(i64::MAX),
)),
_ => anyhow::bail!("invalid duration unit '{unit}' in '{raw}' (expected 'h' or 'd')"),
}
}
fn dir_size(path: &std::path::Path) -> u64 {
walkdir::WalkDir::new(path)
.into_iter()
.filter_map(std::result::Result::ok)
.filter_map(|entry| entry.metadata().ok())
.filter(std::fs::Metadata::is_file)
.map(|metadata| metadata.len())
.sum()
}
fn to_i64<T>(value: T) -> i64
where
i64: TryFrom<T>,
{
i64::try_from(value).unwrap_or(i64::MAX)
}
fn worker_token_keys_from_server_secrets(
server_secrets: &ServerSecrets,
) -> anyhow::Result<WorkerTokenKeys> {
let session_secret = server_secrets
.get(EnvVars::SESSION_SECRET)
.ok_or_else(|| jwt_auth::session_secret_key_error(&auth::KeyDeriveError::Empty))?;
WorkerTokenKeys::from_master_secret(session_secret.as_bytes())
.map_err(|err| jwt_auth::session_secret_key_error(&err))
}
fn build_sandbox_provider_registry(
server_settings: &ServerSettings,
daytona_api_key: Option<String>,
env_lookup: &EnvLookup,
http_client: Option<fabro_http::HttpClient>,
) -> SandboxProviderRegistry {
let provider_settings = &server_settings.server.sandbox.providers;
let mut providers: Vec<Arc<dyn SandboxProvider>> = Vec::new();
if provider_settings.is_enabled(&SandboxProviderKind::LOCAL) {
providers.push(Arc::new(LocalSandboxProvider));
}
if provider_settings.is_enabled(&SandboxProviderKind::DOCKER) {
providers.push(Arc::new(DockerSandboxProvider::new()));
}
if provider_settings.is_enabled(&SandboxProviderKind::DAYTONA) && daytona_api_key.is_some() {
let api_url = env_lookup(EnvVars::DAYTONA_API_URL)
.or_else(|| env_lookup(EnvVars::DAYTONA_SERVER_URL));
let organization_id = env_lookup(EnvVars::DAYTONA_ORGANIZATION_ID);
providers.push(Arc::new(DaytonaSandboxProvider::new(
daytona_api_key,
api_url,
organization_id,
http_client,
)));
}
SandboxProviderRegistry::new(providers)
}
pub(crate) fn automation_dir_for_active_config(active_config_path: &std::path::Path) -> PathBuf {
active_config_path
.parent()
.unwrap_or_else(|| std::path::Path::new("."))
.join("automations")
}
fn mcp_server_dir_for_active_config(active_config_path: &std::path::Path) -> PathBuf {
active_config_path
.parent()
.unwrap_or_else(|| std::path::Path::new("."))
.join("mcps")
}
#[expect(
clippy::disallowed_methods,
reason = "synchronous app-state assembly may run inside an async runtime; a short-lived OS \
thread avoids nested Tokio runtimes"
)]
fn load_store_blocking<T, F, Fut>(description: &'static str, load: F) -> anyhow::Result<T>
where
T: Send + 'static,
F: FnOnce() -> Fut + Send + 'static,
Fut: std::future::Future<Output = anyhow::Result<T>>,
{
std::thread::spawn(move || {
let runtime = TokioRuntimeBuilder::new_current_thread()
.enable_all()
.build()
.with_context(|| format!("build {description} runtime"))?;
runtime.block_on(load())
})
.join()
.expect("store load thread should not panic")
}
pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppState>> {
let AppStateConfig {
resolved_settings,
registry_factory_override,
max_concurrent_runs,
store,
artifact_store,
db_pool,
preloaded_vault,
server_secrets,
env_lookup,
github_api_base_url,
active_config_path,
http_client,
sandbox_provider_registry,
shutdown,
#[cfg(test)]
worker_control_bus,
#[cfg(test)]
worker_runtime,
#[cfg(any(test, feature = "test-support"))]
automation_materializer_override,
} = config;
let automation_migration_pool = db_pool.clone();
load_store_blocking("automation environment migration", move || async move {
fabro_automation::backfill_environment_selectors(&automation_migration_pool)
.await
.map_err(anyhow::Error::new)
})
.context("backfill automation environment selectors")?;
let automation_store = Arc::new(AutomationStore::new(db_pool.clone()));
let local_provider_enabled = resolved_settings
.server_settings
.server
.sandbox
.providers
.is_enabled(&SandboxProviderKind::LOCAL);
let environment_pool = db_pool.clone();
let environment_store = Arc::new(
load_store_blocking("environment store", move || async move {
EnvironmentStore::load(environment_pool, local_provider_enabled)
.await
.map_err(anyhow::Error::new)
})
.context("load environments")?,
);
let run_summaries = store.run_summary_store();
let auth_codes = Arc::new(AuthCodeStore::new(db_pool.clone()));
let auth_sessions = Arc::new(AuthSessionStore::new(db_pool.clone()));
let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);
let mcp_server_pool = db_pool.clone();
let mcp_server_store = Arc::new(
load_store_blocking("MCP server store", move || async move {
McpServerStore::open(mcp_server_pool, mcp_server_dir)
.await
.map_err(anyhow::Error::new)
})
.context("load mcp servers")?,
);
let variables = Arc::new(VariableStore::new(db_pool.clone()));
let secret_store = Arc::new(SecretStore::new(db_pool));
let vault = preloaded_vault;
// Read vault secrets needed for synchronous setup before we wrap the vault in
// an async lock for the rest of AppState.
let daytona_api_key = vault.get(EnvVars::DAYTONA_API_KEY).map(str::to_string);
let llm_source: Arc<dyn CredentialSource> = Arc::new(SqlVaultCredentialSource::vault_only(
Arc::clone(&secret_store),
));
let (global_event_tx, _) = broadcast::channel(4096);
let current_server_settings = Arc::new(resolved_settings.server_settings);
let current_effective_web_url =
effective_web_url(&current_server_settings.server, |name| env_lookup(name));
let current_manifest_run_defaults = Arc::new(resolved_settings.manifest_run_defaults);
let current_manifest_run_settings = resolve_manifest_run_settings_with_catalog(
current_manifest_run_defaults.as_ref(),
&environment_store,
&mcp_server_store,
);
let current_catalog = Arc::new(
Catalog::from_builtin_with_overrides(&resolved_settings.llm_catalog_settings)
.context("building LLM model catalog")?,
);
let sandbox_provider_registry = sandbox_provider_registry.unwrap_or_else(|| {
build_sandbox_provider_registry(
current_server_settings.as_ref(),
daytona_api_key,
&env_lookup,
http_client.clone(),
)
});
let slack_service = {
let slack_settings = &current_server_settings.server.integrations.slack;
if slack_settings.enabled {
let default_channel = slack_settings.default_channel.clone();
match resolve_slack_credentials_status_with_lookup(|name| {
vault.get(name).map(str::to_string)
}) {
SlackCredentialResolution::Configured(credentials) => {
info!(
default_channel_configured = default_channel.is_some(),
"Slack integration enabled"
);
Some(Arc::new(SlackService::new(
credentials.bot_token,
credentials.app_token,
default_channel,
)))
}
SlackCredentialResolution::Missing { env_vars } => {
info!(
missing_env_vars = %env_vars.join(","),
"Slack integration disabled; missing credentials"
);
None
}
}
} else {
info!("Slack integration disabled by server configuration");
None
}
};
let worker_tokens = worker_token_keys_from_server_secrets(&server_secrets)?;
let github_api_base_url = github_api_base_url.unwrap_or_else(fabro_github::github_api_base_url);
let storage_root = PathBuf::from(&current_server_settings.server.storage.root);
let automation_repo_cache = Arc::new(GitRepoCache::new(
Storage::new(&storage_root)
.cache_dir()
.join("automation-repos"),
));
let worker_control_bus: Arc<dyn WorkerControlBus> = {
#[cfg(test)]
{
worker_control_bus.unwrap_or_else(|| Arc::new(LocalWorkerControlBus::new()))
}
#[cfg(not(test))]
{
Arc::new(LocalWorkerControlBus::new())
}
};
let worker_runtime: Arc<dyn WorkerRuntime> = {
#[cfg(test)]
{
worker_runtime.unwrap_or_else(|| Arc::new(LocalWorkerRuntime::new()))
}
#[cfg(not(test))]
{
Arc::new(LocalWorkerRuntime::new())
}
};
Ok(Arc::new(AppState {
runs: Mutex::new(HashMap::new()),
aggregate_billing: Mutex::new(BillingAccumulator::default()),
stores: AppStores {
runs: store,
run_summaries,
auth_codes,
auth_sessions,
automations: automation_store,
environments: environment_store,
mcp_servers: mcp_server_store,
vault: secret_store,
variables,
},
session_runtimes: SessionRuntimeManager::new(),
artifact_store,
automation_repo_cache,
#[cfg(any(test, feature = "test-support"))]
automation_materializer_override,
worker_tokens,
started_at: Instant::now(),
resource_sampler: resource_sampler::ResourceSampler::new(),
max_concurrent_runs,
worker_control_bus,
worker_runtime,
scheduler_notify: Notify::new(),
automation_scheduler_notify: Notify::new(),
pull_request_scheduler_notify: Notify::new(),
pull_request_creation_queue: Mutex::new(
pull_request_supervisor::PendingPullRequestCreationQueue::default(),
),
global_event_tx,
files_in_flight: new_files_in_flight(),
pull_request_create_locks: KeyedMutex::new(),
parent_link_lock: AsyncMutex::new(()),
server_secrets,
llm_source,
manifest_run_defaults: RwLock::new(current_manifest_run_defaults),
manifest_run_settings: RwLock::new(current_manifest_run_settings),
server_settings: RwLock::new(current_server_settings),
effective_web_url: RwLock::new(current_effective_web_url),
catalog: RwLock::new(current_catalog),
env_lookup: Arc::clone(&env_lookup),
github_api_base_url,
active_config_path,
http_client,
sandbox_provider_registry,
shutdown,
shutting_down: AtomicBool::new(false),
registry_factory_override,
slack_service,
slack_started: AtomicBool::new(false),
// Startup snapshot for the sync router build; rotating the webhook
// secret requires a server restart.
github_webhook_secret: vault.get(WEBHOOK_SECRET_ENV).map(str::to_string),
}))
}
const MAX_PAGE_OFFSET: u32 = 1_000_000;
enum DeleteRunOutcome {
Deleted,
AlreadyAbsent,
Preserved(DeleteRunResponse),
}
enum SandboxDeleteOutcome {
/// The durable run store did not exist; nothing to delete.
Absent,
/// The sandbox resource was cleaned up (or there was none to clean).
Cleaned,
/// Sandbox is being handed off to the operator instead of deleted.
Preserved(DeleteRunResponse),
}
async fn delete_run_internal(
state: &AppState,
id: RunId,
force: bool,
) -> Result<DeleteRunOutcome, ApiError> {
if !force {
reject_active_delete_without_force(state, &id).await?;
}
let mut managed_run = if let Ok(mut runs) = state.runs.lock() {
runs.remove(&id)
} else {
None
};
let had_managed_run = managed_run.is_some();
let durable_status = if managed_run.is_some() {
durable_run_status(state, id).await.ok().flatten()
} else {
None
};
let should_signal_cancel = !durable_status.is_some_and(RunStatus::is_terminal);
if let Some(managed_run) = managed_run.as_mut() {
if should_signal_cancel {
if let Some(token) = &managed_run.cancel_token {
token.cancel();
}
if let Some(answer_transport) = managed_run.answer_transport.clone() {
let _ = answer_transport.cancel_run().await;
}
if let Some(cancel_tx) = managed_run.cancel_tx.take() {
let _ = cancel_tx.send(());
}
}
// Terminal runs can still carry a stale worker ref briefly after their
// completion events land, so avoid paying the full cancellation grace.
let delete_grace = if should_signal_cancel && managed_run.status.requires_force_to_delete()
{
WORKER_CANCEL_GRACE
} else {
TERMINAL_DELETE_WORKER_GRACE
};
terminate_worker_for_deletion(
&state.worker_runtime,
managed_run.worker_ref.clone(),
delete_grace,
)
.await;
}
let delete_outcome = delete_run_sandbox_resource(state, id, force).await?;
if let Some(mut managed_run) = managed_run {
if let Some(run_dir) = managed_run.run_dir.take() {
remove_run_dir(&run_dir)
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
}
} else {
let storage = Storage::new(state.server_storage_dir());
let run_dir = storage.run_scratch(&id).root().to_path_buf();
remove_run_dir(&run_dir)
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
}
state
.stores
.runs
.delete_run(&id)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
state
.artifact_store
.delete_for_run(&id)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
match delete_outcome {
SandboxDeleteOutcome::Preserved(response) => Ok(DeleteRunOutcome::Preserved(response)),
SandboxDeleteOutcome::Cleaned => Ok(DeleteRunOutcome::Deleted),
SandboxDeleteOutcome::Absent if had_managed_run => Ok(DeleteRunOutcome::Deleted),
SandboxDeleteOutcome::Absent => Ok(DeleteRunOutcome::AlreadyAbsent),
}
}
async fn delete_run_sandbox_resource(
state: &AppState,
id: RunId,
force: bool,
) -> Result<SandboxDeleteOutcome, ApiError> {
let Ok(run_store) = state.stores.runs.open_run(&id).await else {
return Ok(SandboxDeleteOutcome::Absent);
};
let projection = match run_store.state().await {
Ok(projection) => projection,
Err(err) if force => {
tracing::warn!(
run_id = %id,
error = %render_with_causes(&err.to_string(), &collect_causes(&err)),
"Skipping sandbox provider delete because run projection cannot be loaded"
);
return Ok(SandboxDeleteOutcome::Cleaned);
}
Err(err) => {
return Err(ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
err.to_string(),
));
}
};
let delete_started = matches!(projection.status, RunStatus::Removing);
let can_mark_removing = projection.status.can_transition_to(RunStatus::Removing);
if !delete_started && can_mark_removing {
workflow_event::append_event(&run_store, &id, &workflow_event::Event::RunRemoving)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
}
let preserve = projection
.spec()
.settings
.run
.environment
.lifecycle
.preserve;
let Some(record) = projection
.sandbox
.as_ref()
.and_then(fabro_types::RunSandbox::instance)
.cloned()
else {
return Ok(SandboxDeleteOutcome::Cleaned);
};
let runtime = &record.runtime;
if preserve {
return Ok(SandboxDeleteOutcome::Preserved(DeleteRunResponse {
deleted: true,
sandbox_preserved: true,
sandbox: DeleteRunSandbox {
provider: record.provider,
id: runtime.id.clone(),
},
}));
}
let daytona_api_key = state
.vault_secret(EnvVars::DAYTONA_API_KEY)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
let sandbox = match reconnect_for_run(&record, daytona_api_key, Some(id)).await {
Ok(sandbox) => sandbox,
Err(err) if force || delete_started => {
tracing::warn!(
run_id = %id,
error = %render_with_causes(&err.to_string(), &collect_causes(err.as_ref())),
"Skipping sandbox provider delete during run deletion"
);
return Ok(SandboxDeleteOutcome::Cleaned);
}
Err(err) => {
let detail = render_with_causes(&err.to_string(), &collect_causes(err.as_ref()));
return Err(ApiError::new(StatusCode::CONFLICT, detail));
}
};
if let Err(err) = sandbox.delete().await {
if force || delete_started {
tracing::warn!(
run_id = %id,
error = %err.display_with_causes(),
"Skipping failed sandbox provider delete during run deletion"
);
return Ok(SandboxDeleteOutcome::Cleaned);
}
return Err(ApiError::new(
StatusCode::CONFLICT,
err.display_with_causes(),
));
}
Ok(SandboxDeleteOutcome::Cleaned)
}
async fn reject_active_delete_without_force(
state: &AppState,
run_id: &RunId,
) -> Result<(), ApiError> {
let managed_status = state
.runs
.lock()
.ok()
.and_then(|runs| runs.get(run_id).map(|managed_run| managed_run.status));
if let Some(status) = managed_status {
if status.requires_force_to_delete() {
return Err(ApiError::new(
StatusCode::CONFLICT,
active_run_delete_message(*run_id, status),
));
}
return Ok(());
}
match state.stores.run_summaries.get(run_id, Utc::now()).await {
Ok(Some(summary)) if summary.lifecycle.status.requires_force_to_delete() => {
Err(ApiError::new(
StatusCode::CONFLICT,
active_run_delete_message(*run_id, summary.lifecycle.status),
))
}
Ok(_) => Ok(()),
Err(err) => Err(ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
err.to_string(),
)),
}
}
fn active_run_delete_message(run_id: RunId, status: impl std::fmt::Display) -> String {
let run_id = run_id.to_string();
let short_run_id = &run_id[..12.min(run_id.len())];
format!(
"cannot remove active run {short_run_id} (status: {status}, use force=true or --force to force)"
)
}
async fn terminate_worker_for_deletion(
worker_runtime: &Arc<dyn WorkerRuntime>,
worker_ref: Option<WorkerRef>,
grace: Duration,
) {
let Some(worker_ref) = worker_ref else {
return;
};
worker_runtime.request_stop(&worker_ref).await;
let deadline = Instant::now() + grace;
while Instant::now() < deadline && worker_runtime.is_alive(&worker_ref).await {
sleep(Duration::from_millis(50)).await;
}
if worker_runtime.is_alive(&worker_ref).await {
worker_runtime.force_stop(&worker_ref).await;
let kill_deadline = Instant::now() + Duration::from_secs(1);
while Instant::now() < kill_deadline && worker_runtime.is_alive(&worker_ref).await {
sleep(Duration::from_millis(50)).await;
}
}
}
fn remove_run_dir(run_dir: &std::path::Path) -> std::io::Result<()> {
match std::fs::remove_dir_all(run_dir) {
Ok(()) => Ok(()),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(err),
}
}
#[cfg(test)]
fn compute_queue_positions(runs: &HashMap<RunId, ManagedRun>) -> HashMap<RunId, i64> {
let mut runnable: Vec<(&RunId, &ManagedRun)> = runs
.iter()
.filter(|(_, r)| r.status == RunStatus::Runnable)
.collect();
runnable.sort_by_key(|(_, r)| r.created_at);
runnable
.into_iter()
.enumerate()
.map(|(i, (id, _))| (*id, i64::try_from(i + 1).unwrap()))
.collect()
}
pub(in crate::server) fn counts_toward_scheduler_capacity(status: RunStatus) -> bool {
matches!(
status,
RunStatus::Starting
| RunStatus::Running
| RunStatus::Blocked { .. }
| RunStatus::Paused { .. }
)
}
#[allow(
clippy::result_large_err,
reason = "Run ID parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_run_id_path(id: &str) -> Result<RunId, Response> {
id.parse::<RunId>()
.map_err(|_| ApiError::bad_request("Invalid run ID.").into_response())
}
#[allow(
clippy::result_large_err,
reason = "Stage ID parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_stage_id_path(stage_id: &str) -> Result<StageId, Response> {
StageId::from_str(stage_id)
.map_err(|_| ApiError::bad_request("Invalid stage ID.").into_response())
}
#[allow(
clippy::result_large_err,
reason = "Blob hash parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_blob_hash_path(blob_hash: &str) -> Result<BlobHash, Response> {
BlobHash::from_str(blob_hash)
.map_err(|_| ApiError::bad_request("Invalid blob hash.").into_response())
}
#[allow(
clippy::result_large_err,
reason = "Missing query parameter validation returns HTTP 400 responses directly."
)]
fn required_query_param<T: Clone>(value: Option<&T>, name: &str) -> Result<T, Response> {
value.cloned().ok_or_else(|| {
ApiError::bad_request(format!("Missing {name} query parameter.")).into_response()
})
}
#[allow(
clippy::result_large_err,
reason = "Artifact path validation returns HTTP 400 responses directly."
)]
fn validate_relative_artifact_path(kind: &str, value: &str) -> Result<String, Response> {
if value.is_empty() {
return Err(ApiError::bad_request(format!("{kind} must not be empty")).into_response());
}
if value.contains('\\') {
return Err(
ApiError::bad_request(format!("{kind} must not contain backslashes")).into_response(),
);
}
let segments = value.split('/').collect::<Vec<_>>();
if segments.iter().any(|segment| segment.is_empty()) {
return Err(
ApiError::bad_request(format!("{kind} must not contain empty path segments"))
.into_response(),
);
}
if segments
.iter()
.any(|segment| matches!(*segment, "." | ".."))
{
return Err(ApiError::bad_request(format!(
"{kind} must be a relative path without '.' or '..' segments"
))
.into_response());
}
Ok(segments.join("/"))
}
fn bad_request_response(detail: impl Into<String>) -> Response {
ApiError::bad_request(detail.into()).into_response()
}
fn payload_too_large_response(detail: impl Into<String>) -> Response {
ApiError::new(StatusCode::PAYLOAD_TOO_LARGE, detail.into()).into_response()
}
fn octet_stream_response(bytes: Bytes) -> Response {
(
StatusCode::OK,
[("content-type", "application/octet-stream")],
bytes,
)
.into_response()
}
fn clear_live_run_state(run: &mut ManagedRun) {
run.answer_transport = None;
run.accepted_questions.clear();
run.active_api_targets.clear();
run.active_steerable_stages.clear();
run.active_non_steerable_stages.clear();
run.event_tx = None;
run.cancel_tx = None;
run.cancel_token = None;
run.worker_ref = None;
run.cancel_escalation_worker = None;
}
fn cleanup_worker_control_bus_for_run(state: &AppState, run_id: RunId) {
let bus = Arc::clone(&state.worker_control_bus);
tokio::spawn(async move {
bus.cleanup_run(run_id).await;
});
}
fn reconcile_live_interview_state_for_event(run: &mut ManagedRun, event: &RunEvent) {
match &event.body {
EventBody::InterviewCompleted(props) => {
run.accepted_questions.remove(&props.question_id);
}
EventBody::InterviewTimeout(props) => {
run.accepted_questions.remove(&props.question_id);
}
EventBody::InterviewInterrupted(props) => {
run.accepted_questions.remove(&props.question_id);
}
EventBody::RunCompleted(_) | EventBody::RunFailed(_) => {
run.accepted_questions.clear();
}
_ => {}
}
}
fn claim_run_answer_transport(
state: &AppState,
run_id: RunId,
qid: &str,
) -> Result<RunAnswerTransport, StatusCode> {
let mut runs = state.runs.lock().expect("runs lock poisoned");
let managed_run = runs.get_mut(&run_id).ok_or(StatusCode::NOT_FOUND)?;
let transport = managed_run
.answer_transport
.clone()
.ok_or(StatusCode::CONFLICT)?;
if !managed_run.accepted_questions.insert(qid.to_string()) {
return Err(StatusCode::CONFLICT);
}
Ok(transport)
}
fn release_run_answer_claim(state: &AppState, run_id: RunId, qid: &str) {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.accepted_questions.remove(qid);
}
}
#[derive(Clone)]
struct LiveWorkerProcess {
run_id: RunId,
worker_ref: WorkerRef,
}
/// Pick the terminal failure for a run that never produced its own terminal
/// event. A pending cancel wins over whatever failure the caller observed, so a
/// run that was cancelled while its worker was launching or dying is recorded
/// as cancelled rather than as broken.
fn failure_honoring_pending_cancel(
pending_control: Option<RunControlAction>,
otherwise: impl FnOnce() -> (WorkflowError, FailureReason),
) -> (WorkflowError, FailureReason) {
if pending_control == Some(RunControlAction::Cancel) {
(WorkflowError::Cancelled, FailureReason::Cancelled)
} else {
otherwise()
}
}
fn failure_for_incomplete_run(
pending_control: Option<RunControlAction>,
terminated_message: String,
) -> (WorkflowError, FailureReason) {
failure_honoring_pending_cancel(pending_control, || {
(
WorkflowError::engine(terminated_message),
FailureReason::Terminated,
)
})
}
pub(crate) async fn reconcile_incomplete_runs_on_startup(
state: &Arc<AppState>,
) -> anyhow::Result<usize> {
const RECONCILABLE_STATUSES: &[RunStatusKind] = &[
RunStatusKind::Runnable,
RunStatusKind::Starting,
RunStatusKind::Running,
RunStatusKind::Blocked,
RunStatusKind::Paused,
RunStatusKind::Removing,
];
let summaries = state
.stores
.run_summaries
.list_by_statuses(RECONCILABLE_STATUSES, chrono::Utc::now())
.await?;
let mut reconciled = 0usize;
for summary in summaries {
let run_store = state.stores.runs.open_run(&summary.id).await?;
let (error, reason) = failure_for_incomplete_run(
summary.lifecycle.pending_control,
"Fabro server restarted before the run reached a terminal state.".to_string(),
);
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&error,
fabro_types::RunTiming::default(),
reason,
None,
None,
None,
None,
);
workflow_event::append_event(&run_store, &summary.id, &failure_event).await?;
reconciled += 1;
}
Ok(reconciled)
}
fn live_worker_processes(state: &AppState) -> Vec<LiveWorkerProcess> {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.iter()
.filter_map(|(run_id, managed_run)| {
managed_run
.worker_ref
.clone()
.map(|worker_ref| LiveWorkerProcess {
run_id: *run_id,
worker_ref,
})
})
.collect()
}
async fn persist_shutdown_run_failures(
state: &Arc<AppState>,
workers: &[LiveWorkerProcess],
) -> anyhow::Result<()> {
let run_ids = workers
.iter()
.map(|worker| worker.run_id)
.collect::<HashSet<_>>();
for run_id in run_ids {
let run_store = state.stores.runs.open_run(&run_id).await?;
let run_state = run_store.state().await?;
if run_state.status.is_terminal() {
continue;
}
let (error, reason) = failure_for_incomplete_run(
run_state.pending_control,
"Fabro server shut down before the run reached a terminal state.".to_string(),
);
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&error,
fabro_types::RunTiming::default(),
reason,
None,
None,
None,
None,
);
workflow_event::append_event(&run_store, &run_id, &failure_event).await?;
}
Ok(())
}
pub(crate) async fn shutdown_active_workers(state: &Arc<AppState>) -> anyhow::Result<usize> {
shutdown_active_workers_with_grace(state, WORKER_CANCEL_GRACE, Duration::from_millis(50)).await
}
async fn shutdown_active_workers_with_grace(
state: &Arc<AppState>,
grace: Duration,
poll_interval: Duration,
) -> anyhow::Result<usize> {
state.begin_shutdown();
let workers = live_worker_processes(state.as_ref());
join_all(
workers
.iter()
.map(|worker| state.worker_runtime.request_stop(&worker.worker_ref)),
)
.await;
let survivors = poll_until_dead(state.as_ref(), &workers, grace, poll_interval).await;
if !survivors.is_empty() {
join_all(
survivors
.iter()
.map(|worker_ref| state.worker_runtime.force_stop(worker_ref)),
)
.await;
// Wait for the kernel to reap the killed workers so callers can
// assume the processes are actually gone when shutdown returns.
let kill_deadline = Instant::now() + Duration::from_secs(1);
while Instant::now() < kill_deadline
&& !alive_refs(state.as_ref(), &survivors).await.is_empty()
{
sleep(poll_interval).await;
}
}
persist_shutdown_run_failures(state, &workers).await?;
Ok(workers.len())
}
/// Poll until either the deadline expires or every worker is dead, returning
/// the set of workers still alive when polling stopped.
async fn poll_until_dead(
state: &AppState,
workers: &[LiveWorkerProcess],
grace: Duration,
poll_interval: Duration,
) -> Vec<WorkerRef> {
let refs: Vec<WorkerRef> = workers.iter().map(|w| w.worker_ref.clone()).collect();
let deadline = Instant::now() + grace;
loop {
let alive = alive_refs(state, &refs).await;
if alive.is_empty() || Instant::now() >= deadline {
return alive;
}
sleep(poll_interval).await;
}
}
async fn alive_refs(state: &AppState, refs: &[WorkerRef]) -> Vec<WorkerRef> {
let liveness = join_all(refs.iter().map(|r| state.worker_runtime.is_alive(r))).await;
refs.iter()
.zip(liveness)
.filter(|(_, alive)| *alive)
.map(|(r, _)| r.clone())
.collect()
}
async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow::Result<()> {
let run_store = state.stores.runs.open_run(&run_id).await?;
let run_state = run_store.state().await?;
if run_state.status.is_terminal() {
return Ok(());
}
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&WorkflowError::Cancelled,
fabro_types::RunTiming::default(),
FailureReason::Cancelled,
None,
None,
None,
None,
);
workflow_event::append_event(&run_store, &run_id, &failure_event).await
}
async fn finish_cancelled_run_before_execution(state: &Arc<AppState>, run_id: RunId) {
if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await {
error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status");
}
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.status = RunStatus::Failed {
reason: FailureReason::Cancelled,
};
clear_live_run_state(managed_run);
}
drop(runs);
cleanup_worker_control_bus_for_run(state.as_ref(), run_id);
state.scheduler_notify.notify_one();
}
/// Reject the run before execution if its effective sandbox provider is
/// disabled by server policy. Returns `true` when the run was rejected.
async fn reject_run_if_sandbox_provider_disabled(
state: &Arc<AppState>,
server_settings: &ServerSettings,
run_id: RunId,
settings: &RunNamespace,
) -> bool {
let provider = run_manifest::effective_sandbox_provider(settings);
let Some(error) = run_manifest::sandbox_provider_policy_error(server_settings, &provider)
else {
return false;
};
tracing::warn!(run_id = %run_id, error = %error, "Sandbox provider disabled by server policy");
fail_run_before_execution(state, run_id, FailureReason::LaunchFailed, error).await;
true
}
async fn fail_run_before_execution(
state: &Arc<AppState>,
run_id: RunId,
reason: FailureReason,
message: String,
) {
match state.stores.runs.open_run(&run_id).await {
Ok(run_store) => {
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&WorkflowError::engine(message.clone()),
fabro_types::RunTiming::default(),
reason,
None,
None,
None,
None,
);
if let Err(err) =
workflow_event::append_event(&run_store, &run_id, &failure_event).await
{
error!(run_id = %run_id, error = %err, "Failed to persist run failure status");
}
}
Err(err) => {
error!(run_id = %run_id, error = %err, "Failed to open run store while persisting run failure");
}
}
fail_managed_run(state, run_id, reason, message);
state.scheduler_notify.notify_one();
}
async fn forward_run_events_to_global(
state: Arc<AppState>,
run_id: RunId,
mut run_events: broadcast::Receiver<EventEnvelope>,
) {
loop {
match run_events.recv().await {
Ok(event) => {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
reconcile_live_interview_state_for_event(managed_run, &event.event);
}
let _ = state.global_event_tx.send(event);
}
Err(RecvError::Lagged(_)) => {}
Err(RecvError::Closed) => break,
}
}
}
fn managed_run(
dot_source: String,
status: RunStatus,
created_at: chrono::DateTime<chrono::Utc>,
run_dir: std::path::PathBuf,
execution_mode: RunExecutionMode,
) -> ManagedRun {
ManagedRun {
dot_source,
status,
error: None,
created_at,
answer_transport: None,
accepted_questions: HashSet::new(),
active_api_targets: HashMap::new(),
active_steerable_stages: HashMap::new(),
active_non_steerable_stages: HashMap::new(),
event_tx: None,
checkpoint: None,
cancel_tx: None,
cancel_token: None,
worker_ref: None,
cancel_escalation_worker: None,
run_dir: Some(run_dir),
execution_mode,
}
}
fn worker_mode_arg(mode: RunExecutionMode) -> &'static str {
match mode {
RunExecutionMode::Start => "start",
RunExecutionMode::Resume => "resume",
}
}
async fn load_pending_control(
state: &AppState,
run_id: RunId,
) -> anyhow::Result<Option<RunControlAction>> {
Ok(state
.stores
.run_summaries
.get(&run_id, Utc::now())
.await?
.and_then(|summary| summary.lifecycle.pending_control))
}
async fn durable_run_status(state: &AppState, run_id: RunId) -> anyhow::Result<Option<RunStatus>> {
Ok(state
.stores
.run_summaries
.get(&run_id, Utc::now())
.await?
.map(|summary| summary.lifecycle.status))
}
fn fail_managed_run(state: &Arc<AppState>, run_id: RunId, reason: FailureReason, message: String) {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.status = RunStatus::Failed { reason };
managed_run.error = Some(message);
clear_live_run_state(managed_run);
}
cleanup_worker_control_bus_for_run(state.as_ref(), run_id);
}
fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) {
use fabro_types::EventBody;
let mut runs = state.runs.lock().expect("runs lock poisoned");
let Some(managed_run) = runs.get_mut(&run_id) else {
return;
};
if matches!(&event.body, EventBody::RunRunnable(_)) {
// Scheduling is owned by the start/approve lifecycle handlers, which
// set the live status and notify the scheduler explicitly. Direct
// event ingestion still records durable history, but must not make
// externally injected events schedulable.
return;
}
match &event.body {
EventBody::RunSubmitted(_) => managed_run.status = RunStatus::Submitted,
EventBody::RunPending(props) => {
managed_run.status = RunStatus::Pending {
reason: props.reason,
};
}
EventBody::RunStarting(_) => managed_run.status = RunStatus::Starting,
EventBody::RunRunning(_) => managed_run.status = RunStatus::Running,
EventBody::RunBlocked(props) => {
managed_run.status = match managed_run.status {
RunStatus::Paused { .. } => RunStatus::Paused {
prior_block: Some(props.blocked_reason),
},
_ => RunStatus::Blocked {
blocked_reason: props.blocked_reason,
},
};
}
EventBody::RunUnblocked(_) => {
managed_run.status = match managed_run.status {
RunStatus::Paused {
prior_block: Some(_) | None,
} => RunStatus::Paused { prior_block: None },
_ => RunStatus::Running,
};
}
EventBody::RunPaused(_) => {
let prior_block = match managed_run.status {
RunStatus::Blocked { blocked_reason } => Some(blocked_reason),
RunStatus::Paused { prior_block } => prior_block,
_ => None,
};
managed_run.status = RunStatus::Paused { prior_block };
}
EventBody::RunUnpaused(_) => {
managed_run.status = match managed_run.status {
RunStatus::Paused {
prior_block: Some(blocked_reason),
} => RunStatus::Blocked { blocked_reason },
_ => RunStatus::Running,
};
}
EventBody::RunRemoving(_) => managed_run.status = RunStatus::Removing,
EventBody::RunCompleted(_) => {
let EventBody::RunCompleted(props) = &event.body else {
unreachable!(
"outer match arm already verified event.body is EventBody::RunCompleted"
)
};
managed_run.status = RunStatus::Succeeded {
reason: props.reason,
};
managed_run.error = None;
managed_run.active_api_targets.clear();
managed_run.active_steerable_stages.clear();
managed_run.active_non_steerable_stages.clear();
cleanup_worker_control_bus_for_run(state, run_id);
}
EventBody::RunFailed(props) => {
managed_run.status = RunStatus::Failed {
reason: props.failure.reason,
};
managed_run.error = Some(render_compact_with_causes(
&props.failure.detail.message,
&props.failure.detail.causes,
));
managed_run.active_api_targets.clear();
managed_run.active_steerable_stages.clear();
managed_run.active_non_steerable_stages.clear();
cleanup_worker_control_bus_for_run(state, run_id);
}
// Track active agent sessions by steerability. Activated/deactivated
// are leased by session id so stale deactivations cannot clear a newer
// binding for the same stage.
EventBody::AgentSessionActivated(props) => {
if let (Some(stage_id), Some(session_id)) =
(event.stage_id.as_ref(), event.session_id.as_ref())
{
if props.capabilities.contains(&SessionCapability::Steer) {
managed_run
.active_steerable_stages
.insert(stage_id.clone(), session_id.clone());
managed_run.active_non_steerable_stages.remove(stage_id);
let acp_provider: &'static str = AgentBackend::Acp.into();
if props.provider.as_deref() == Some(acp_provider) {
managed_run.active_api_targets.remove(stage_id);
} else {
managed_run
.active_api_targets
.insert(stage_id.clone(), PairTarget {
stage_id: stage_id.clone(),
node_label: event
.node_label
.clone()
.unwrap_or_else(|| stage_id.node_id().to_string()),
});
}
} else {
managed_run
.active_non_steerable_stages
.insert(stage_id.clone(), session_id.clone());
managed_run.active_steerable_stages.remove(stage_id);
managed_run.active_api_targets.remove(stage_id);
}
}
}
EventBody::AgentSessionDeactivated(_) => {
if let (Some(stage_id), Some(session_id)) =
(event.stage_id.as_ref(), event.session_id.as_ref())
{
if managed_run
.active_steerable_stages
.get(stage_id)
.is_some_and(|current| current == session_id)
{
managed_run.active_steerable_stages.remove(stage_id);
managed_run.active_api_targets.remove(stage_id);
}
if managed_run
.active_non_steerable_stages
.get(stage_id)
.is_some_and(|current| current == session_id)
{
managed_run.active_non_steerable_stages.remove(stage_id);
}
}
}
// ACP sessions are steerable via `agent.session.activated`; terminal
// ACP events and stage lifecycle events are still backstops for cleanup.
EventBody::AgentAcpCompleted(_)
| EventBody::AgentAcpCancelled(_)
| EventBody::AgentAcpTimedOut(_)
| EventBody::StageCompleted(_)
| EventBody::StageFailed(_) => {
if let Some(stage_id) = &event.stage_id {
managed_run.active_api_targets.remove(stage_id);
managed_run.active_steerable_stages.remove(stage_id);
managed_run.active_non_steerable_stages.remove(stage_id);
}
}
_ => {}
}
}
async fn drain_worker_stderr(
run_id: RunId,
stderr: std::pin::Pin<Box<dyn AsyncRead + Send + 'static>>,
) -> anyhow::Result<()> {
let mut lines = BufReader::new(stderr).lines();
while let Some(line) = lines.next_line().await? {
tracing::warn!(run_id = %run_id, "Worker stderr: {line}");
}
Ok(())
}
async fn fail_worker_launch(
state: &Arc<AppState>,
run_store: &fabro_store::RunDatabase,
run_id: RunId,
err: anyhow::Error,
) {
tracing::error!(run_id = %run_id, error = %err, "Failed to spawn worker");
let pending_control = match run_store.state().await {
Ok(run_state) => run_state.pending_control,
Err(state_err) => {
tracing::warn!(
run_id = %run_id,
error = %state_err,
"Failed to load run state after worker launch failure"
);
None
}
};
let launch_message = format!("Failed to spawn worker: {err}");
let (error, reason) = failure_honoring_pending_cancel(pending_control, || {
(
WorkflowError::engine_with_anyhow("Failed to spawn worker", err),
FailureReason::LaunchFailed,
)
});
let message = if reason == FailureReason::Cancelled {
"Run cancelled before worker launch completed".to_string()
} else {
launch_message
};
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&error,
fabro_types::RunTiming::default(),
reason,
None,
None,
None,
None,
);
let _ = workflow_event::append_event(run_store, &run_id, &failure_event).await;
fail_managed_run(state, run_id, reason, message);
state.scheduler_notify.notify_one();
}
async fn append_worker_exit_failure(
run_store: &fabro_store::RunDatabase,
run_id: RunId,
worker_exit: &WorkerExit,
) {
let state = match run_store.state().await {
Ok(state) => state,
Err(err) => {
tracing::warn!(run_id = %run_id, error = %err, "Failed to load run state after worker exit");
return;
}
};
let terminal = state.status.is_terminal();
if terminal {
return;
}
let (error, reason) = failure_for_incomplete_run(
state.pending_control,
format!(
"Worker exited before emitting a terminal run event: {}",
worker_exit.detail
),
);
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&error,
fabro_types::RunTiming::default(),
reason,
None,
None,
None,
None,
);
if let Err(err) = workflow_event::append_event(run_store, &run_id, &failure_event).await {
tracing::warn!(run_id = %run_id, error = %err, "Failed to append worker exit failure");
}
}
#[expect(
clippy::disallowed_methods,
reason = "Worker subprocess startup resolves Cargo's test binary env override when present."
)]
fn worker_launch_spec(
state: &AppState,
run_id: RunId,
mode: RunExecutionMode,
run_dir: &std::path::Path,
agent_fabro_tools_enabled: bool,
github_app_private_key: Option<String>,
) -> anyhow::Result<WorkerLaunchSpec> {
let current_exe = std::env::current_exe().context("reading current executable path")?;
let executable =
std::env::var_os(EnvVars::CARGO_BIN_EXE_FABRO).map_or(current_exe, PathBuf::from);
let storage_dir = state.server_storage_dir();
let runtime_directory = Storage::new(&storage_dir).runtime_directory();
let daemon = ServerDaemon::read(&runtime_directory)?.with_context(|| {
format!(
"server record {} is missing",
runtime_directory.record_path().display()
)
})?;
let scopes = if agent_fabro_tools_enabled {
WorkerScopeSet::run_worker_with_agent_run_tools()
} else {
WorkerScopeSet::run_worker()
};
let worker_token = issue_worker_token_with_scopes(state.worker_token_keys(), &run_id, scopes)
.map_err(|_| anyhow::anyhow!("failed to sign worker token"))?;
let log_destination = resolved_log_destination(state)?;
let fabro_log = if (state.env_lookup)(EnvVars::FABRO_LOG).is_none() {
state.server_settings().server.logging.level.clone()
} else {
None
};
Ok(WorkerLaunchSpec {
executable,
server_target: daemon.bind.to_target(),
storage_dir,
run_dir: run_dir.to_path_buf(),
run_id,
mode: worker_mode_arg(mode),
worker_token,
log_destination,
fabro_log,
active_config_path: state.active_config_path().to_path_buf(),
github_app_private_key,
})
}
fn resolved_log_destination(state: &AppState) -> anyhow::Result<LogDestination> {
let env_value = (state.env_lookup)(EnvVars::FABRO_LOG_DESTINATION);
fabro_config::resolve_log_destination_with_env(
state.server_settings().server.logging.destination,
env_value.as_deref(),
)
}
fn runtime_question_from_interview_record(question: &InterviewQuestionRecord) -> Question {
Question {
id: question.id.clone(),
text: question.text.clone(),
question_type: question.question_type,
options: question.options.clone(),
allow_freeform: question.allow_freeform,
default: None,
timeout_seconds: question.timeout_seconds,
stage: question.stage.clone(),
metadata: HashMap::new(),
context_display: question.context_display.clone(),
review_target: question.review_target.clone(),
}
}
fn api_question_from_interview_record(question: &InterviewQuestionRecord) -> ApiQuestion {
ApiQuestion {
id: question.id.clone(),
text: question.text.clone(),
stage: question.stage.clone(),
question_type: question.question_type,
options: question.options.clone(),
allow_freeform: question.allow_freeform,
timeout_seconds: question.timeout_seconds,
context_display: question.context_display.clone(),
review_target: question.review_target.clone(),
}
}
fn api_question_from_pending_interview(record: &PendingInterviewRecord) -> ApiQuestion {
api_question_from_interview_record(&record.question)
}
#[allow(
clippy::result_large_err,
reason = "Pending-interview lookup maps storage failures to HTTP responses."
)]
async fn load_pending_interview(
state: &AppState,
run_id: RunId,
qid: &str,
) -> Result<LoadedPendingInterview, Response> {
let projection = state
.load_run_projection(&run_id)
.await
.map_err(IntoResponse::into_response)?;
let Some(record) = projection.pending_interviews.get(qid) else {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response());
};
Ok(LoadedPendingInterview {
run_id,
qid: qid.to_string(),
question: record.question.clone(),
})
}
#[allow(
clippy::result_large_err,
reason = "Interview answer validation returns HTTP 400 responses directly."
)]
fn validate_answer_for_question(
question: &InterviewQuestionRecord,
answer: &Answer,
) -> Result<(), Response> {
match (&question.question_type, &answer.value) {
(
QuestionType::YesNo | QuestionType::Confirmation,
fabro_interview::AnswerValue::Yes | fabro_interview::AnswerValue::No,
)
| (
_,
fabro_interview::AnswerValue::Interrupted
| fabro_interview::AnswerValue::Skipped
| fabro_interview::AnswerValue::Timeout,
) => Ok(()),
(QuestionType::MultipleChoice, fabro_interview::AnswerValue::Selected(key)) => {
if question.options.iter().any(|option| option.key == *key) {
Ok(())
} else {
Err(ApiError::bad_request("Invalid option key.").into_response())
}
}
(QuestionType::MultiSelect, fabro_interview::AnswerValue::MultiSelected(keys)) => {
if keys
.iter()
.all(|key| question.options.iter().any(|option| option.key == *key))
{
Ok(())
} else {
Err(ApiError::bad_request("Invalid option key.").into_response())
}
}
(QuestionType::Freeform, fabro_interview::AnswerValue::Text(text))
if !text.trim().is_empty() =>
{
Ok(())
}
(_, fabro_interview::AnswerValue::Text(text))
if question.allow_freeform && !text.trim().is_empty() =>
{
Ok(())
}
_ => Err(ApiError::bad_request("Answer does not match question type.").into_response()),
}
}
#[allow(
clippy::result_large_err,
reason = "Interview submission maps validation failures to HTTP responses."
)]
async fn submit_pending_interview_answer(
state: &AppState,
pending: &LoadedPendingInterview,
submission: AnswerSubmission,
) -> Result<(), Response> {
validate_answer_for_question(&pending.question, &submission.answer)?;
deliver_answer_to_run(state, pending.run_id, &pending.qid, submission).await
}
#[allow(
clippy::result_large_err,
reason = "Interview delivery maps run-state failures to HTTP responses."
)]
async fn deliver_answer_to_run(
state: &AppState,
run_id: RunId,
qid: &str,
submission: AnswerSubmission,
) -> Result<(), Response> {
let transport = match claim_run_answer_transport(state, run_id, qid) {
Ok(transport) => transport,
Err(StatusCode::NOT_FOUND) => {
return Err(ApiError::not_found("Run not found.").into_response());
}
Err(StatusCode::CONFLICT) => {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response());
}
Err(status) => {
return Err(
ApiError::new(status, "Run is not ready to accept answers.").into_response()
);
}
};
if let Ok(()) = transport.submit(qid, submission).await {
Ok(())
} else {
release_run_answer_claim(state, run_id, qid);
Err(ApiError::new(
StatusCode::SERVICE_UNAVAILABLE,
"Failed to deliver answer to the active run.",
)
.into_response())
}
}
#[allow(
clippy::result_large_err,
reason = "Answer request parsing returns HTTP 400 responses directly."
)]
fn answer_from_request(
req: SubmitAnswerRequest,
question: &InterviewQuestionRecord,
) -> Result<Answer, Response> {
match req {
SubmitAnswerRequest::YesRequest(_) => Ok(Answer::yes()),
SubmitAnswerRequest::NoRequest(_) => Ok(Answer::no()),
SubmitAnswerRequest::SelectedRequest(req) => {
let key = req.option_key;
let option = question
.options
.iter()
.find(|option| option.key == key)
.cloned();
match option {
Some(option) => Ok(Answer::selected(key, option)),
None => Err(ApiError::bad_request("Invalid option key.").into_response()),
}
}
SubmitAnswerRequest::MultiSelectedRequest(req) => {
for key in &req.option_keys {
let valid = question.options.iter().any(|option| option.key == *key);
if !valid {
return Err(ApiError::bad_request("Invalid option key.").into_response());
}
}
Ok(Answer::multi_selected(req.option_keys))
}
SubmitAnswerRequest::TextRequest(req) => Ok(Answer::text(req.text)),
}
}
/// Execute a single run: transitions runnable → starting → running →
/// completed/failed/cancelled.
async fn execute_run(state: Arc<AppState>, run_id: RunId) {
if state.is_shutting_down() {
return;
}
if state.registry_factory_override.is_some() {
Box::pin(execute_run_in_process(state, run_id)).await;
return;
}
Box::pin(execute_run_subprocess(state, run_id)).await;
}
async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
// Transition to Starting and set up cancel infrastructure
let (cancel_rx, run_dir, event_tx, cancel_token, execution_mode) = {
let mut runs = state.runs.lock().expect("runs lock poisoned");
let managed_run = match runs.get_mut(&run_id) {
Some(r) if r.status == RunStatus::Runnable => r,
_ => return,
};
let Some(run_dir) = managed_run.run_dir.clone() else {
return;
};
let (cancel_tx, cancel_rx) = oneshot::channel::<()>();
let cancel_token = CancellationToken::new();
let (event_tx, _) = broadcast::channel(256);
managed_run.status = RunStatus::Starting;
managed_run.cancel_tx = Some(cancel_tx);
managed_run.cancel_token = Some(cancel_token.clone());
managed_run.event_tx = Some(event_tx);
(
cancel_rx,
run_dir,
managed_run.event_tx.clone(),
cancel_token,
managed_run.execution_mode,
)
};
// Create interviewer and event plumbing (this is the "provisioning" phase)
let interviewer = Arc::new(ControlInterviewer::new());
let interview_runtime: Arc<dyn Interviewer> = interviewer.clone();
let emitter = Emitter::new(run_id);
if let Some(tx_clone) = event_tx {
emitter.on_event(move |event| {
let _ = tx_clone.send(event.clone());
});
}
let registry_override = state
.registry_factory_override
.as_ref()
.map(|factory| Arc::new(factory(Arc::clone(&interview_runtime))));
let emitter = Arc::new(emitter);
let steering_hub = Arc::new(fabro_workflow::SteeringHub::new(Arc::clone(&emitter)));
// Transition to Running, populate interviewer
let cancelled_during_setup = {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
if managed_run.status == RunStatus::Starting {
managed_run.status = RunStatus::Running;
managed_run.answer_transport = Some(RunAnswerTransport::InProcess {
interviewer: Arc::clone(&interviewer),
steering_hub: Arc::clone(&steering_hub),
});
false
} else {
// Was cancelled during setup
clear_live_run_state(managed_run);
state.scheduler_notify.notify_one();
true
}
} else {
false
}
};
if cancelled_during_setup {
if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await {
error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status");
}
return;
}
let run_store = match state.stores.runs.open_run(&run_id).await {
Ok(run_store) => run_store,
Err(e) => {
tracing::error!(run_id = %run_id, error = %e, "Failed to open run store");
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.status = RunStatus::Failed {
reason: FailureReason::WorkflowError,
};
managed_run.error = Some(format!("Failed to open run store: {e}"));
clear_live_run_state(managed_run);
}
state.scheduler_notify.notify_one();
return;
}
};
tokio::spawn(forward_run_events_to_global(
Arc::clone(&state),
run_id,
run_store.subscribe(),
));
let persisted = match Persisted::load_from_store(&run_store.clone().into(), &run_dir).await {
Ok(persisted) => persisted,
Err(e) => {
tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run");
fail_run_before_execution(
&state,
run_id,
FailureReason::WorkflowError,
format!("Failed to load persisted run: {e}"),
)
.await;
return;
}
};
let server_settings = state.server_settings();
let github_settings = &server_settings.server.integrations.github;
if cancel_token.is_cancelled() {
finish_cancelled_run_before_execution(&state, run_id).await;
return;
}
if reject_run_if_sandbox_provider_disabled(
&state,
&server_settings,
run_id,
&persisted.run_spec().settings.run,
)
.await
{
return;
}
let github_app_result = {
let run_spec = persisted.run_spec();
let settings = &run_spec.settings.run;
let clone_can_use_github_credentials = settings.execution.mode != RunMode::DryRun
&& settings.environment.provider.clones_workspace()
&& run_spec
.repo_origin_url()
.is_some_and(|origin| !origin.trim().is_empty());
let pull_request_can_use_github_credentials =
settings.execution.mode != RunMode::DryRun && settings.pull_request.is_some();
if settings.integrations.github.is_token_requested() {
state.github_credentials(github_settings).await
} else if clone_can_use_github_credentials || pull_request_can_use_github_credentials {
match state.github_credentials(github_settings).await {
Ok(github_app) => Ok(github_app),
Err(err) => {
tracing::warn!(
run_id = %run_id,
error = %err,
"GitHub credentials unavailable; pull request creation will be skipped"
);
Ok(None)
}
}
} else {
Ok(None)
}
};
let github_app = match github_app_result {
Ok(github_app) => github_app,
Err(e) => {
if cancel_token.is_cancelled() {
finish_cancelled_run_before_execution(&state, run_id).await;
return;
}
tracing::error!(run_id = %run_id, error = %e, "Invalid GitHub credentials");
fail_run_before_execution(
&state,
run_id,
FailureReason::WorkflowError,
format!("Invalid GitHub credentials: {e}"),
)
.await;
return;
}
};
let github_integration = match persisted
.run_spec()
.settings
.run
.integrations
.github
.resolve_integration()
{
Ok(integration) => integration,
Err(err) => {
tracing::error!(
run_id = %run_id,
error = %err,
"GitHub permission interpolation failed"
);
fail_run_before_execution(
&state,
run_id,
FailureReason::WorkflowError,
format!("Failed to resolve GitHub permissions: {err}"),
)
.await;
return;
}
};
let vault = match state.stores.vault.snapshot().await {
Ok(vault) => vault,
Err(err) => {
tracing::error!(run_id = %run_id, error = ?err, "Loading run secrets failed");
fail_run_before_execution(
&state,
run_id,
FailureReason::WorkflowError,
"Loading run secrets failed".to_string(),
)
.await;
return;
}
};
let services = operations::StartServices {
run_id,
cancel_token: cancel_token.clone(),
emitter: Arc::clone(&emitter),
interviewer: Arc::clone(&interview_runtime),
steering_hub: Arc::clone(&steering_hub),
run_store: run_store.clone().into(),
event_sink: workflow_event::RunEventSink::store(run_store.clone()),
artifact_sink: Some(ArtifactSink::Store(state.artifact_store.clone())),
run_control: None,
github_app,
github_integration,
vault: Arc::new(AsyncRwLock::new(vault.into_vault())),
catalog: state.catalog(),
on_node: None,
registry_override,
fabro_run_tools: None,
};
let execution = async {
match execution_mode {
RunExecutionMode::Start => operations::start(&run_dir, services).await,
RunExecutionMode::Resume => operations::resume(&run_dir, services).await,
}
};
let result = tokio::select! {
result = execution => ExecutionResult::Completed(Box::new(result)),
_ = cancel_rx => {
cancel_token.cancel();
ExecutionResult::CancelledBySignal
}
};
if matches!(&result, ExecutionResult::CancelledBySignal) {
if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await {
error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status");
}
}
// Save final projection
let final_projection = match run_store.state().await {
Ok(state) => Some(state),
Err(err) => {
tracing::warn!(run_id = %run_id, error = %err, "Failed to load run state from store");
None
}
};
// Accumulate aggregate usage after execution completes.
if let Some(ref projection) = final_projection {
if projection.current_checkpoint().is_some() {
let mut agg = state
.aggregate_billing
.lock()
.expect("aggregate_billing lock poisoned");
accumulate_billing_rollup(
&mut agg,
&fabro_workflow::billing_rollup_from_projection(projection, None),
);
}
}
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
match &result {
ExecutionResult::Completed(result) => {
// A run can fail either before it produces a `Started` or in
// its own outcome; both carry the same `WorkflowError`.
let outcome = match result.as_ref() {
Ok(started) => started.finalized.outcome.as_ref().map(|_| ()),
Err(e) => Err(e),
};
match outcome {
Ok(()) => {
info!(run_id = %run_id, "Run completed");
managed_run.status = RunStatus::Succeeded {
reason: SuccessReason::Completed,
};
}
Err(WorkflowError::Cancelled) => {
info!(run_id = %run_id, "Run cancelled");
managed_run.status = RunStatus::Failed {
reason: FailureReason::Cancelled,
};
}
Err(e) => {
let detail = e.display_with_causes();
error!(run_id = %run_id, error = %detail, "Run failed");
managed_run.status = RunStatus::Failed {
reason: e.failure_reason(),
};
managed_run.error = Some(detail);
}
}
}
ExecutionResult::CancelledBySignal => {
info!(run_id = %run_id, "Run cancelled");
managed_run.status = RunStatus::Failed {
reason: FailureReason::Cancelled,
};
}
}
managed_run.checkpoint = final_projection
.as_ref()
.and_then(|projection| projection.current_checkpoint().cloned());
managed_run.run_dir = Some(run_dir);
clear_live_run_state(managed_run);
}
drop(runs);
state.scheduler_notify.notify_one();
}
async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
let (run_dir, execution_mode) = {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if state.is_shutting_down() {
return;
}
let managed_run = match runs.get_mut(&run_id) {
Some(run) if run.status == RunStatus::Runnable => run,
_ => return,
};
let Some(run_dir) = managed_run.run_dir.clone() else {
return;
};
managed_run.status = RunStatus::Starting;
(run_dir, managed_run.execution_mode)
};
let run_store = match state.stores.runs.open_run(&run_id).await {
Ok(run_store) => run_store,
Err(err) => {
tracing::error!(run_id = %run_id, error = %err, "Failed to open run store");
fail_managed_run(
&state,
run_id,
FailureReason::WorkflowError,
format!("Failed to open run store: {err}"),
);
state.scheduler_notify.notify_one();
return;
}
};
tokio::spawn(forward_run_events_to_global(
Arc::clone(&state),
run_id,
run_store.subscribe(),
));
let run_state = match run_store.state().await {
Ok(run_state) => run_state,
Err(err) => {
tracing::error!(run_id = %run_id, error = %err, "Failed to load run state");
fail_managed_run(
&state,
run_id,
FailureReason::WorkflowError,
format!("Failed to load run state: {err}"),
);
state.scheduler_notify.notify_one();
return;
}
};
let agent_fabro_tools_enabled = run_state.spec.settings.run.agent.fabro_tools;
if reject_run_if_sandbox_provider_disabled(
&state,
&state.server_settings(),
run_id,
&run_state.spec.settings.run,
)
.await
{
return;
}
let github_app_private_key = match state.vault_secret(EnvVars::GITHUB_APP_PRIVATE_KEY).await {
Ok(value) => value,
Err(err) => {
fail_run_before_execution(
&state,
run_id,
FailureReason::WorkflowError,
"Loading worker secrets failed".to_string(),
)
.await;
tracing::error!(run_id = %run_id, error = ?err, "Loading worker secrets failed");
return;
}
};
let state_for_build = Arc::clone(&state);
let run_dir_for_build = run_dir.clone();
let start_result = spawn_blocking(move || {
worker_launch_spec(
state_for_build.as_ref(),
run_id,
execution_mode,
&run_dir_for_build,
agent_fabro_tools_enabled,
github_app_private_key,
)
})
.await
.context("worker_launch_spec task failed")
.and_then(|inner| inner);
let launch_result = match start_result {
Ok(spec) => state.worker_runtime.start(spec).await,
Err(err) => Err(err),
};
let started_worker = match launch_result {
Ok(worker) => worker,
Err(err) => {
fail_worker_launch(&state, &run_store, run_id, err).await;
return;
}
};
let worker_ref = started_worker.worker_ref.clone();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.worker_ref = Some(worker_ref.clone());
managed_run.run_dir = Some(run_dir.clone());
managed_run.answer_transport = Some(RunAnswerTransport::Worker {
run_id,
bus: Arc::clone(&state.worker_control_bus),
});
}
}
let stderr_task = tokio::spawn(drain_worker_stderr(run_id, started_worker.stderr));
let worker_exit = match started_worker.wait.await {
Ok(exit) => exit,
Err(err) => {
tracing::error!(run_id = %run_id, error = %err, "Failed while waiting on worker");
let message = format!("Worker wait failed: {err}");
state.worker_runtime.force_stop(&worker_ref).await;
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&WorkflowError::engine_with_source("Worker wait failed", err),
fabro_types::RunTiming::default(),
FailureReason::Terminated,
None,
None,
None,
None,
);
let _ = workflow_event::append_event(&run_store, &run_id, &failure_event).await;
fail_managed_run(&state, run_id, FailureReason::Terminated, message);
state.scheduler_notify.notify_one();
return;
}
};
match stderr_task.await {
Ok(Ok(())) => {}
Ok(Err(err)) => {
tracing::warn!(run_id = %run_id, error = %err, "Worker stderr drain failed");
}
Err(err) => {
tracing::warn!(run_id = %run_id, error = %err, "Worker stderr task panicked");
}
}
let superseded = {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.get(&run_id)
.is_some_and(|managed_run| managed_run.worker_ref.as_ref() != Some(&worker_ref))
};
if superseded {
tracing::info!(
run_id = %run_id,
worker_ref = ?worker_ref,
"Skipping stale worker cleanup for superseded run execution"
);
return;
}
append_worker_exit_failure(&run_store, run_id, &worker_exit).await;
let final_state = match run_store.state().await {
Ok(state) => state,
Err(err) => {
tracing::warn!(run_id = %run_id, error = %err, "Failed to load final run state from store");
fail_managed_run(
&state,
run_id,
FailureReason::WorkflowError,
format!("Failed to load final run state: {err}"),
);
state.scheduler_notify.notify_one();
return;
}
};
if final_state.current_checkpoint().is_some() {
let mut agg = state
.aggregate_billing
.lock()
.expect("aggregate_billing lock poisoned");
accumulate_billing_rollup(
&mut agg,
&fabro_workflow::billing_rollup_from_projection(&final_state, None),
);
}
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
if final_state.status != managed_run.status {
managed_run.status = final_state.status;
} else if !worker_exit.success {
managed_run.status = RunStatus::Failed {
reason: FailureReason::Terminated,
};
}
managed_run.error = final_state
.conclusion
.as_ref()
.and_then(|conclusion| {
conclusion.failure.as_ref().map(|failure| {
render_compact_with_causes(&failure.detail.message, &failure.detail.causes)
})
})
.or_else(|| managed_run.error.clone());
managed_run.checkpoint = final_state.current_checkpoint().cloned();
managed_run.run_dir = Some(run_dir);
clear_live_run_state(managed_run);
}
drop(runs);
state.scheduler_notify.notify_one();
}
/// Background task that promotes runnable runs when capacity is available.
pub fn spawn_scheduler(state: Arc<AppState>) {
tokio::spawn(async move {
loop {
tokio::select! {
() = state.scheduler_notify.notified() => {},
() = sleep(std::time::Duration::from_secs(1)) => {},
}
if state.is_shutting_down() {
break;
}
let runs_to_start = {
let runs = state.runs.lock().expect("runs lock poisoned");
let active = runs
.values()
.filter(|r| counts_toward_scheduler_capacity(r.status))
.count();
let available = state.max_concurrent_runs.saturating_sub(active);
if available == 0 {
Vec::new()
} else {
let mut runnable: Vec<_> = runs
.iter()
.filter(|(_, r)| r.status == RunStatus::Runnable)
.map(|(id, r)| (*id, r.created_at))
.collect();
runnable.sort_by_key(|(_, created_at)| *created_at);
runnable
.into_iter()
.take(available)
.map(|(id, _)| id)
.collect::<Vec<_>>()
}
};
for id in runs_to_start {
if state.is_shutting_down() {
break;
}
let state_clone = Arc::clone(&state);
tokio::spawn(
execute_run(state_clone, id).instrument(tracing::info_span!("run", id = %id)),
);
}
}
});
}
async fn append_control_request(
state: &AppState,
run_id: RunId,
action: RunControlAction,
actor: Option<Principal>,
) -> anyhow::Result<()> {
let run_store = state.stores.runs.open_run(&run_id).await?;
let event = match action {
RunControlAction::Cancel => workflow_event::Event::RunCancelRequested { actor },
RunControlAction::Pause => workflow_event::Event::RunPauseRequested { actor },
RunControlAction::Unpause => workflow_event::Event::RunUnpauseRequested { actor },
};
if action == RunControlAction::Cancel {
workflow_event::append_event_if(&run_store, &run_id, &event, |projection| {
projection.pending_control != Some(RunControlAction::Cancel)
})
.await
.map(|_| ())
} else {
workflow_event::append_event(&run_store, &run_id, &event).await
}
}
/// Returns a 409 response with an actionable "unarchive first" message if the
/// run is currently archived. Returns `None` otherwise (including when the run
/// doesn't exist — the caller's own not-found handling will surface that).
async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option<Response> {
let summary = state
.stores
.run_summaries
.get(run_id, Utc::now())
.await
.ok()
.flatten()?;
summary.lifecycle.archived_at.is_some().then(|| {
ApiError::new(
StatusCode::CONFLICT,
operations::archived_rejection_message(run_id),
)
.into_response()
})
}
#[cfg(test)]
#[expect(
clippy::disallowed_methods,
reason = "server unit tests stage fixtures with sync std::fs writes"
)]
mod tests;
Reference in a new issue View git blame Copy permalink