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

feat(run): harden server-supervised worker lifecycle

Browse source 
Move subprocess workers fully behind the server-owned run store by
switching worker/server coordination to HTTP-backed run events and
control state. Reconcile stale in-flight runs on boot, terminate live
workers during shutdown, and update process titles to reflect server and
worker lifecycle phases.
This commit is contained in:
Bryan Helmkamp 2026-04-07 07:59:31 -04:00
parent 625b05dd78
commit ba02af2f88
50 changed files with 2732 additions and 499 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1868,6 +1868,7 @@ dependencies = [
"fabro-interview",
"fabro-llm",
"fabro-model",
"fabro-proc",
"fabro-retro",
"fabro-sandbox",
"fabro-store",

20
docs/api-reference/fabro-api.yaml
View file

@ -2573,6 +2573,14 @@ components:
type: integer
description: Position in the queue (1-based). Only present when status is `queued`.
example: 3
status_reason:
allOf:
- $ref: "#/components/schemas/StatusReason"
nullable: true
pending_control:
allOf:
- $ref: "#/components/schemas/RunControlAction"
nullable: true
created_at:
type: string
format: date-time
@ -2868,6 +2876,14 @@ components:
- sandbox_init_failed
- sandbox_initializing
RunControlAction:
description: Run control action requested by the API.
type: string
enum:
- cancel
- pause
- unpause
RunStatusRecord:
description: Internal run status record from the event projection.
type: object
@ -3048,6 +3064,10 @@ components:
status_reason:
type: string
nullable: true
pending_control:
allOf:
- $ref: "#/components/schemas/RunControlAction"
nullable: true
duration_ms:
type: integer
format: int64

32
lib/crates/fabro-cli/src/args.rs
View file

@ -718,17 +718,29 @@ pub(crate) struct AttachArgs {
pub(crate) run: String,
}
#[derive(Debug, Clone, Copy, ValueEnum)]
pub(crate) enum RunWorkerMode {
Start,
Resume,
}
#[derive(Args)]
pub(crate) struct RunnerArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) struct RunWorkerArgs {
/// Fabro server target: http(s) URL or absolute Unix socket path
#[arg(long)]
pub(crate) server: String,
/// Run scratch directory
#[arg(long)]
pub(crate) run_dir: PathBuf,
/// Run ID
#[arg(long)]
pub(crate) run_id: fabro_types::RunId,
/// Resume from checkpoint instead of fresh start
#[arg(long)]
pub(crate) resume: bool,
/// Worker mode
#[arg(long, value_enum)]
pub(crate) mode: RunWorkerMode,
}
#[derive(Args, Debug, Clone, Default)]
@ -797,9 +809,9 @@ pub(crate) enum RunCommands {
Start(StartArgs),
/// Attach to a running or finished workflow run
Attach(AttachArgs),
/// Internal: queue or resume a workflow run via the server
#[command(name = "__runner", hide = true)]
Runner(RunnerArgs),
/// Internal: execute a single workflow run locally
#[command(name = "__run-worker", hide = true)]
RunWorker(RunWorkerArgs),
/// Show the diff of changes from a workflow run
#[command(hide = true)]
Diff(DiffArgs),
@ -822,7 +834,7 @@ impl RunCommands {
Self::Create(_) => "create",
Self::Start(_) => "start",
Self::Attach(_) => "attach",
Self::Runner(_) => "__runner",
Self::RunWorker(_) => "__run-worker",
Self::Diff(_) => "diff",
Self::Logs(_) => "logs",
Self::Resume(_) => "resume",

52
lib/crates/fabro-cli/src/commands/run/logs.rs
View file

@ -14,6 +14,8 @@ use crate::args::{GlobalArgs, LogsArgs};
use crate::server_client;
use crate::server_runs::ServerSummaryLookup;
const FOLLOW_TERMINAL_GRACE: Duration = Duration::from_millis(500);
pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let lookup = ServerSummaryLookup::connect(&args.server).await?;
let run = lookup.resolve(&args.run)?;
@ -54,12 +56,6 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
}
if args.follow {
if events
.iter()
.any(|event| matches!(event_name(event), Some("run.completed" | "run.failed")))
{
return Ok(());
}
follow_store_logs(
client,
&run_id,
@ -149,6 +145,7 @@ async fn follow_store_logs(
let stdout = io::stdout();
let mut out = stdout.lock();
let mut next_seq = seq;
let mut terminal_deadline = None;
loop {
match time::timeout(
@ -158,6 +155,7 @@ async fn follow_store_logs(
.await
{
Ok(Ok(events)) => {
let had_events = !events.is_empty();
let saw_terminal = events
.iter()
.any(|event| matches!(event_name(event), Some("run.completed" | "run.failed")));
@ -173,27 +171,37 @@ async fn follow_store_logs(
out.flush()?;
next_seq = event.seq.saturating_add(1);
}
if saw_terminal {
flush_remaining_store_events(
client, run_id, next_seq, pretty, styles, &mut out,
)
.await?;
debug!("Observed terminal event while following logs, stopping follow");
break;
if saw_terminal || (terminal_deadline.is_some() && had_events) {
terminal_deadline = Some(time::Instant::now() + FOLLOW_TERMINAL_GRACE);
}
}
Err(_) => {
if run_concluded(client, run_id).await? {
flush_remaining_store_events(
client, run_id, next_seq, pretty, styles, &mut out,
)
.await?;
debug!("Run reached terminal status, stopping follow");
break;
terminal_deadline
.get_or_insert_with(|| time::Instant::now() + FOLLOW_TERMINAL_GRACE);
}
}
Ok(Err(err)) => return Err(err),
}
let Some(deadline) = terminal_deadline else {
continue;
};
if time::Instant::now() < deadline {
continue;
}
let flushed_next_seq =
flush_remaining_store_events(client, run_id, next_seq, pretty, styles, &mut out)
.await?;
if flushed_next_seq > next_seq {
next_seq = flushed_next_seq;
terminal_deadline = Some(time::Instant::now() + FOLLOW_TERMINAL_GRACE);
continue;
}
debug!("Run reached terminal status and log tail is quiet, stopping follow");
break;
}
Ok(())
@ -220,12 +228,13 @@ async fn flush_remaining_store_events(
pretty: bool,
styles: &Styles,
out: &mut dyn Write,
) -> Result<()> {
) -> Result<u32> {
let events = client
.list_run_events(run_id, Some(next_seq), None)
.await
.context("Failed to list server-backed run events while finalizing follow")?;
let mut next_seq = next_seq;
for event in events {
let line = event_payload_line(&event)?;
if pretty {
@ -235,9 +244,10 @@ async fn flush_remaining_store_events(
} else {
writeln!(out, "{line}")?;
}
next_seq = event.seq.saturating_add(1);
}
out.flush()?;
Ok(())
Ok(next_seq)
}
fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result<String> {

11
lib/crates/fabro-cli/src/commands/run/mod.rs
View file

@ -1,7 +1,7 @@
use anyhow::Result;
use fabro_util::terminal::Styles;
use crate::args::{AttachArgs, GlobalArgs, RunArgs, RunCommands, RunnerArgs, StartArgs};
use crate::args::{AttachArgs, GlobalArgs, RunArgs, RunCommands, RunWorkerArgs, StartArgs};
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
use crate::user_config::settings_layer_with_storage_dir;
@ -76,11 +76,12 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
}
Ok(())
}
RunCommands::Runner(RunnerArgs {
storage_dir,
RunCommands::RunWorker(RunWorkerArgs {
server,
run_dir,
run_id,
resume,
}) => runner::execute(run_id, storage_dir.clone_path(), resume).await,
mode,
}) => runner::execute(run_id, server, run_dir, mode).await,
RunCommands::Diff(args) => diff::run(args, globals).await,
RunCommands::Logs(args) => {
let styles = Styles::detect_stdout();

339
lib/crates/fabro-cli/src/commands/run/runner.rs
View file

@ -1,46 +1,325 @@
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Result, anyhow};
use fabro_types::{RunId, RunStatus};
use tokio::time::sleep;
use anyhow::{Context, Result, anyhow};
use fabro_config::RunScratch;
use fabro_interview::FileInterviewer;
use fabro_store::{Database, EventPayload, RunDatabase};
use fabro_types::{EventBody, RunEvent, RunId, Settings, StatusReason};
use fabro_workflow::event::{Emitter, RunEventSink};
use fabro_workflow::run_control::RunControlState;
use object_store::memory::InMemory as MemoryObjectStore;
#[cfg(unix)]
use tokio::signal::unix::{SignalKind, signal};
use crate::args::RunWorkerMode;
use crate::server_client;
use crate::user_config::load_settings;
use crate::shared::github::build_github_app_credentials;
const STORE_FLUSH_INTERVAL: Duration = Duration::from_millis(100);
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum WorkerTitlePhase {
Start,
Resume,
Init,
Running,
Waiting,
Paused,
Succeeded,
Failed,
Cancelled,
}
pub(crate) async fn execute(
run_id: RunId,
storage_dir: Option<PathBuf>,
resume: bool,
server: String,
run_dir: PathBuf,
mode: RunWorkerMode,
) -> Result<()> {
let _ = fabro_proc::title_init();
set_worker_title(&run_id, initial_worker_title_phase(mode));
let storage_dir = match storage_dir {
Some(storage_dir) => storage_dir,
None => load_settings()?.storage_dir(),
let client = server_client::connect_server_target_direct(&server).await?;
let run_store = load_seed_run_store(&client, &run_id).await?;
let run_state = run_store
.state()
.await
.with_context(|| format!("failed to load run state for {run_id}"))?;
let run_record = run_state
.run
.as_ref()
.ok_or_else(|| anyhow!("Run {run_id} has no run record in store"))?;
let scratch = RunScratch::new(&run_dir);
let interviewer = Arc::new(FileInterviewer::new(
scratch.interview_request_path(),
scratch.interview_response_path(),
scratch.interview_claim_path(),
));
let run_control = RunControlState::new();
install_signal_handlers(Arc::clone(&run_control))?;
let github_app = maybe_build_github_app_credentials(&run_record.settings)?;
let event_client = client.clone_for_reuse();
let services = fabro_workflow::operations::StartServices {
run_id,
cancel_token: None,
emitter: Arc::new(Emitter::new(run_id)),
interviewer,
run_store: run_store.clone(),
event_sink: RunEventSink::fanout(vec![
RunEventSink::store(run_store),
RunEventSink::callback(move |event| {
update_worker_title_from_event(&event);
let client = event_client.clone_for_reuse();
async move { client.append_run_event(&event.run_id, &event).await }
}),
]),
run_control: Some(run_control),
github_app,
on_node: None,
registry_override: None,
};
let client = server_client::connect_server(&storage_dir).await?;
client.start_run(&run_id, resume).await?;
loop {
let state = client.get_run_state(&run_id).await?;
let Some(status) = state.status.as_ref().map(|record| record.status) else {
return Err(anyhow!("Run {run_id} has no status record in store"));
};
match status {
RunStatus::Succeeded => return Ok(()),
RunStatus::Failed | RunStatus::Dead => {
return Err(anyhow!("Run {run_id} finished with status {status}"));
}
RunStatus::Submitted
| RunStatus::Starting
| RunStatus::Running
| RunStatus::Paused
| RunStatus::Removing => {
sleep(Duration::from_millis(100)).await;
}
match mode {
RunWorkerMode::Start => {
fabro_workflow::operations::start(&run_dir, services).await?;
}
RunWorkerMode::Resume => {
fabro_workflow::operations::resume(&run_dir, services).await?;
}
}
Ok(())
}
fn open_memory_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(MemoryObjectStore::new()),
"",
STORE_FLUSH_INTERVAL,
))
}
async fn load_seed_run_store(
client: &server_client::ServerStoreClient,
run_id: &RunId,
) -> Result<RunDatabase> {
let events = client
.list_run_events(run_id, None, None)
.await
.with_context(|| format!("failed to fetch run events for {run_id}"))?;
let payloads = events
.into_iter()
.map(|event| event.payload)
.collect::<Vec<_>>();
seed_run_store(run_id, &payloads).await
}
async fn seed_run_store(run_id: &RunId, events: &[EventPayload]) -> Result<RunDatabase> {
let store = open_memory_store();
let run_store = store
.create_run(run_id)
.await
.with_context(|| format!("failed to create in-memory run store for {run_id}"))?;
for payload in events {
run_store
.append_event(payload)
.await
.with_context(|| format!("failed to seed in-memory run store for {run_id}"))?;
}
Ok(run_store)
}
fn set_worker_title(run_id: &RunId, phase: WorkerTitlePhase) {
fabro_proc::title_set(&worker_title(run_id, phase));
}
fn initial_worker_title_phase(mode: RunWorkerMode) -> WorkerTitlePhase {
match mode {
RunWorkerMode::Start => WorkerTitlePhase::Start,
RunWorkerMode::Resume => WorkerTitlePhase::Resume,
}
}
fn worker_title(run_id: &RunId, phase: WorkerTitlePhase) -> String {
let short_id: String = run_id.to_string().chars().take(12).collect();
let phase = match phase {
WorkerTitlePhase::Start => "start",
WorkerTitlePhase::Resume => "resume",
WorkerTitlePhase::Init => "init",
WorkerTitlePhase::Running => "running",
WorkerTitlePhase::Waiting => "waiting",
WorkerTitlePhase::Paused => "paused",
WorkerTitlePhase::Succeeded => "succeeded",
WorkerTitlePhase::Failed => "failed",
WorkerTitlePhase::Cancelled => "cancelled",
};
format!("fabro {short_id} {phase}")
}
fn worker_title_phase_for_event(body: &EventBody) -> Option<WorkerTitlePhase> {
match body {
EventBody::RunStarting(_) => Some(WorkerTitlePhase::Init),
EventBody::RunRunning(_) | EventBody::RunUnpaused(_) => Some(WorkerTitlePhase::Running),
EventBody::InterviewStarted(_) => Some(WorkerTitlePhase::Waiting),
EventBody::InterviewCompleted(_) | EventBody::InterviewTimeout(_) => {
Some(WorkerTitlePhase::Running)
}
EventBody::RunPaused(_) => Some(WorkerTitlePhase::Paused),
EventBody::RunCompleted(_) => Some(WorkerTitlePhase::Succeeded),
EventBody::RunFailed(props) => Some(if props.reason == Some(StatusReason::Cancelled) {
WorkerTitlePhase::Cancelled
} else {
WorkerTitlePhase::Failed
}),
_ => None,
}
}
fn update_worker_title_from_event(event: &RunEvent) {
if let Some(phase) = worker_title_phase_for_event(&event.body) {
set_worker_title(&event.run_id, phase);
}
}
fn maybe_build_github_app_credentials(
settings: &Settings,
) -> Result<Option<fabro_github::GitHubAppCredentials>> {
let needs_github_app = settings
.sandbox_settings()
.and_then(|sandbox| sandbox.provider.as_deref())
.is_some_and(|provider| provider == "daytona")
|| settings
.pull_request
.as_ref()
.is_some_and(|pull_request| pull_request.enabled)
|| settings.github_permissions().is_some();
if needs_github_app {
build_github_app_credentials(settings.app_id())
} else {
Ok(None)
}
}
fn install_signal_handlers(run_control: Arc<RunControlState>) -> Result<()> {
#[cfg(unix)]
{
let mut pause = signal(SignalKind::user_defined1())?;
let pause_control = Arc::clone(&run_control);
tokio::spawn(async move {
while pause.recv().await.is_some() {
pause_control.request_pause();
}
});
let mut unpause = signal(SignalKind::user_defined2())?;
tokio::spawn(async move {
while unpause.recv().await.is_some() {
run_control.request_unpause();
}
});
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
WorkerTitlePhase, initial_worker_title_phase, worker_title, worker_title_phase_for_event,
};
use crate::args::RunWorkerMode;
use fabro_types::fixtures;
use fabro_types::run_event::{
InterviewCompletedProps, InterviewStartedProps, RunCompletedProps, RunControlEffectProps,
RunFailedProps, RunStatusTransitionProps,
};
use fabro_types::{EventBody, StatusReason};
#[test]
fn worker_title_uses_short_run_id_and_phase() {
let short_id: String = fixtures::RUN_1.to_string().chars().take(12).collect();
assert_eq!(
worker_title(&fixtures::RUN_1, WorkerTitlePhase::Start),
format!("fabro {short_id} start")
);
assert_eq!(
worker_title(&fixtures::RUN_1, WorkerTitlePhase::Succeeded),
format!("fabro {short_id} succeeded")
);
}
#[test]
fn initial_worker_title_phase_matches_mode() {
assert_eq!(
initial_worker_title_phase(RunWorkerMode::Start),
WorkerTitlePhase::Start
);
assert_eq!(
initial_worker_title_phase(RunWorkerMode::Resume),
WorkerTitlePhase::Resume
);
}
#[test]
fn worker_title_phase_tracks_lifecycle_events() {
assert_eq!(
worker_title_phase_for_event(&EventBody::RunStarting(RunStatusTransitionProps {
reason: None,
})),
Some(WorkerTitlePhase::Init)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::RunPaused(RunControlEffectProps::default())),
Some(WorkerTitlePhase::Paused)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::InterviewStarted(InterviewStartedProps {
question: "Approve?".to_string(),
question_type: "yes_no".to_string(),
})),
Some(WorkerTitlePhase::Waiting)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::InterviewCompleted(InterviewCompletedProps {
question: "Approve?".to_string(),
answer: "yes".to_string(),
duration_ms: 10,
})),
Some(WorkerTitlePhase::Running)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::RunCompleted(RunCompletedProps {
duration_ms: 10,
artifact_count: 0,
status: "success".to_string(),
reason: None,
total_cost: None,
final_git_commit_sha: None,
final_patch: None,
usage: None,
})),
Some(WorkerTitlePhase::Succeeded)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::RunFailed(RunFailedProps {
error: "cancelled".to_string(),
duration_ms: 10,
reason: Some(StatusReason::Cancelled),
git_commit_sha: None,
})),
Some(WorkerTitlePhase::Cancelled)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::RunFailed(RunFailedProps {
error: "boom".to_string(),
duration_ms: 10,
reason: Some(StatusReason::Terminated),
git_commit_sha: None,
})),
Some(WorkerTitlePhase::Failed)
);
}
}

3
lib/crates/fabro-cli/src/commands/server/foreground.rs
View file

@ -15,9 +15,6 @@ pub(crate) async fn execute(
storage_dir: Option<PathBuf>,
styles: &'static Styles,
) -> Result<()> {
let _ = fabro_proc::title_init();
fabro_proc::title_set(&format!("fabro: server {bind}"));
serve_args.bind = Some(bind.to_string());
let _record_guard = scopeguard::guard(record_path, |path| {

33
lib/crates/fabro-cli/src/main.rs
View file

@ -491,37 +491,52 @@ mod tests {
}
#[test]
fn parse_runner_command() {
fn parse_run_worker_command() {
let cli = Cli::try_parse_from([
"fabro",
"__runner",
"__run-worker",
"--server",
"/tmp/fabro.sock",
"--run-dir",
"/tmp/run",
"--run-id",
"01ARZ3NDEKTSV4RRFFQ69G5FAV",
"--mode",
"start",
])
.expect("should parse");
match *cli.command {
Commands::RunCmd(RunCommands::Runner(args)) => {
Commands::RunCmd(RunCommands::RunWorker(args)) => {
assert_eq!(args.server, "/tmp/fabro.sock");
assert_eq!(args.run_dir, std::path::PathBuf::from("/tmp/run"));
assert_eq!(args.run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap());
assert!(!args.resume);
assert!(matches!(args.mode, args::RunWorkerMode::Start));
}
_ => panic!("unexpected command variant"),
}
}
#[test]
fn parse_runner_with_resume() {
fn parse_run_worker_with_resume_mode() {
let cli = Cli::try_parse_from([
"fabro",
"__runner",
"__run-worker",
"--server",
"http://127.0.0.1:3000",
"--run-dir",
"/tmp/run",
"--run-id",
"01ARZ3NDEKTSV4RRFFQ69G5FAV",
"--resume",
"--mode",
"resume",
])
.expect("should parse");
match *cli.command {
Commands::RunCmd(RunCommands::Runner(args)) => {
Commands::RunCmd(RunCommands::RunWorker(args)) => {
assert_eq!(args.server, "http://127.0.0.1:3000");
assert_eq!(args.run_dir, std::path::PathBuf::from("/tmp/run"));
assert_eq!(args.run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap());
assert!(args.resume);
assert!(matches!(args.mode, args::RunWorkerMode::Resume));
}
_ => panic!("unexpected command variant"),
}

15
lib/crates/fabro-cli/src/server_client.rs
View file

@ -3,7 +3,7 @@ use std::num::NonZeroU64;
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context as _, Result, anyhow};
use anyhow::{Context as _, Result, anyhow, bail};
use fabro_api::types;
use fabro_server::bind::Bind;
use fabro_store::{EventEnvelope, RunSummary, StageId};
@ -113,6 +113,19 @@ pub(crate) async fn connect_server(storage_dir: &Path) -> Result<ServerStoreClie
})
}
pub(crate) async fn connect_server_target_direct(target: &str) -> Result<ServerStoreClient> {
let client = if target.starts_with("http://") || target.starts_with("https://") {
connect_remote_api_client(target, None)?
} else {
let path = Path::new(target);
if !path.is_absolute() {
bail!("server target must be an http(s) URL or absolute Unix socket path");
}
connect_unix_socket_api_client(path).await?
};
Ok(ServerStoreClient { client })
}
pub(crate) async fn connect_server_only(args: &ServerTargetArgs) -> Result<ServerStoreClient> {
let settings = user_config::load_settings()?;
let target = user_config::resolve_server_target(args, &settings)?;

165
lib/crates/fabro-cli/tests/it/cmd/runner.rs
View file

@ -1,33 +1,56 @@
use fabro_store::EventEnvelope;
use fabro_test::{fabro_snapshot, test_context};
use fabro_types::{EventBody, RunEvent};
use super::support::run_state;
use super::support::{run_events, run_state, server_target};
use crate::support::{fabro_json_snapshot, unique_run_id};
const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
fn stored_worker_events(run_dir: &std::path::Path) -> Vec<RunEvent> {
run_events(run_dir).iter().map(run_event).collect()
}
fn run_event(event: &EventEnvelope) -> RunEvent {
RunEvent::try_from(&event.payload).expect("stored event should parse")
}
fn assert_worker_succeeded(run_dir: &std::path::Path, stdout: &[u8]) {
assert!(
stdout.is_empty(),
"worker should not emit event transport on stdout"
);
let events = stored_worker_events(run_dir);
assert!(events.iter().any(|event| matches!(
&event.body,
EventBody::RunCompleted(props) if props.status == "success"
)));
}
#[test]
fn help() {
let context = test_context!();
let mut cmd = context.command();
cmd.args(["__runner", "--help"]);
cmd.args(["__run-worker", "--help"]);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
----- stdout -----
Internal: queue or resume a workflow run via the server
Internal: execute a single workflow run locally
Usage: fabro __runner [OPTIONS] --run-id <RUN_ID>
Usage: fabro __run-worker [OPTIONS] --server <SERVER> --run-dir <RUN_DIR> --run-id <RUN_ID> --mode <MODE>
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--run-id <RUN_ID> Run ID
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--resume Resume from checkpoint instead of fresh start
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--run-dir <RUN_DIR> Run scratch directory
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--run-id <RUN_ID> Run ID
--mode <MODE> Worker mode [possible values: start, resume]
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-h, --help Print help
----- stderr -----
");
}
@ -63,32 +86,30 @@ digraph CachedGraph {
.success();
let run_dir = context.find_run_dir(&run_id);
let server = server_target(&context.storage_dir);
std::fs::remove_file(&workflow_path).unwrap();
context
let output = context
.command()
.args(["__runner", "--run-id", run_id.as_str()])
.args([
"__run-worker",
"--server",
server.as_str(),
"--run-dir",
run_dir.to_str().unwrap(),
"--run-id",
run_id.as_str(),
"--mode",
"start",
])
.timeout(SHARED_DAEMON_TIMEOUT)
.assert()
.success();
.success()
.get_output()
.stdout
.clone();
let conclusion = serde_json::to_value(
run_state(&run_dir)
.conclusion
.expect("conclusion should exist"),
)
.unwrap();
fabro_json_snapshot!(
context,
serde_json::json!({
"status": conclusion["status"],
}),
@r#"
{
"status": "success"
}
"#
);
assert_worker_succeeded(&run_dir, &output);
}
#[test]
@ -147,16 +168,23 @@ digraph GitHubApp {
context.write_home(".fabro/settings.toml", "version = 1\n");
let server = server_target(&context.storage_dir);
let mut cmd = context.command();
cmd.env("GITHUB_APP_PRIVATE_KEY", "%%%not-base64%%%");
cmd.args(["__runner", "--run-id", run_id.as_str()]);
cmd.args([
"__run-worker",
"--server",
server.as_str(),
"--run-dir",
run_dir.to_str().unwrap(),
"--run-id",
run_id.as_str(),
"--mode",
"start",
]);
cmd.timeout(SHARED_DAEMON_TIMEOUT);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
----- stdout -----
----- stderr -----
");
let assert = cmd.assert().success();
assert_worker_succeeded(&run_dir, &assert.get_output().stdout);
}
#[test]
@ -190,30 +218,28 @@ digraph DetachedStoreOnly {
.success();
let run_dir = context.find_run_dir(&run_id);
context
let server = server_target(&context.storage_dir);
let output = context
.command()
.args(["__runner", "--run-id", run_id.as_str()])
.args([
"__run-worker",
"--server",
server.as_str(),
"--run-dir",
run_dir.to_str().unwrap(),
"--run-id",
run_id.as_str(),
"--mode",
"start",
])
.timeout(SHARED_DAEMON_TIMEOUT)
.assert()
.success();
.success()
.get_output()
.stdout
.clone();
let conclusion = serde_json::to_value(
run_state(&run_dir)
.conclusion
.expect("conclusion should exist"),
)
.unwrap();
fabro_json_snapshot!(
context,
serde_json::json!({
"status": conclusion["status"],
}),
@r#"
{
"status": "success"
}
"#
);
assert_worker_succeeded(&run_dir, &output);
}
#[test]
@ -246,6 +272,8 @@ digraph Test {
.unwrap()
.trim()
.to_string();
let run_dir = context.find_run_dir(&run_id);
let server = server_target(&context.storage_dir);
context
.command()
@ -277,13 +305,24 @@ digraph Test {
"#);
let mut cmd = context.command();
cmd.args(["__runner", "--run-id", &run_id, "--resume"]);
cmd.args([
"__run-worker",
"--server",
&server,
"--run-dir",
run_dir.to_str().unwrap(),
"--run-id",
&run_id,
"--mode",
"resume",
]);
cmd.timeout(SHARED_DAEMON_TIMEOUT);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
success: false
exit_code: 1
----- stdout -----
----- stderr -----
error: Precondition failed: run already finished successfully — nothing to resume
");
let inspect_after = context

2
lib/crates/fabro-cli/tests/it/cmd/server_start.rs
View file

@ -156,7 +156,7 @@ fn concurrent_autostart_converges_on_one_shared_daemon_and_cleans_up() {
String::from_utf8(output.stdout)
.expect("ps output should be UTF-8")
.lines()
.filter(|line| line.contains("fabro: server") && line.contains(socket_path))
.filter(|line| line.contains("fabro server") && line.contains(socket_path))
.count()
}

12
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -664,6 +664,18 @@ fn server_endpoint(storage_dir: &Path) -> Option<(reqwest::Client, String)> {
}
}
pub(crate) fn server_target(storage_dir: &Path) -> String {
let record_path = Storage::new(storage_dir).server_state().record_path();
let record = std::fs::read_to_string(record_path)
.ok()
.and_then(|content| serde_json::from_str::<TestServerRecord>(&content).ok())
.expect("server record should exist");
match record.bind {
Bind::Unix(path) => path.to_string_lossy().to_string(),
Bind::Tcp(addr) => format!("http://{addr}"),
}
}
async fn get_server_json<T: serde::de::DeserializeOwned>(run_dir: &Path, path: &str) -> T {
let runs_dir = run_dir.parent().expect("run dir should have parent");
let storage_dir = runs_dir.parent().expect("runs dir should have parent");

4
lib/crates/fabro-cli/tests/it/cmd/system_prune.rs
View file

@ -39,7 +39,7 @@ fn system_prune_dry_run_lists_matching_runs_without_deleting() {
let context = test_context!();
let run = setup_completed_fast_dry_run(&context);
let mut filters = context.filters();
filters.push((r"\d{8}-dry-run-".to_string(), "[DATE]-dry-run-".to_string()));
filters.push((r"\b\d{8}-\[ULID\]".to_string(), "[DATE]-[ULID]".to_string()));
filters.push((
r"\b\d+(\.\d+)?\s(?:[KMGT]?B|B)\b".to_string(),
"[SIZE]".to_string(),
@ -58,7 +58,7 @@ fn system_prune_dry_run_lists_matching_runs_without_deleting() {
success: true
exit_code: 0
----- stdout -----
would delete: 20260406-[ULID] (Simple)
would delete: [DATE]-[ULID] (Simple)
----- stderr -----
1 run(s) would be deleted ([SIZE] freed). Pass --yes to confirm.

6
lib/crates/fabro-proc/src/lib.rs
View file

@ -9,9 +9,11 @@ mod title;
pub use title::{init as title_init, set as title_set};
pub use signal::process_alive;
pub use signal::{process_alive, process_group_alive};
#[cfg(unix)]
pub use signal::{sigkill, sigterm, sigterm_process_group};
pub use signal::{
sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2,
};
#[cfg(unix)]
pub use flock::{flock_unlock, try_flock_exclusive};

54
lib/crates/fabro-proc/src/signal.rs
View file

@ -18,6 +18,27 @@ pub fn process_alive(pid: u32) -> bool {
}
}
/// Check whether any process in the given process group is alive.
///
/// On Unix, sends signal 0 to `-pgid` via `kill(2)`. Returns `false` if the
/// process-group id does not fit in `i32`. On non-Unix platforms,
/// conservatively returns `true`.
pub fn process_group_alive(pgid: u32) -> bool {
#[cfg(unix)]
{
let Ok(pgid) = i32::try_from(pgid) else {
return false;
};
// SAFETY: kill(-pgid, 0) is a read-only probe for the process group.
unsafe { libc::kill(-pgid, 0) == 0 }
}
#[cfg(not(unix))]
{
let _ = pgid;
true
}
}
/// Send SIGTERM to a single process.
#[cfg(unix)]
pub fn sigterm(pid: u32) {
@ -50,3 +71,36 @@ pub fn sigterm_process_group(pid: u32) {
}
}
}
/// Send SIGKILL to an entire process group.
#[cfg(unix)]
pub fn sigkill_process_group(pid: u32) {
if let Ok(pid) = i32::try_from(pid) {
// SAFETY: kill with -pid signals the process group.
unsafe {
libc::kill(-pid, libc::SIGKILL);
}
}
}
/// Send SIGUSR1 to a single process.
#[cfg(unix)]
pub fn sigusr1(pid: u32) {
if let Ok(pid) = i32::try_from(pid) {
// SAFETY: kill with a valid pid and SIGUSR1 is safe.
unsafe {
libc::kill(pid, libc::SIGUSR1);
}
}
}
/// Send SIGUSR2 to a single process.
#[cfg(unix)]
pub fn sigusr2(pid: u32) {
if let Ok(pid) = i32::try_from(pid) {
// SAFETY: kill with a valid pid and SIGUSR2 is safe.
unsafe {
libc::kill(pid, libc::SIGUSR2);
}
}
}

1
lib/crates/fabro-server/Cargo.toml
View file

@ -24,6 +24,7 @@ fabro-github = { path = "../fabro-github" }
fabro-agent = { path = "../fabro-agent" }
fabro-llm = { path = "../fabro-llm" }
fabro-model = { path = "../fabro-model" }
fabro-proc = { path = "../fabro-proc" }
fabro-retro = { path = "../fabro-retro" }
fabro-types = { path = "../fabro-types" }
fabro-util = { path = "../fabro-util" }

2
lib/crates/fabro-server/src/demo/mod.rs
View file

@ -191,6 +191,8 @@ pub(crate) async fn get_run_status(
status: RunStatus::Running,
error: None,
queue_position: None,
status_reason: None,
pending_control: None,
created_at: item.created_at,
}),
)

141
lib/crates/fabro-server/src/serve.rs
View file

@ -9,6 +9,7 @@ use fabro_util::terminal::Styles;
use object_store::ObjectStore;
use object_store::local::LocalFileSystem;
use tokio::net::{TcpListener, UnixListener};
use tokio::sync::watch;
use tokio::time::interval;
use tracing::{error, info, warn};
@ -20,11 +21,21 @@ use crate::bind::{self, Bind};
use crate::github_webhooks::WebhookManager;
use crate::jwt_auth::{AuthMode, AuthStrategy, resolve_auth_mode_with_lookup};
use crate::secret_store::SecretStore;
use crate::server::{build_app_state_with_path, build_router, spawn_scheduler};
use crate::tls::{ClientAuth, build_rustls_config, serve_tls};
use crate::server::{
build_app_state_with_path, build_router, reconcile_incomplete_runs_on_startup,
shutdown_active_workers, spawn_scheduler,
};
use crate::tls::{ClientAuth, build_rustls_config, serve_tls_with_shutdown};
use fabro_llm::client::Client as LlmClient;
use fabro_sandbox::SandboxProvider;
#[derive(Clone, Copy)]
enum ServerTitlePhase {
Boot,
Listening,
Stopping,
}
#[derive(Args, Clone)]
pub struct ServeArgs {
/// Address to bind to (host:port for TCP, or path containing / for Unix socket)
@ -103,6 +114,9 @@ pub async fn serve_command(
styles: &'static Styles,
storage_dir_override: Option<PathBuf>,
) -> anyhow::Result<()> {
let _ = fabro_proc::title_init();
set_server_title(ServerTitlePhase::Boot, None);
let config_path = args.config.clone();
let disk_settings = load_settings(config_path.as_deref())?;
let active_config_path = resolved_config_path(config_path.as_deref());
@ -188,6 +202,13 @@ pub async fn serve_command(
active_config_path,
matches!(&auth_mode, AuthMode::Disabled),
)?;
let reconciled = reconcile_incomplete_runs_on_startup(&state).await?;
if reconciled > 0 {
info!(
reconciled_runs = reconciled,
"Reconciled stale in-flight runs on startup"
);
}
spawn_scheduler(Arc::clone(&state));
let router = build_router(Arc::clone(&state), auth_mode);
@ -196,19 +217,6 @@ pub async fn serve_command(
None => Bind::Tcp("127.0.0.1:3000".parse().unwrap()),
};
info!(bind = %bind_addr, dry_run = dry_run_mode, "API server started");
eprintln!(
"{}",
styles.bold.apply_to(format!(
"Fabro server listening on {}",
styles.cyan.apply_to(&bind_addr)
)),
);
if dry_run_mode {
eprintln!("{}", styles.dim.apply_to("(dry-run mode)"));
}
// Optionally start webhook listener
let webhook_app_id = {
let cfg = shared_settings.read().expect("config lock poisoned");
@ -257,6 +265,17 @@ pub async fn serve_command(
None => None,
};
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let shutdown_state = Arc::clone(&state);
tokio::spawn(async move {
shutdown_signal().await;
set_server_title(ServerTitlePhase::Stopping, None);
if let Err(err) = shutdown_active_workers(&shutdown_state).await {
error!(error = %err, "Failed to stop active workers during shutdown");
}
let _ = shutdown_tx.send(true);
});
// Spawn config polling task
let settings_for_poll = Arc::clone(&shared_settings);
let config_path_for_poll = config_path.clone();
@ -300,8 +319,8 @@ pub async fn serve_command(
.as_ref()
.and_then(|a| a.tls.clone());
match bind_addr {
Bind::Unix(ref path) => {
match &bind_addr {
Bind::Unix(path) => {
if tls_settings.is_some() {
warn!("TLS is configured but not supported on Unix sockets; ignoring TLS settings");
}
@ -312,8 +331,9 @@ pub async fn serve_command(
}
let listener = UnixListener::bind(path)?;
announce_server_ready(&bind_addr, styles, dry_run_mode);
axum::serve(listener, router)
.with_graceful_shutdown(shutdown_signal())
.with_graceful_shutdown(wait_for_shutdown(shutdown_rx.clone()))
.await?;
}
Bind::Tcp(addr) => {
@ -325,12 +345,19 @@ pub async fn serve_command(
let tls_acceptor = tokio_rustls::TlsAcceptor::from(rustls_config);
info!("TLS enabled");
announce_server_ready(&bind_addr, styles, dry_run_mode);
// TLS uses a manual accept loop and cannot use with_graceful_shutdown
serve_tls(listener, tls_acceptor, router).await?;
serve_tls_with_shutdown(
listener,
tls_acceptor,
router,
wait_for_shutdown(shutdown_rx.clone()),
)
.await?;
} else {
announce_server_ready(&bind_addr, styles, dry_run_mode);
axum::serve(listener, router)
.with_graceful_shutdown(shutdown_signal())
.with_graceful_shutdown(wait_for_shutdown(shutdown_rx.clone()))
.await?;
}
}
@ -372,6 +399,51 @@ async fn shutdown_signal() {
info!("Shutdown signal received, stopping server");
}
async fn wait_for_shutdown(mut shutdown_rx: watch::Receiver<bool>) {
if *shutdown_rx.borrow() {
return;
}
let _ = shutdown_rx.changed().await;
}
fn announce_server_ready(bind_addr: &Bind, styles: &'static Styles, dry_run_mode: bool) {
set_server_title(ServerTitlePhase::Listening, Some(bind_addr));
info!(bind = %bind_addr, dry_run = dry_run_mode, "API server started");
eprintln!(
"{}",
styles.bold.apply_to(format!(
"Fabro server listening on {}",
styles.cyan.apply_to(bind_addr)
)),
);
if dry_run_mode {
eprintln!("{}", styles.dim.apply_to("(dry-run mode)"));
}
}
fn set_server_title(phase: ServerTitlePhase, bind: Option<&Bind>) {
fabro_proc::title_set(&server_title(phase, bind));
}
fn server_title(phase: ServerTitlePhase, bind: Option<&Bind>) -> String {
match phase {
ServerTitlePhase::Boot => "fabro server boot".to_string(),
ServerTitlePhase::Listening => {
let bind = bind.expect("listening server title requires a bind");
format!("fabro server {}", server_bind_title(bind))
}
ServerTitlePhase::Stopping => "fabro server stopping".to_string(),
}
}
fn server_bind_title(bind: &Bind) -> String {
match bind {
Bind::Unix(path) => format!("unix:{}", path.display()),
Bind::Tcp(addr) => format!("tcp:{addr}"),
}
}
/// Derive client certificate verification mode from the resolved auth strategies.
fn client_auth_from_mode(auth_mode: &AuthMode) -> ClientAuth {
let strategies = match auth_mode {
@ -395,7 +467,10 @@ fn client_auth_from_mode(auth_mode: &AuthMode) -> ClientAuth {
mod tests {
use std::path::PathBuf;
use super::{ServeArgs, apply_runtime_settings};
use super::{
ServeArgs, ServerTitlePhase, apply_runtime_settings, server_bind_title, server_title,
};
use crate::bind::Bind;
use fabro_types::Settings;
#[test]
@ -419,4 +494,26 @@ mod tests {
Some(PathBuf::from("/srv/fabro-storage"))
);
}
#[test]
fn server_title_formats_boot_listening_and_stopping() {
let bind = Bind::Tcp("127.0.0.1:3000".parse().unwrap());
assert_eq!(
server_title(ServerTitlePhase::Boot, None),
"fabro server boot"
);
assert_eq!(
server_title(ServerTitlePhase::Listening, Some(&bind)),
"fabro server tcp:127.0.0.1:3000"
);
assert_eq!(
server_bind_title(&Bind::Unix(PathBuf::from("/tmp/fabro.sock"))),
"unix:/tmp/fabro.sock"
);
assert_eq!(
server_title(ServerTitlePhase::Stopping, None),
"fabro server stopping"
);
}
}

1576
lib/crates/fabro-server/src/server.rs
View file

File diff suppressed because it is too large Load diff

21
lib/crates/fabro-server/src/tls.rs
View file

@ -1,5 +1,6 @@
use std::path::Path;
use std::sync::Arc;
use std::{future::Future, pin::Pin};
use rustls::ServerConfig;
use rustls::server::WebPkiClientVerifier;
@ -69,6 +70,19 @@ pub async fn serve_tls(
tls_acceptor: tokio_rustls::TlsAcceptor,
router: axum::Router,
) -> anyhow::Result<()> {
serve_tls_with_shutdown(listener, tls_acceptor, router, std::future::pending()).await
}
/// Serve requests over TLS until the supplied shutdown future resolves.
pub async fn serve_tls_with_shutdown<F>(
listener: TcpListener,
tls_acceptor: tokio_rustls::TlsAcceptor,
router: axum::Router,
shutdown: F,
) -> anyhow::Result<()>
where
F: Future<Output = ()> + Send,
{
use hyper::body::Incoming;
use hyper::service::service_fn;
use hyper_util::rt::{TokioExecutor, TokioIo};
@ -76,9 +90,14 @@ pub async fn serve_tls(
use tower_service::Service;
let builder = Builder::new(TokioExecutor::new());
let mut shutdown = Pin::from(Box::new(shutdown));
loop {
let (tcp_stream, remote_addr) = listener.accept().await?;
let accepted = tokio::select! {
() = &mut shutdown => return Ok(()),
accepted = listener.accept() => accepted?,
};
let (tcp_stream, remote_addr) = accepted;
let tls_acceptor = tls_acceptor.clone();
let router = router.clone();

13
lib/crates/fabro-server/tests/it/api/system.rs
View file

@ -3,7 +3,6 @@ use std::time::Duration;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_config::Storage;
use fabro_server::server::create_app_state_with_options;
use fabro_types::{RunId, Settings};
use http_body_util::BodyExt;
use tempfile::tempdir;
@ -11,8 +10,8 @@ use tokio::time::timeout;
use tower::ServiceExt;
use crate::helpers::{
MINIMAL_DOT, api, body_json, minimal_manifest_json_with_dry_run, test_app_with_scheduler,
test_settings, wait_for_run_status,
MINIMAL_DOT, api, body_json, minimal_manifest_json_with_dry_run, test_app_state_with_options,
test_app_with_scheduler, test_settings, wait_for_run_status,
};
fn temp_storage_settings() -> (tempfile::TempDir, Settings) {
@ -50,7 +49,7 @@ async fn get_system_info_returns_runtime_fields() {
let (_temp, settings) = temp_storage_settings();
let expected_storage_dir = settings.storage_dir.clone().unwrap();
let app = fabro_server::server::build_router(
create_app_state_with_options(settings, 5),
test_app_state_with_options(settings, 5),
fabro_server::jwt_auth::AuthMode::Disabled,
);
@ -78,7 +77,7 @@ async fn get_system_info_returns_runtime_fields() {
async fn get_system_disk_usage_returns_summary_and_verbose_rows() {
let (_temp, settings) = temp_storage_settings();
let storage_dir = settings.storage_dir.clone().unwrap();
let app = test_app_with_scheduler(create_app_state_with_options(settings, 5));
let app = test_app_with_scheduler(test_app_state_with_options(settings, 5));
let run_id = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
start_run(&app, &run_id).await;
@ -111,7 +110,7 @@ async fn get_system_disk_usage_returns_summary_and_verbose_rows() {
async fn prune_runs_supports_dry_run_and_deletion() {
let (_temp, settings) = temp_storage_settings();
let storage_dir = settings.storage_dir.clone().unwrap();
let app = test_app_with_scheduler(create_app_state_with_options(settings, 5));
let app = test_app_with_scheduler(test_app_state_with_options(settings, 5));
let run_id = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
start_run(&app, &run_id).await;
@ -156,7 +155,7 @@ async fn prune_runs_supports_dry_run_and_deletion() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn attach_events_streams_only_matching_run_ids() {
let (_temp, settings) = temp_storage_settings();
let app = test_app_with_scheduler(create_app_state_with_options(settings, 5));
let app = test_app_with_scheduler(test_app_state_with_options(settings, 5));
let run_one = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
let run_two = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;

15
lib/crates/fabro-server/tests/it/helpers.rs
View file

@ -5,7 +5,8 @@ use axum::body::{Body, to_bytes};
use axum::http::{Request, StatusCode};
use fabro_server::jwt_auth::AuthMode;
use fabro_server::server::{
AppState, build_router, create_app_state, create_app_state_with_options, spawn_scheduler,
AppState, build_router, create_app_state, create_app_state_with_settings_and_registry_factory,
spawn_scheduler,
};
use fabro_types::Settings;
use fabro_types::settings::{LocalSandboxSettings, SandboxSettings, WorktreeMode};
@ -26,6 +27,16 @@ pub(crate) fn test_app_state() -> Arc<AppState> {
create_app_state()
}
pub(crate) fn test_app_state_with_options(
settings: Settings,
max_concurrent_runs: usize,
) -> Arc<AppState> {
let _ = max_concurrent_runs;
create_app_state_with_settings_and_registry_factory(settings, |interviewer| {
fabro_workflow::handler::default_registry(interviewer, || None)
})
}
pub(crate) fn test_settings() -> Settings {
Settings {
sandbox: Some(SandboxSettings {
@ -46,7 +57,7 @@ pub(crate) fn dry_run_settings() -> Settings {
}
pub(crate) fn dry_run_app() -> axum::Router {
let state = create_app_state_with_options(dry_run_settings(), 5);
let state = test_app_state_with_options(dry_run_settings(), 5);
spawn_scheduler(Arc::clone(&state));
build_router(state, AuthMode::Disabled)
}

12
lib/crates/fabro-server/tests/it/scenario/lifecycle.rs
View file

@ -17,7 +17,7 @@ use tower::ServiceExt;
use crate::helpers::{
POLL_ATTEMPTS, POLL_INTERVAL, api, body_json, minimal_manifest_json, run_json, test_settings,
wait_for_run_status, wait_for_run_status_not_in,
wait_for_run_status,
};
fn gate_registry(interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
@ -158,10 +158,9 @@ async fn full_http_lifecycle_cancel() {
.unwrap();
app.clone().oneshot(req).await.unwrap();
// Subscribe as soon as the scheduler has created the live event stream.
// Waiting past "starting" races with stage events because `/events`
// subscribes to future broadcast messages only; it does not replay.
wait_for_run_status_not_in(&app, &run_id, &["queued"]).await;
// Wait until the worker has reached the human gate so cancel exercises the
// live-running path rather than racing the in-memory queue transition.
let _question_id = wait_for_question_id(&app, &run_id).await;
// Cancel it
let req = Request::builder()
@ -172,7 +171,8 @@ async fn full_http_lifecycle_cancel() {
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert_eq!(body["status"], "cancelled");
assert_eq!(body["status"], "running");
assert_eq!(body["pending_control"], "cancel");
// Verify the durable store view converges to cancelled failure.
let status = wait_for_run_status(&app, &run_id, &["failed"]).await;

9
lib/crates/fabro-server/tests/it/scenario/run_completion.rs
View file

@ -1,17 +1,16 @@
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_server::server::create_app_state_with_options;
use tokio::time::sleep;
use tower::ServiceExt;
use crate::helpers::{
MINIMAL_DOT, api, create_and_start_run, dry_run_settings, test_app_with_scheduler,
wait_for_run_status,
MINIMAL_DOT, api, create_and_start_run, dry_run_settings, test_app_state_with_options,
test_app_with_scheduler, wait_for_run_status,
};
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn run_completes_and_status_is_completed() {
let state = create_app_state_with_options(dry_run_settings(), 5);
let state = test_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
let run_id = create_and_start_run(&app, MINIMAL_DOT).await;
@ -22,7 +21,7 @@ async fn run_completes_and_status_is_completed() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn attach_run_events_returns_sse_stream() {
let state = create_app_state_with_options(dry_run_settings(), 5);
let state = test_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
let run_id = create_and_start_run(&app, MINIMAL_DOT).await;

7
lib/crates/fabro-server/tests/it/scenario/sse.rs
View file

@ -2,15 +2,14 @@ use std::time::Duration;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_server::server::create_app_state_with_options;
use http_body_util::BodyExt;
use tokio::time::{sleep, timeout};
use tower::ServiceExt;
use crate::helpers::{
POLL_ATTEMPTS, POLL_INTERVAL, api, body_json, create_and_start_run_from_manifest,
dry_run_settings, minimal_manifest_json_with_dry_run, test_app_with_scheduler,
wait_for_run_status_not_in,
dry_run_settings, minimal_manifest_json_with_dry_run, test_app_state_with_options,
test_app_with_scheduler, wait_for_run_status_not_in,
};
const SIMPLE_DOT: &str = r#"digraph SSETest {
@ -39,7 +38,7 @@ async fn wait_for_checkpoint(app: &axum::Router, run_id: &str) -> serde_json::Va
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sse_stream_contains_expected_event_types() {
let state = create_app_state_with_options(dry_run_settings(), 5);
let state = test_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
let run_id =

5
lib/crates/fabro-server/tests/it/scenario/usage.rs
View file

@ -1,17 +1,16 @@
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_server::server::create_app_state_with_options;
use tokio::time::sleep;
use tower::ServiceExt;
use crate::helpers::{
MINIMAL_DOT, POLL_ATTEMPTS, POLL_INTERVAL, api, body_json, create_and_start_run,
dry_run_settings, test_app_with_scheduler, wait_for_run_status,
dry_run_settings, test_app_state_with_options, test_app_with_scheduler, wait_for_run_status,
};
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn aggregate_usage_increments_after_run_completes() {
let state = create_app_state_with_options(dry_run_settings(), 5);
let state = test_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
let run_id = create_and_start_run(&app, MINIMAL_DOT).await;

26
lib/crates/fabro-store/src/run_state.rs
View file

@ -12,8 +12,8 @@ use fabro_types::run_event::{
};
use fabro_types::{
Checkpoint, Conclusion, EventBody, FailureSignature, NodeStatusRecord, Outcome,
PullRequestRecord, Retro, RunEvent, RunId, RunRecord, RunStatus, RunStatusRecord,
SandboxRecord, StageStatus, StageUsage, StartRecord, StatusReason, TokenUsage,
PullRequestRecord, Retro, RunControlAction, RunEvent, RunId, RunRecord, RunStatus,
RunStatusRecord, SandboxRecord, StageStatus, StageUsage, StartRecord, StatusReason, TokenUsage,
};
#[derive(Debug, Clone, Default, serde::Serialize)]
@ -22,6 +22,7 @@ pub struct RunProjection {
pub graph_source: Option<String>,
pub start: Option<StartRecord>,
pub status: Option<RunStatusRecord>,
pub pending_control: Option<RunControlAction>,
pub checkpoint: Option<Checkpoint>,
pub checkpoints: Vec<(u32, Checkpoint)>,
pub conclusion: Option<Conclusion>,
@ -120,13 +121,32 @@ impl RunProjection {
EventBody::RunRemoving(props) => {
self.status = Some(run_status_record(RunStatus::Removing, props.reason, ts));
}
EventBody::RunCancelRequested(_) => {
self.pending_control = Some(RunControlAction::Cancel);
}
EventBody::RunPauseRequested(_) => {
self.pending_control = Some(RunControlAction::Pause);
}
EventBody::RunUnpauseRequested(_) => {
self.pending_control = Some(RunControlAction::Unpause);
}
EventBody::RunPaused(_) => {
self.status = Some(run_status_record(RunStatus::Paused, None, ts));
self.pending_control = None;
}
EventBody::RunUnpaused(_) => {
self.status = Some(run_status_record(RunStatus::Running, None, ts));
self.pending_control = None;
}
EventBody::RunCompleted(props) => {
self.status = Some(run_status_record(RunStatus::Succeeded, props.reason, ts));
self.pending_control = None;
self.conclusion = Some(conclusion_from_completed(props, ts)?);
self.final_patch.clone_from(&props.final_patch);
}
EventBody::RunFailed(props) => {
self.status = Some(run_status_record(RunStatus::Failed, props.reason, ts));
self.pending_control = None;
self.conclusion = Some(conclusion_from_failed(props, ts));
}
EventBody::RunRewound(_) => {
@ -330,6 +350,7 @@ impl RunProjection {
start_time: self.start.as_ref().map(|start| start.start_time),
status: self.status.as_ref().map(|status| status.status),
status_reason: self.status.as_ref().and_then(|status| status.reason),
pending_control: self.pending_control,
duration_ms: self
.conclusion
.as_ref()
@ -355,6 +376,7 @@ impl RunProjection {
fn reset_for_rewind(&mut self) {
self.status = None;
self.pending_control = None;
self.checkpoint = None;
self.checkpoints.clear();
self.conclusion = None;

103
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -233,7 +233,9 @@ mod tests {
use super::*;
use chrono::{DateTime, Utc};
use fabro_types::{AttrValue, Graph, RunRecord, RunStatus, Settings, StatusReason};
use fabro_types::{
AttrValue, Graph, RunControlAction, RunRecord, RunStatus, Settings, StatusReason,
};
use futures::TryStreamExt;
use object_store::memory::InMemory;
use object_store::path::Path;
@ -347,6 +349,18 @@ mod tests {
.unwrap();
}
async fn append_running(run: &RunDatabase, label: &str, created_at: DateTime<Utc>) {
append_created(run, label, created_at).await;
run.append_event(&event_payload(
label,
"2026-03-27T12:00:01Z",
"run.running",
&serde_json::json!({}),
))
.await
.unwrap();
}
async fn list_paths(store: Arc<dyn ObjectStore>, prefix: &str) -> Vec<String> {
let mut items = store
.list(Some(&Path::from(prefix.to_string())))
@ -406,6 +420,93 @@ mod tests {
assert!(matches!(err, StoreError::ReadOnly));
}
#[tokio::test]
async fn control_request_events_set_pending_control_without_overwriting_status() {
let (_object_store, store) = make_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_running(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:02Z",
"run.pause.requested",
&serde_json::json!({ "action": "pause" }),
))
.await
.unwrap();
let summary = store.list_runs(&ListRunsQuery::default()).await.unwrap();
assert_eq!(summary.len(), 1);
assert_eq!(summary[0].status, Some(RunStatus::Running));
assert_eq!(summary[0].pending_control, Some(RunControlAction::Pause));
}
#[tokio::test]
async fn control_effect_events_clear_pending_control_and_update_status() {
let (_object_store, store) = make_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_running(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:02Z",
"run.pause.requested",
&serde_json::json!({ "action": "pause" }),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:03Z",
"run.paused",
&serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:04Z",
"run.unpause.requested",
&serde_json::json!({ "action": "unpause" }),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"run.unpaused",
&serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:06Z",
"run.cancel.requested",
&serde_json::json!({ "action": "cancel" }),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:07Z",
"run.failed",
&serde_json::json!({
"error": "cancelled",
"duration_ms": 1,
"reason": "cancelled",
}),
))
.await
.unwrap();
let summary = store.list_runs(&ListRunsQuery::default()).await.unwrap();
assert_eq!(summary.len(), 1);
assert_eq!(summary[0].status, Some(RunStatus::Failed));
assert_eq!(summary[0].status_reason, Some(StatusReason::Cancelled));
assert_eq!(summary[0].pending_control, None);
}
#[tokio::test]
async fn reader_sees_cached_projection_and_recent_events_for_active_run() {
let (_object_store, store) = make_store();

3
lib/crates/fabro-store/src/types.rs
View file

@ -4,7 +4,7 @@ use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::{Result, StoreError};
use fabro_types::{RunEvent, RunId, RunStatus, StatusReason};
use fabro_types::{RunControlAction, RunEvent, RunId, RunStatus, StatusReason};
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSummary {
@ -17,6 +17,7 @@ pub struct RunSummary {
pub start_time: Option<DateTime<Utc>>,
pub status: Option<RunStatus>,
pub status_reason: Option<StatusReason>,
pub pending_control: Option<RunControlAction>,
pub duration_ms: Option<u64>,
pub total_cost: Option<f64>,
}

3
lib/crates/fabro-types/src/lib.rs
View file

@ -41,7 +41,8 @@ pub use settings::Settings;
pub use stage_id::StageId;
pub use start::StartRecord;
pub use status::{
InvalidTransition, ParseRunStatusError, RunStatus, RunStatusRecord, StatusReason,
InvalidTransition, ParseRunStatusError, RunControlAction, RunStatus, RunStatusRecord,
StatusReason,
};
pub use usage::StageUsage;

15
lib/crates/fabro-types/src/run_event/mod.rs
View file

@ -71,6 +71,16 @@ pub enum EventBody {
RunRunning(RunStatusTransitionProps),
#[serde(rename = "run.removing")]
RunRemoving(RunStatusTransitionProps),
#[serde(rename = "run.cancel.requested")]
RunCancelRequested(RunControlRequestedProps),
#[serde(rename = "run.pause.requested")]
RunPauseRequested(RunControlRequestedProps),
#[serde(rename = "run.unpause.requested")]
RunUnpauseRequested(RunControlRequestedProps),
#[serde(rename = "run.paused")]
RunPaused(RunControlEffectProps),
#[serde(rename = "run.unpaused")]
RunUnpaused(RunControlEffectProps),
#[serde(rename = "run.rewound")]
RunRewound(RunRewoundProps),
#[serde(rename = "run.completed")]
@ -296,6 +306,11 @@ impl EventBody {
Self::RunStarting(_) => "run.starting",
Self::RunRunning(_) => "run.running",
Self::RunRemoving(_) => "run.removing",
Self::RunCancelRequested(_) => "run.cancel.requested",
Self::RunPauseRequested(_) => "run.pause.requested",
Self::RunUnpauseRequested(_) => "run.unpause.requested",
Self::RunPaused(_) => "run.paused",
Self::RunUnpaused(_) => "run.unpaused",
Self::RunRewound(_) => "run.rewound",
Self::RunCompleted(_) => "run.completed",
Self::RunFailed(_) => "run.failed",

10
lib/crates/fabro-types/src/run_event/run.rs
View file

@ -2,7 +2,7 @@ use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use crate::{Graph, Settings, StatusReason};
use crate::{Graph, RunControlAction, Settings, StatusReason};
use super::{RunNoticeLevel, TokenUsage};
@ -51,6 +51,14 @@ pub struct RunStatusTransitionProps {
pub reason: Option<StatusReason>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunControlRequestedProps {
pub action: RunControlAction,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
pub struct RunControlEffectProps {}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunRewoundProps {
pub target_checkpoint_ordinal: usize,

8
lib/crates/fabro-types/src/status.rs
View file

@ -136,6 +136,14 @@ pub enum StatusReason {
SandboxInitializing,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunControlAction {
Cancel,
Pause,
Unpause,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunStatusRecord {
pub status: RunStatus,

240
lib/crates/fabro-workflow/src/event.rs
View file

@ -1,8 +1,10 @@
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use ::fabro_types::run_event as fabro_types;
use ::fabro_types::{RunEvent, RunId, StageStatus, StatusReason};
use ::fabro_types::{RunControlAction, RunEvent, RunId, StageStatus, StatusReason};
use anyhow::{Context, Result};
use chrono::Utc;
use fabro_store::{EventPayload, RunDatabase};
@ -10,7 +12,8 @@ use fabro_util::json::normalize_json_value;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeMap;
use tokio::sync::{mpsc, oneshot};
use tokio::io::{AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot};
use uuid::Uuid;
use crate::error::FabroError;
@ -77,6 +80,11 @@ pub enum Event {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunCancelRequested,
RunPauseRequested,
RunUnpauseRequested,
RunPaused,
RunUnpaused,
RunRewound {
target_checkpoint_ordinal: usize,
target_node_id: String,
@ -529,6 +537,21 @@ impl Event {
Self::RunRemoving { reason } => {
info!(?reason, "Run removing");
}
Self::RunCancelRequested => {
info!("Run cancel requested");
}
Self::RunPauseRequested => {
info!("Run pause requested");
}
Self::RunUnpauseRequested => {
info!("Run unpause requested");
}
Self::RunPaused => {
info!("Run paused");
}
Self::RunUnpaused => {
info!("Run unpaused");
}
Self::RunRewound {
target_checkpoint_ordinal,
target_node_id,
@ -1069,6 +1092,11 @@ pub fn event_name(event: &Event) -> &'static str {
Event::RunStarting { .. } => "run.starting",
Event::RunRunning { .. } => "run.running",
Event::RunRemoving { .. } => "run.removing",
Event::RunCancelRequested => "run.cancel.requested",
Event::RunPauseRequested => "run.pause.requested",
Event::RunUnpauseRequested => "run.unpause.requested",
Event::RunPaused => "run.paused",
Event::RunUnpaused => "run.unpaused",
Event::RunRewound { .. } => "run.rewound",
Event::WorkflowRunCompleted { .. } => "run.completed",
Event::WorkflowRunFailed { .. } => "run.failed",
@ -1370,6 +1398,23 @@ fn event_body_from_event(event: &Event) -> EventBody {
Event::RunRemoving { reason } => {
EventBody::RunRemoving(fabro_types::RunStatusTransitionProps { reason: *reason })
}
Event::RunCancelRequested => {
EventBody::RunCancelRequested(fabro_types::RunControlRequestedProps {
action: RunControlAction::Cancel,
})
}
Event::RunPauseRequested => {
EventBody::RunPauseRequested(fabro_types::RunControlRequestedProps {
action: RunControlAction::Pause,
})
}
Event::RunUnpauseRequested => {
EventBody::RunUnpauseRequested(fabro_types::RunControlRequestedProps {
action: RunControlAction::Unpause,
})
}
Event::RunPaused => EventBody::RunPaused(fabro_types::RunControlEffectProps::default()),
Event::RunUnpaused => EventBody::RunUnpaused(fabro_types::RunControlEffectProps::default()),
Event::RunRewound {
target_checkpoint_ordinal,
target_node_id,
@ -2311,30 +2356,114 @@ pub async fn append_event(run_store: &RunDatabase, run_id: &RunId, event: &Event
.map_err(anyhow::Error::from)
}
enum StoreProgressCommand {
Event(EventPayload),
pub async fn append_event_to_sink(
sink: &RunEventSink,
run_id: &RunId,
event: &Event,
) -> Result<()> {
let stored = to_run_event(run_id, event);
sink.write_run_event(&stored).await
}
#[derive(Clone)]
pub enum RunEventSink {
Store(RunDatabase),
JsonLines(Arc<AsyncMutex<Pin<Box<dyn AsyncWrite + Send>>>>),
Callback(Arc<RunEventSinkCallback>),
Composite(Vec<RunEventSink>),
}
type RunEventSinkFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>;
type RunEventSinkCallback = dyn Fn(RunEvent) -> RunEventSinkFuture + Send + Sync + 'static;
impl RunEventSink {
#[must_use]
pub fn store(run_store: RunDatabase) -> Self {
Self::Store(run_store)
}
#[must_use]
pub fn json_lines<W>(writer: W) -> Self
where
W: AsyncWrite + Send + 'static,
{
Self::JsonLines(Arc::new(AsyncMutex::new(Box::pin(writer))))
}
#[must_use]
pub fn callback<F, Fut>(callback: F) -> Self
where
F: Fn(RunEvent) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<()>> + Send + 'static,
{
Self::Callback(Arc::new(move |event| Box::pin(callback(event))))
}
#[must_use]
pub fn fanout(sinks: Vec<Self>) -> Self {
let mut flattened = Vec::new();
for sink in sinks {
match sink {
Self::Composite(inner) => flattened.extend(inner),
other => flattened.push(other),
}
}
Self::Composite(flattened)
}
pub async fn write_run_event(&self, event: &RunEvent) -> Result<()> {
let mut pending = vec![self];
while let Some(sink) = pending.pop() {
match sink {
Self::Store(run_store) => {
let payload = build_redacted_event_payload(event, &event.run_id)?;
run_store
.append_event(&payload)
.await
.map(|_| ())
.map_err(anyhow::Error::from)?;
}
Self::JsonLines(writer) => {
let line = redacted_event_json(event)?;
let mut writer = writer.lock().await;
writer.write_all(line.as_bytes()).await?;
writer.write_all(b"\n").await?;
writer.flush().await?;
}
Self::Callback(callback) => callback(event.clone()).await?,
Self::Composite(sinks) => {
pending.extend(sinks.iter().rev());
}
}
}
Ok(())
}
}
enum RunEventCommand {
Event(RunEvent),
Flush(oneshot::Sender<()>),
}
#[derive(Clone)]
pub struct StoreProgressLogger {
tx: mpsc::UnboundedSender<StoreProgressCommand>,
pub struct RunEventLogger {
tx: mpsc::UnboundedSender<RunEventCommand>,
}
impl StoreProgressLogger {
impl RunEventLogger {
#[must_use]
pub fn new(run_store: RunDatabase) -> Self {
pub fn new(sink: RunEventSink) -> Self {
let (tx, mut rx) = mpsc::unbounded_channel();
tokio::spawn(async move {
while let Some(command) = rx.recv().await {
match command {
StoreProgressCommand::Event(payload) => {
if let Err(err) = run_store.append_event(&payload).await {
tracing::warn!(error = %err, "Failed to append event to run store");
RunEventCommand::Event(event) => {
if let Err(err) = sink.write_run_event(&event).await {
tracing::warn!(error = %err, "Failed to write run event");
}
}
StoreProgressCommand::Flush(tx) => {
RunEventCommand::Flush(tx) => {
let _ = tx.send(());
}
}
@ -2346,34 +2475,47 @@ impl StoreProgressLogger {
pub fn register(&self, emitter: &Emitter) {
let tx = self.tx.clone();
emitter.on_event(
move |event| match build_redacted_event_payload(event, &event.run_id) {
Ok(payload) => {
if tx.send(StoreProgressCommand::Event(payload)).is_err() {
tracing::warn!(
"Store progress logger channel closed while appending event"
);
}
}
Err(err) => {
tracing::warn!(error = %err, "Failed to build store event payload");
}
},
);
emitter.on_event(move |event| {
if tx.send(RunEventCommand::Event(event.clone())).is_err() {
tracing::warn!("Run event logger channel closed while forwarding event");
}
});
}
pub async fn flush(&self) {
let (tx, rx) = oneshot::channel();
if self.tx.send(StoreProgressCommand::Flush(tx)).is_err() {
tracing::warn!("Store progress logger channel closed before flush");
if self.tx.send(RunEventCommand::Flush(tx)).is_err() {
tracing::warn!("Run event logger channel closed before flush");
return;
}
if rx.await.is_err() {
tracing::warn!("Store progress logger flush dropped before completion");
tracing::warn!("Run event logger flush dropped before completion");
}
}
}
#[derive(Clone)]
pub struct StoreProgressLogger {
inner: RunEventLogger,
}
impl StoreProgressLogger {
#[must_use]
pub fn new(run_store: RunDatabase) -> Self {
Self {
inner: RunEventLogger::new(RunEventSink::store(run_store)),
}
}
pub fn register(&self, emitter: &Emitter) {
self.inner.register(emitter);
}
pub async fn flush(&self) {
self.inner.flush().await;
}
}
/// Current time as epoch milliseconds.
fn epoch_millis() -> i64 {
let millis = std::time::SystemTime::now()
@ -2725,6 +2867,46 @@ mod tests {
assert_eq!(line["properties"]["code"], "example");
}
#[tokio::test]
async fn run_event_sink_json_lines_writes_canonical_event_lines() {
use tokio::io::{AsyncBufReadExt, BufReader};
let (writer, reader) = tokio::io::duplex(4096);
let sink = RunEventSink::json_lines(writer);
let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested);
sink.write_run_event(&event).await.unwrap();
let mut reader = BufReader::new(reader);
let mut line = String::new();
reader.read_line(&mut line).await.unwrap();
let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_7).unwrap();
assert_eq!(payload.as_value()["event"], "run.pause.requested");
assert_eq!(payload.as_value()["properties"]["action"], "pause");
}
#[tokio::test]
async fn run_event_logger_registers_emitter_events_to_json_lines() {
use tokio::io::{AsyncBufReadExt, BufReader};
let (writer, reader) = tokio::io::duplex(4096);
let sink = RunEventSink::json_lines(writer);
let logger = RunEventLogger::new(sink);
let emitter = Emitter::new(fixtures::RUN_8);
logger.register(&emitter);
emitter.emit(&Event::RunPaused);
logger.flush().await;
let mut reader = BufReader::new(reader);
let mut line = String::new();
reader.read_line(&mut line).await.unwrap();
let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_8).unwrap();
assert_eq!(payload.as_value()["event"], "run.paused");
}
#[test]
fn build_redacted_event_payload_requires_id() {
let stored = to_run_event(

1
lib/crates/fabro-workflow/src/handler/manager_loop.rs
View file

@ -258,6 +258,7 @@ impl Handler for SubWorkflowHandler {
sandbox,
registry,
on_node: None,
run_control: None,
hook_runner,
env,
dry_run,

1
lib/crates/fabro-workflow/src/lib.rs
View file

@ -132,6 +132,7 @@ pub mod pipeline;
pub mod pull_request;
pub mod records;
mod retry;
pub mod run_control;
pub(crate) mod run_dir;
pub mod run_dump;
pub mod run_lookup;

12
lib/crates/fabro-workflow/src/lifecycle/mod.rs
View file

@ -31,6 +31,7 @@ use crate::event::Emitter;
use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode;
use crate::outcome::{Outcome, StageUsage};
use crate::run_control::RunControlState;
use crate::run_options::RunOptions;
use fabro_graphviz::graph::types::Graph as GvGraph;
use fabro_hooks::HookRunner;
@ -60,6 +61,8 @@ pub(crate) struct WorkflowLifecycle {
git: GitLifecycle,
artifact: ArtifactLifecycle,
on_node: crate::OnNodeCallback,
emitter: Arc<Emitter>,
run_control: Option<Arc<RunControlState>>,
/// Set in on_edge_selected when loop_restart approved; read+cleared by EventLifecycle::on_run_start
restarted_from: Arc<Mutex<Option<(String, String)>>>,
/// Shared git checkpoint result (written by git, read by event)
@ -85,6 +88,7 @@ impl WorkflowLifecycle {
run_options: &Arc<RunOptions>,
is_resume: bool,
on_node: crate::OnNodeCallback,
run_control: Option<Arc<RunControlState>>,
) -> Self {
let run_scratch = RunScratch::new(run_dir);
let restarted_from: Arc<Mutex<Option<(String, String)>>> = Arc::new(Mutex::new(None));
@ -171,6 +175,8 @@ impl WorkflowLifecycle {
git,
artifact,
on_node,
emitter: Arc::clone(emitter),
run_control,
restarted_from,
checkpoint_git_result,
is_initial_resume: AtomicBool::new(is_resume),
@ -254,6 +260,9 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
node: &WorkflowNode,
state: &WfRunState,
) -> CoreResult<WfNodeDecision> {
if let Some(run_control) = &self.run_control {
run_control.wait_if_paused(self.emitter.as_ref()).await;
}
if let Some(on_node) = &self.on_node {
on_node(node.id());
}
@ -282,6 +291,9 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
ctx: &AttemptResultContext<'_, WorkflowGraph>,
state: &WfRunState,
) -> CoreResult<()> {
if let Some(run_control) = &self.run_control {
run_control.wait_if_paused(self.emitter.as_ref()).await;
}
self.artifact.after_attempt(ctx, state).await?;
self.event.after_attempt(ctx, state).await?;
Ok(())

6
lib/crates/fabro-workflow/src/operations/resume.rs
View file

@ -3,7 +3,7 @@ use std::path::Path;
use fabro_config::RunScratch;
use crate::error::FabroError;
use crate::event::{Event, append_event};
use crate::event::{Event, append_event_to_sink};
use crate::outcome::StageStatus;
use crate::run_status::RunStatus;
@ -40,8 +40,8 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
.ok_or_else(|| FabroError::Precondition("no checkpoint to resume from".to_string()))?;
cleanup_resume_artifacts(run_dir);
append_event(
&services.run_store,
append_event_to_sink(
&services.event_sink,
&services.run_id,
&Event::RunSubmitted { reason: None },
)

149
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -15,8 +15,7 @@ use fabro_types::{RunId, Settings};
use crate::context::Context;
use crate::error::FabroError;
use crate::event::{
Emitter, Event, EventBody, RunNoticeLevel, StoreProgressLogger, append_event,
event_payload_from_redacted_json, redacted_event_json, to_run_event,
Emitter, Event, EventBody, RunEventLogger, RunEventSink, RunNoticeLevel, append_event_to_sink,
};
use crate::git::MetadataStore;
use crate::handler::HandlerRegistry;
@ -27,6 +26,7 @@ use crate::pipeline::{
classify_engine_result,
};
use crate::records::Checkpoint;
use crate::run_control::RunControlState;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::run_status::{RunStatus, StatusReason};
use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle};
@ -49,6 +49,7 @@ struct RunSession {
devcontainer: Option<DevcontainerSpec>,
seed_context: Option<Context>,
run_store: RunDatabase,
event_sink: RunEventSink,
git: Option<GitCheckpointOptions>,
github_app: Option<fabro_github::GitHubAppCredentials>,
worktree_mode: Option<WorktreeMode>,
@ -61,6 +62,7 @@ struct RunSession {
pr_model: String,
workflow_path: Option<PathBuf>,
workflow_bundle: Option<Arc<WorkflowBundle>>,
run_control: Option<Arc<RunControlState>>,
}
pub struct StartServices {
@ -69,6 +71,8 @@ pub struct StartServices {
pub emitter: Arc<Emitter>,
pub interviewer: Arc<dyn Interviewer>,
pub run_store: RunDatabase,
pub event_sink: RunEventSink,
pub run_control: Option<Arc<RunControlState>>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub on_node: crate::OnNodeCallback,
pub registry_override: Option<Arc<HandlerRegistry>>,
@ -115,11 +119,12 @@ pub(super) async fn execute_persisted_run(
let cancel_token = services.cancel_token.clone();
let run_id = services.run_id;
let run_store = services.run_store.clone();
let event_sink = services.event_sink.clone();
if let Err(err) = run_store.state().await {
let error = FabroError::engine(err.to_string());
let _ = persist_detached_failure(
run_id,
&run_store,
&event_sink,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -128,8 +133,8 @@ pub(super) async fn execute_persisted_run(
.await;
return Err(error);
}
if let Err(err) = append_event(
&run_store,
if let Err(err) = append_event_to_sink(
&event_sink,
&run_id,
&Event::RunStarting {
reason: Some(StatusReason::SandboxInitializing),
@ -140,7 +145,7 @@ pub(super) async fn execute_persisted_run(
let error = FabroError::engine(err.to_string());
let _ = persist_detached_failure(
run_id,
&run_store,
&event_sink,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -151,14 +156,14 @@ pub(super) async fn execute_persisted_run(
}
let mut bootstrap_guard =
DetachedRunBootstrapGuard::arm(run_id, run_dir, run_store.clone(), cancel_token.clone());
DetachedRunBootstrapGuard::arm(run_id, run_dir, event_sink.clone(), cancel_token.clone());
let persisted = match Persisted::load_from_store(&services.run_store, run_dir).await {
Ok(persisted) => persisted,
Err(err) => {
let _ = persist_detached_failure(
run_id,
&run_store,
&event_sink,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -175,7 +180,7 @@ pub(super) async fn execute_persisted_run(
Err(err) => {
let _ = persist_detached_failure(
run_id,
&run_store,
&event_sink,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -189,7 +194,7 @@ pub(super) async fn execute_persisted_run(
bootstrap_guard.defuse();
let mut completion_guard =
DetachedRunCompletionGuard::arm(run_id, run_store.clone(), cancel_token);
DetachedRunCompletionGuard::arm(run_id, event_sink.clone(), cancel_token);
let run_start = Instant::now();
let started = Box::pin(session.run(persisted, checkpoint)).await;
@ -199,8 +204,15 @@ pub(super) async fn execute_persisted_run(
Ok(started)
}
Err(err) => {
persist_terminal_engine_failure(run_id, &run_store, run_dir, &err, run_start.elapsed())
.await;
persist_terminal_engine_failure(
run_id,
&run_store,
&event_sink,
run_dir,
&err,
run_start.elapsed(),
)
.await;
completion_guard.defuse();
Err(err)
}
@ -210,6 +222,7 @@ pub(super) async fn execute_persisted_run(
async fn persist_terminal_engine_failure(
run_id: RunId,
run_store: &RunDatabase,
event_sink: &RunEventSink,
_run_dir: &Path,
error: &FabroError,
duration: Duration,
@ -225,8 +238,8 @@ async fn persist_terminal_engine_failure(
None,
)
.await;
if let Err(err) = append_event(
run_store,
if let Err(err) = append_event_to_sink(
event_sink,
&run_id,
&Event::WorkflowRunFailed {
error: error.clone(),
@ -356,6 +369,8 @@ impl RunSession {
Ok(Self {
cancel_token: services.cancel_token,
emitter: services.emitter,
event_sink: services.event_sink,
run_control: services.run_control,
sandbox,
llm: LlmSpec {
model: model.clone(),
@ -487,7 +502,7 @@ impl RunSession {
});
}
let store_progress_logger = StoreProgressLogger::new(self.run_store.clone());
let store_progress_logger = RunEventLogger::new(self.event_sink.clone());
store_progress_logger.register(self.emitter.as_ref());
let init_options = InitOptions {
@ -508,6 +523,7 @@ impl RunSession {
git: self.git,
worktree_mode: self.worktree_mode,
registry_override: self.registry_override,
run_control: self.run_control,
checkpoint,
seed_context: self.seed_context,
};
@ -590,7 +606,7 @@ impl RunSession {
struct DetachedRunBootstrapGuard {
run_id: RunId,
run_store: RunDatabase,
event_sink: RunEventSink,
cancel_token: Option<Arc<AtomicBool>>,
active: bool,
}
@ -599,12 +615,12 @@ impl DetachedRunBootstrapGuard {
fn arm(
run_id: RunId,
_run_dir: &Path,
run_store: RunDatabase,
event_sink: RunEventSink,
cancel_token: Option<Arc<AtomicBool>>,
) -> Self {
Self {
run_id,
run_store,
event_sink,
cancel_token,
active: true,
}
@ -628,11 +644,11 @@ impl Drop for DetachedRunBootstrapGuard {
StatusReason::SandboxInitFailed
};
let run_id = self.run_id;
let run_store = self.run_store.clone();
let event_sink = self.event_sink.clone();
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
let _ = append_event(
&run_store,
let _ = append_event_to_sink(
&event_sink,
&run_id,
&Event::WorkflowRunFailed {
error: FabroError::engine(format!("{reason:?}")),
@ -652,16 +668,16 @@ const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization
const POSTRUN_CANCELLED_MESSAGE: &str = "Run cancelled before post-run finalization completed.";
struct DetachedRunCompletionGuard {
run_store: RunDatabase,
event_sink: RunEventSink,
run_id: RunId,
cancel_token: Option<Arc<AtomicBool>>,
active: bool,
}
impl DetachedRunCompletionGuard {
fn arm(run_id: RunId, run_store: RunDatabase, cancel_token: Option<Arc<AtomicBool>>) -> Self {
fn arm(run_id: RunId, event_sink: RunEventSink, cancel_token: Option<Arc<AtomicBool>>) -> Self {
Self {
run_store,
event_sink,
run_id,
cancel_token,
active: true,
@ -698,35 +714,12 @@ impl Drop for DetachedRunCompletionGuard {
} else {
"postrun_aborted"
};
let serialized_notice = {
let stored = to_run_event(
&self.run_id,
&Event::RunNotice {
level: RunNoticeLevel::Error,
code: code.to_string(),
message: message.to_string(),
},
);
let line = match redacted_event_json(&stored) {
Ok(line) => line,
Err(err) => {
tracing::warn!(error = %err, "Failed to serialize post-run abort event");
String::new()
}
};
if line.is_empty() {
None
} else {
Some((self.run_id, line))
}
};
let run_store = self.run_store.clone();
let event_sink = self.event_sink.clone();
let run_id = self.run_id;
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
let _ = append_event(
&run_store,
let _ = append_event_to_sink(
&event_sink,
&run_id,
&Event::WorkflowRunFailed {
error: FabroError::engine(message.to_string()),
@ -736,29 +729,16 @@ impl Drop for DetachedRunCompletionGuard {
},
)
.await;
if let Some((run_id, line)) = serialized_notice.or_else(|| {
let stored = to_run_event(
&run_id,
&Event::RunNotice {
level: RunNoticeLevel::Error,
code: code.to_string(),
message: message.to_string(),
},
);
redacted_event_json(&stored).ok().map(|line| (run_id, line))
}) {
match event_payload_from_redacted_json(&line, &run_id) {
Ok(payload) => {
let _ = run_store.append_event(&payload).await;
}
Err(err) => {
tracing::warn!(
error = %err,
"Failed to build post-run abort event payload"
);
}
}
}
let _ = append_event_to_sink(
&event_sink,
&run_id,
&Event::RunNotice {
level: RunNoticeLevel::Error,
code: code.to_string(),
message: message.to_string(),
},
)
.await;
});
}
}
@ -766,7 +746,7 @@ impl Drop for DetachedRunCompletionGuard {
async fn persist_detached_failure(
run_id: RunId,
run_store: &RunDatabase,
event_sink: &RunEventSink,
_run_dir: &Path,
phase: &'static str,
reason: StatusReason,
@ -774,8 +754,8 @@ async fn persist_detached_failure(
) -> Result<(), FabroError> {
let message = error.to_string();
if let Err(err) = append_event(
run_store,
if let Err(err) = append_event_to_sink(
event_sink,
&run_id,
&Event::WorkflowRunFailed {
error: error.clone(),
@ -794,17 +774,8 @@ async fn persist_detached_failure(
code: format!("{phase}_failed"),
message: message.clone(),
};
let stored = to_run_event(&run_id, &event);
let line = redacted_event_json(&stored).map_err(|err| FabroError::Io(err.to_string()))?;
match event_payload_from_redacted_json(&line, &run_id) {
Ok(payload) => {
if let Err(err) = run_store.append_event(&payload).await {
tracing::warn!(error = %err, "Failed to append detached failure event to store");
}
}
Err(err) => {
tracing::warn!(error = %err, "Failed to build detached failure event payload");
}
if let Err(err) = append_event_to_sink(event_sink, &run_id, &event).await {
tracing::warn!(error = %err, "Failed to append detached failure notice");
}
Ok(())
@ -896,6 +867,8 @@ mod tests {
emitter,
interviewer: Arc::new(fabro_interview::AutoApproveInterviewer),
run_store: store.open_run(&fixtures::RUN_1).await.unwrap(),
event_sink: RunEventSink::store(store.open_run(&fixtures::RUN_1).await.unwrap()),
run_control: None,
github_app: None,
on_node: None,
registry_override: Some(registry),
@ -1030,7 +1003,7 @@ mod tests {
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::new(),
};
append_event(
crate::event::append_event(
&services.run_store,
&services.run_id,
&Event::CheckpointCompleted {

2
lib/crates/fabro-workflow/src/pipeline/execute.rs
View file

@ -47,6 +47,7 @@ pub async fn execute(init: Initialized) -> Executed {
sandbox,
registry,
on_node,
run_control,
hook_runner,
env,
dry_run,
@ -101,6 +102,7 @@ pub async fn execute(init: Initialized) -> Executed {
&settings_arc,
checkpoint.is_some(),
on_node,
run_control,
);
if let Some(ref cp) = checkpoint {

3
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -216,6 +216,7 @@ async fn execute_test_run_with_options(
devcontainer: None,
git: git_options,
worktree_mode: None,
run_control: None,
registry_override,
checkpoint: None,
seed_context: None,
@ -271,6 +272,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
devcontainer: None,
git: None,
worktree_mode: None,
run_control: None,
registry_override: None,
checkpoint: None,
seed_context: None,
@ -336,6 +338,7 @@ async fn run_with_lifecycle(
devcontainer: None,
git: None,
worktree_mode: None,
run_control: None,
registry_override: Some(Arc::new(registry)),
checkpoint: None,
seed_context: None,

3
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -651,6 +651,7 @@ pub async fn initialize(
sandbox,
registry,
on_node: None,
run_control: options.run_control,
hook_runner,
env,
dry_run: options.dry_run,
@ -800,6 +801,7 @@ mod tests {
devcontainer: None,
git: None,
worktree_mode: None,
run_control: None,
registry_override: None,
checkpoint: None,
seed_context: None,
@ -875,6 +877,7 @@ mod tests {
devcontainer: None,
git: None,
worktree_mode: None,
run_control: None,
registry_override: None,
checkpoint: None,
seed_context: None,

3
lib/crates/fabro-workflow/src/pipeline/types.rs
View file

@ -22,6 +22,7 @@ use crate::file_resolver::FileResolver;
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::records::{Checkpoint, Conclusion, RunRecord};
use crate::run_control::RunControlState;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::transforms::Transform;
use crate::workflow_bundle::WorkflowBundle;
@ -245,6 +246,7 @@ pub struct InitOptions {
pub git: Option<GitCheckpointOptions>,
pub worktree_mode: Option<WorktreeMode>,
pub registry_override: Option<Arc<HandlerRegistry>>,
pub run_control: Option<Arc<RunControlState>>,
pub checkpoint: Option<Checkpoint>,
pub seed_context: Option<Context>,
}
@ -264,6 +266,7 @@ pub struct Initialized {
pub sandbox: Arc<dyn Sandbox>,
pub registry: Arc<HandlerRegistry>,
pub on_node: crate::OnNodeCallback,
pub run_control: Option<Arc<RunControlState>>,
pub hook_runner: Option<Arc<HookRunner>>,
pub env: HashMap<String, String>,
pub dry_run: bool,

45
lib/crates/fabro-workflow/src/run_control.rs Normal file
View file

@ -0,0 +1,45 @@
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::Notify;
use crate::event::{Emitter, Event};
#[derive(Default)]
pub struct RunControlState {
pause_requested: AtomicBool,
notify: Notify,
}
impl RunControlState {
#[must_use]
pub fn new() -> Arc<Self> {
Arc::new(Self::default())
}
pub fn request_pause(&self) {
self.pause_requested.store(true, Ordering::Relaxed);
self.notify.notify_waiters();
}
pub fn request_unpause(&self) {
self.pause_requested.store(false, Ordering::Relaxed);
self.notify.notify_waiters();
}
pub fn pause_requested(&self) -> bool {
self.pause_requested.load(Ordering::Relaxed)
}
pub async fn wait_if_paused(&self, emitter: &Emitter) {
if !self.pause_requested() {
return;
}
emitter.emit(&Event::RunPaused);
while self.pause_requested() {
self.notify.notified().await;
}
emitter.emit(&Event::RunUnpaused);
}
}

1
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -112,6 +112,7 @@ async fn initialized(
sandbox,
registry: Arc::new(registry),
on_node: None,
run_control: None,
hook_runner: options.hook_runner,
env: options.env,
dry_run: run_options.dry_run_enabled(),

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

@ -128,6 +128,7 @@ models/root-response.ts
models/run-artifact-entry.ts
models/run-artifact-list-response.ts
models/run-checkpoint.ts
models/run-control-action.ts
models/run-error.ts
models/run-event.ts
models/run-list-item.ts

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

@ -108,6 +108,7 @@ export * from './root-response-urls';
export * from './run-artifact-entry';
export * from './run-artifact-list-response';
export * from './run-checkpoint';
export * from './run-control-action';
export * from './run-error';
export * from './run-event';
export * from './run-list-item';

30
lib/packages/fabro-api-client/src/models/run-control-action.ts Normal file
View file

@ -0,0 +1,30 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Run control action requested by the API.
*/
export const RunControlAction = {
CANCEL: 'cancel',
PAUSE: 'pause',
UNPAUSE: 'unpause'
} as const;
export type RunControlAction = typeof RunControlAction[keyof typeof RunControlAction];

8
lib/packages/fabro-api-client/src/models/run-status-response.ts
View file

@ -13,12 +13,18 @@
*/
// May contain unused imports in some cases
// @ts-ignore
import type { RunControlAction } from './run-control-action';
// May contain unused imports in some cases
// @ts-ignore
import type { RunError } from './run-error';
// May contain unused imports in some cases
// @ts-ignore
import type { RunStatus } from './run-status';
// May contain unused imports in some cases
// @ts-ignore
import type { StatusReason } from './status-reason';
/**
* Current status of a run with optional error and queue position.
@ -34,6 +40,8 @@ export interface RunStatusResponse {
* Position in the queue (1-based). Only present when status is `queued`.
*/
'queue_position'?: number;
'status_reason'?: StatusReason;
'pending_control'?: RunControlAction;
/**
* Timestamp when the run was created.
*/

6
lib/packages/fabro-api-client/src/models/store-run-summary.ts
View file

@ -13,6 +13,9 @@
*/
// May contain unused imports in some cases
// @ts-ignore
import type { RunControlAction } from './run-control-action';
/**
* Durable run summary derived from the backing store.
@ -27,7 +30,10 @@ export interface StoreRunSummary {
'start_time'?: string;
'status'?: string;
'status_reason'?: string;
'pending_control'?: RunControlAction;
'duration_ms'?: number;
'total_cost'?: number;
}