scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge remote-tracking branch 'origin/main'

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-04 13:33:32 -04:00
commit ddebb888b6
No known key found for this signature in database
10 changed files with 371 additions and 330 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -125,16 +125,16 @@ paths:
schema:
$ref: "#/components/schemas/PaginatedRunList"
post:
operationId: startRun
operationId: createRun
tags: [Runs]
summary: Start Run
description: Queues a new workflow run from a Graphviz graph source. The run is created in `queued` status and will be picked up by the scheduler.
summary: Create Run
description: Creates a new workflow run from a Graphviz graph source. The run is created in `submitted` status. Use `POST /api/v1/runs/{id}/start` to begin execution.
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/StartRunRequest"
$ref: "#/components/schemas/CreateRunRequest"
responses:
"201":
description: Run created
@ -199,6 +199,34 @@ paths:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/start:
post:
operationId: startRun
tags: [Runs]
summary: Start Run
description: Starts a submitted run, queuing it for execution. Returns 409 if the run is not in `submitted` status.
parameters:
- $ref: "#/components/parameters/RunId"
responses:
"200":
description: Run started
content:
application/json:
schema:
$ref: "#/components/schemas/RunStatusResponse"
"404":
description: Run not found
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"409":
description: Run is not in submitted status
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/pause:
post:
operationId: pauseRun
@ -307,29 +335,6 @@ paths:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/context:
get:
operationId: retrieveRunContext
tags: [Run Internals]
summary: Retrieve Run Context
description: Returns the key-value context map accumulated during the run. Empty if the run has not started.
parameters:
- $ref: "#/components/parameters/RunId"
responses:
"200":
description: Context key-value map
content:
application/json:
schema:
type: object
additionalProperties: true
"404":
description: Run not found
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/events:
get:
operationId: streamRunEvents
@ -1915,6 +1920,7 @@ components:
description: Lifecycle status of a run.
type: string
enum:
- submitted
- queued
- starting
- running
@ -1923,8 +1929,8 @@ components:
- cancelled
- paused
StartRunRequest:
description: Request body for starting a new run from a Graphviz graph source.
CreateRunRequest:
description: Request body for creating a new run from a Graphviz graph source.
type: object
required:
- dot_source

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

@ -51,13 +51,27 @@ pub(crate) async fn list_runs(
paginated_response(runs::list_items(), &pagination)
}
pub(crate) async fn start_run_stub(
pub(crate) async fn create_run_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
) -> Response {
(
StatusCode::CREATED,
Json(serde_json::json!({"id": "demo-run-new", "status": "queued", "created_at": "2026-03-06T14:30:00Z"})),
Json(serde_json::json!({"id": "demo-run-new", "status": "submitted", "created_at": "2026-03-06T14:30:00Z"})),
)
.into_response()
}
pub(crate) async fn start_run_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
(
StatusCode::OK,
Json(
serde_json::json!({"id": id, "status": "queued", "created_at": "2026-03-06T14:30:00Z"}),
),
)
.into_response()
}
@ -188,14 +202,6 @@ pub(crate) async fn checkpoint_stub(
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
}
pub(crate) async fn context_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
Path(_id): Path<String>,
) -> Response {
(StatusCode::OK, Json(serde_json::json!({}))).into_response()
}
pub(crate) async fn cancel_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,

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

@ -46,7 +46,6 @@ use crate::sessions::{SessionStore, new_session_store};
use crate::static_files;
use crate::web_auth;
use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
use fabro_workflow::context::Context;
use fabro_workflow::event::Emitter;
use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput};
use fabro_workflow::pipeline::Persisted;
@ -56,9 +55,9 @@ use fabro_api::types::AggregateUsageTotals;
pub use fabro_api::types::{
AggregateUsage, ApiQuestion, ApiQuestionOption, CompletionContentPart, CompletionMessage,
CompletionMessageRole, CompletionResponse, CompletionToolChoiceMode, CompletionUsage,
CreateCompletionRequest, ModelReference, PaginatedRunList, PaginationMeta,
QuestionType as ApiQuestionType, RunError, RunStatus, RunStatusResponse, StartRunRequest,
SubmitAnswerRequest, TokenUsage, UsageByModel,
CreateCompletionRequest, CreateRunRequest, ModelReference, PaginatedRunList, PaginationMeta,
QuestionType as ApiQuestionType, RunError, RunStatus, RunStatusResponse, SubmitAnswerRequest,
TokenUsage, UsageByModel,
};
pub fn default_page_limit() -> u32 {
@ -98,7 +97,6 @@ struct ManagedRun {
// Populated when running:
interviewer: Option<Arc<WebInterviewer>>,
event_tx: Option<broadcast::Sender<RunEvent>>,
context: Option<Context>,
checkpoint: Option<Checkpoint>,
cancel_tx: Option<oneshot::Sender<()>>,
cancel_token: Option<Arc<AtomicBool>>,
@ -197,14 +195,14 @@ pub fn build_router(state: Arc<AppState>, auth_mode: AuthMode) -> Router {
fn demo_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs", get(demo::list_runs).post(demo::start_run_stub))
.route("/runs", get(demo::list_runs).post(demo::create_run_stub))
.route("/runs/{id}", get(demo::get_run_status))
.route("/runs/{id}/questions", get(demo::get_questions_stub))
.route("/runs/{id}/questions/{qid}/answer", post(demo::answer_stub))
.route("/runs/{id}/events", get(demo::run_events_stub))
.route("/runs/{id}/checkpoint", get(demo::checkpoint_stub))
.route("/runs/{id}/context", get(demo::context_stub))
.route("/runs/{id}/cancel", post(demo::cancel_stub))
.route("/runs/{id}/start", post(demo::start_run_stub))
.route("/runs/{id}/pause", post(demo::pause_stub))
.route("/runs/{id}/unpause", post(demo::unpause_stub))
.route("/runs/{id}/graph", get(demo::get_run_graph))
@ -273,14 +271,14 @@ fn demo_routes() -> Router<Arc<AppState>> {
fn real_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs", get(list_runs).post(start_run))
.route("/runs", get(list_runs).post(create_run))
.route("/runs/{id}", get(get_run_status))
.route("/runs/{id}/questions", get(get_questions))
.route("/runs/{id}/questions/{qid}/answer", post(submit_answer))
.route("/runs/{id}/events", get(get_events))
.route("/runs/{id}/checkpoint", get(get_checkpoint))
.route("/runs/{id}/context", get(get_context))
.route("/runs/{id}/cancel", post(cancel_run))
.route("/runs/{id}/start", post(start_run))
.route("/runs/{id}/pause", post(pause_run))
.route("/runs/{id}/unpause", post(unpause_run))
.route("/runs/{id}/graph", get(get_graph))
@ -530,13 +528,13 @@ fn clear_live_run_state(run: &mut ManagedRun) {
run.cancel_token = None;
}
async fn start_run(
async fn create_run(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Json(req): Json<StartRunRequest>,
Json(req): Json<CreateRunRequest>,
) -> Response {
let run_id = RunId::new();
info!(run_id = %run_id, "Run queued");
info!(run_id = %run_id, "Run created");
let settings = state.settings.read().unwrap().clone();
let created = match Box::pin(operations::create(
state.store.as_ref(),
@ -588,12 +586,11 @@ async fn start_run(
run_id,
ManagedRun {
dot_source: req.dot_source,
status: RunStatus::Queued,
status: RunStatus::Submitted,
error: None,
created_at,
interviewer: None,
event_tx: None,
context: None,
checkpoint: None,
cancel_tx: None,
cancel_token: None,
@ -602,13 +599,11 @@ async fn start_run(
);
}
state.scheduler_notify.notify_one();
(
StatusCode::CREATED,
Json(RunStatusResponse {
id: run_id.to_string(),
status: RunStatus::Queued,
status: RunStatus::Submitted,
error: None,
queue_position: None,
created_at,
@ -617,6 +612,42 @@ async fn start_run(
.into_response()
}
async fn start_run(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let mut runs = state.runs.lock().expect("runs lock poisoned");
match runs.get_mut(&id) {
Some(managed_run) => {
if managed_run.status != RunStatus::Submitted {
return ApiError::new(StatusCode::CONFLICT, "Run is not in submitted status.")
.into_response();
}
managed_run.status = RunStatus::Queued;
let response = (
StatusCode::OK,
Json(RunStatusResponse {
id: id.to_string(),
status: RunStatus::Queued,
error: None,
queue_position: None,
created_at: managed_run.created_at,
}),
)
.into_response();
drop(runs);
state.scheduler_notify.notify_one();
response
}
None => ApiError::not_found("Run not found.").into_response(),
}
}
/// Execute a single run: transitions queued → starting → running → completed/failed/cancelled.
async fn execute_run(state: Arc<AppState>, run_id: RunId) {
// Transition to Starting and set up cancel infrastructure
@ -649,7 +680,6 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
// Create interviewer and event plumbing (this is the "provisioning" phase)
let interviewer = Arc::new(WebInterviewer::new());
let context = Context::new();
let emitter = Emitter::new(run_id);
if let Some(tx_clone) = event_tx {
emitter.on_event(move |event| {
@ -662,7 +692,7 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
.map(|factory| Arc::new(factory(Arc::clone(&interviewer) as Arc<dyn Interviewer>)));
let emitter = Arc::new(emitter);
// Transition to Running, populate interviewer + context
// Transition to Running, populate interviewer
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
@ -674,7 +704,6 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
}
managed_run.status = RunStatus::Running;
managed_run.interviewer = Some(Arc::clone(&interviewer));
managed_run.context = Some(context);
}
}
@ -808,11 +837,6 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
}
}
managed_run.checkpoint = checkpoint;
if let Ok(started) = &result {
if let Some(ctx) = &started.final_context {
managed_run.context = Some(ctx.clone());
}
}
managed_run.run_dir = Some(run_dir);
clear_live_run_state(managed_run);
}
@ -1083,25 +1107,6 @@ async fn get_checkpoint(
}
}
async fn get_context(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => match &managed_run.context {
Some(ctx) => (StatusCode::OK, Json(ctx.snapshot())).into_response(),
None => (StatusCode::OK, Json(serde_json::json!({}))).into_response(),
},
None => ApiError::not_found("Run not found.").into_response(),
}
}
async fn cancel_run(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
@ -1114,7 +1119,7 @@ async fn cancel_run(
let mut runs = state.runs.lock().expect("runs lock poisoned");
match runs.get_mut(&id) {
Some(managed_run) => match managed_run.status {
RunStatus::Queued | RunStatus::Starting | RunStatus::Running => {
RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting | RunStatus::Running => {
if let Some(token) = &managed_run.cancel_token {
token.store(true, Ordering::Relaxed);
}
@ -1688,6 +1693,31 @@ mod tests {
format!("/api/v1{path}")
}
/// Create a run via POST /runs, then start it via POST /runs/{id}/start.
/// Returns the run_id string.
async fn create_and_start_run(app: &Router, dot_source: &str) -> String {
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": dot_source})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().to_string();
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
app.clone().oneshot(req).await.unwrap();
run_id
}
#[tokio::test]
async fn test_model_unknown_returns_404() {
let app = test_app_with();
@ -1892,19 +1922,7 @@ mod tests {
let state = create_app_state();
let app = test_app_with_scheduler(state);
// Start a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id = create_and_start_run(&app, MINIMAL_DOT).await;
// Give run a moment to start
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
@ -1920,7 +1938,7 @@ mod tests {
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert_eq!(body["id"].as_str().unwrap(), run_id.to_string());
assert_eq!(body["id"].as_str().unwrap(), run_id);
let status = body["status"].as_str().unwrap();
assert!(
status == "queued"
@ -2044,11 +2062,10 @@ mod tests {
}
#[tokio::test]
async fn get_context_returns_map() {
async fn create_run_returns_submitted() {
let state = create_app_state();
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Start a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
@ -2058,22 +2075,76 @@ mod tests {
))
.unwrap();
let response = app.oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
let body = body_json(response.into_body()).await;
assert_eq!(body["status"], "submitted");
}
#[tokio::test]
async fn start_run_transitions_to_queued() {
let state = create_app_state();
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Create a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id = body["id"].as_str().unwrap();
// Get context
// Start it
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/context")))
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
let response = app.oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert!(body.is_object());
assert_eq!(body["status"], "queued");
}
#[tokio::test]
async fn start_run_conflict_when_not_submitted() {
let state = create_app_state();
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Create a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap();
// Start it (transitions to queued)
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
app.clone().oneshot(req).await.unwrap();
// Start it again — should 409
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
let response = app.oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::CONFLICT);
}
#[tokio::test]
@ -2131,19 +2202,8 @@ mod tests {
let state = create_app_state();
let app = test_app_with_scheduler(state);
// Start a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().unwrap();
// Wait for scheduler to promote run (creates event_tx)
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@ -2182,19 +2242,8 @@ mod tests {
let state = create_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
// Start a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().unwrap();
// Poll until run completes
let mut status = String::new();
@ -2360,19 +2409,8 @@ mod tests {
let state = create_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
// Start a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().unwrap();
// Poll until run completes
let mut status = String::new();
@ -2407,7 +2445,7 @@ mod tests {
}
#[tokio::test]
async fn post_runs_returns_queued_status() {
async fn post_runs_returns_submitted_status() {
let state = create_app_state();
let app = build_router(state, AuthMode::Disabled);
@ -2425,7 +2463,7 @@ mod tests {
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
// Check status is queued (no scheduler running)
// Check status is submitted (no start, no scheduler running)
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}")))
@ -2434,7 +2472,7 @@ mod tests {
let response = app.oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
assert_eq!(body["status"].as_str().unwrap(), "queued");
assert_eq!(body["status"].as_str().unwrap(), "submitted");
}
#[tokio::test]
@ -2529,18 +2567,8 @@ mod tests {
let state = create_app_state_with_options(initial_settings.clone(), 5);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({ "dot_source": dot })).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id_str = create_and_start_run(&app, &dot).await;
let run_id = run_id_str.parse::<RunId>().unwrap();
*state.settings.write().unwrap() = Settings::default();
@ -2562,7 +2590,7 @@ mod tests {
let state = create_app_state();
let app = build_router(state, AuthMode::Disabled);
// Submit a run (no scheduler, stays queued)
// Submit a run (no start, stays submitted)
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
@ -2610,18 +2638,8 @@ mod tests {
let state = create_app_state_with_options(settings, 5);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().unwrap();
let runner = tokio::spawn(execute_run(Arc::clone(&state), run_id));
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@ -2678,18 +2696,8 @@ mod tests {
});
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_id_str = create_and_start_run(&app, MINIMAL_DOT).await;
let run_id = run_id_str.parse::<RunId>().unwrap();
let runner = tokio::spawn(execute_run(Arc::clone(&state), run_id));
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
@ -2718,22 +2726,12 @@ mod tests {
let state = create_app_state();
let app = build_router(state, AuthMode::Disabled);
// Submit two runs (no scheduler, both stay queued)
// Create and start two runs (no scheduler, both stay queued)
let mut run_ids = Vec::new();
for _ in 0..2 {
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
run_ids.push(body["id"].as_str().unwrap().to_string());
}
let id = create_and_start_run(&app, MINIMAL_DOT).await;
run_ids.push(id);
let id = create_and_start_run(&app, MINIMAL_DOT).await;
run_ids.push(id);
// Check queue positions via individual status
let req = Request::builder()
@ -2760,22 +2758,12 @@ mod tests {
let state = create_app_state_with_options(Settings::default(), 1);
let app = test_app_with_scheduler(state);
// Submit two runs with max_concurrent_runs=1
// Create and start two runs with max_concurrent_runs=1
let mut run_ids = Vec::new();
for _ in 0..2 {
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
run_ids.push(body["id"].as_str().unwrap().to_string());
}
let id = create_and_start_run(&app, MINIMAL_DOT).await;
run_ids.push(id);
let id = create_and_start_run(&app, MINIMAL_DOT).await;
run_ids.push(id);
// Give scheduler time to pick up the first run
tokio::time::sleep(std::time::Duration::from_millis(50)).await;

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

@ -531,7 +531,7 @@ mod server_lifecycle {
fabro_server::jwt_auth::AuthMode::Disabled,
);
// 1. Start run
// 1. Create run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
@ -546,6 +546,15 @@ mod server_lifecycle {
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().to_string();
// 1b. Start the run
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
// 2. Poll for question to appear (run goes start -> work -> gate, then blocks)
let question_id = wait_for_question_id(&app, &run_id).await;
@ -567,18 +576,7 @@ mod server_lifecycle {
let final_status = wait_for_run_status(&app, &run_id, &["completed", "failed"]).await;
assert_eq!(final_status, "completed");
// 5. Verify context endpoint returns an object
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/context")))
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let ctx_body = body_json(response.into_body()).await;
assert!(ctx_body.is_object(), "context should be an object");
// 6. Verify no pending questions
// 5. Verify no pending questions
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/questions")))
@ -601,7 +599,7 @@ mod server_lifecycle {
fabro_server::jwt_auth::AuthMode::Disabled,
);
// Start a run that will block at the human gate
// Create and start a run that will block at the human gate
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
@ -614,6 +612,13 @@ mod server_lifecycle {
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().to_string();
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.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.
@ -890,7 +895,7 @@ mod serve_dry_run {
async fn dry_run_serve_starts_and_runs_workflow() {
let app = dry_run_app().await;
// POST /runs to start a run
// POST /runs to create a run
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
@ -907,6 +912,15 @@ mod serve_dry_run {
let run_id = body["id"].as_str().unwrap().to_string();
assert!(!run_id.is_empty());
// POST /runs/{id}/start to queue it
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let status = wait_for_run_status(&app, &run_id, &["completed", "failed"]).await;
assert_eq!(status, "completed");
}

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

@ -42,6 +42,7 @@ models/control-info.ts
models/control-performance.ts
models/control-reference.ts
models/create-completion-request.ts
models/create-run-request.ts
models/create-session-request.ts
models/create-session-response.ts
models/create-signoff-request.ts
@ -143,7 +144,6 @@ models/smoothness-rating.ts
models/stage-retro.ts
models/stage-status.ts
models/stage-turn.ts
models/start-run-request.ts
models/steer-request.ts
models/submit-answer-request.ts
models/system-stage-turn.ts

75
lib/packages/fabro-api-client/src/api/run-internals-api.ts
View file

@ -183,47 +183,6 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
options: localVarRequestOptions,
};
},
/**
* Returns the key-value context map accumulated during the run. Empty if the run has not started.
* @summary Retrieve Run Context
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
retrieveRunContext: async (id: string, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('retrieveRunContext', 'id', id)
const localVarPath = `/api/v1/runs/{id}/context`
.replace(`{${"id"}}`, encodeURIComponent(String(id)));
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
if (configuration) {
baseOptions = configuration.baseOptions;
}
const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options};
const localVarHeaderParameter = {} as any;
const localVarQueryParameter = {} as any;
// authentication mTLS required
await setApiKeyToObject(localVarHeaderParameter, "X-mTLS-Client-CN", configuration)
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
localVarHeaderParameter['Accept'] = 'application/json';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Returns the structured settings used to launch this run.
* @summary Retrieve Run Settings
@ -318,19 +277,6 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.retrieveRunCheckpoint']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns the key-value context map accumulated during the run. Empty if the run has not started.
* @summary Retrieve Run Context
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async retrieveRunContext(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<{ [key: string]: any; }>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.retrieveRunContext(id, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.retrieveRunContext']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns the structured settings used to launch this run.
* @summary Retrieve Run Settings
@ -388,16 +334,6 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
retrieveRunCheckpoint(id: string, options?: RawAxiosRequestConfig): AxiosPromise<RunCheckpoint> {
return localVarFp.retrieveRunCheckpoint(id, options).then((request) => request(axios, basePath));
},
/**
* Returns the key-value context map accumulated during the run. Empty if the run has not started.
* @summary Retrieve Run Context
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
retrieveRunContext(id: string, options?: RawAxiosRequestConfig): AxiosPromise<{ [key: string]: any; }> {
return localVarFp.retrieveRunContext(id, options).then((request) => request(axios, basePath));
},
/**
* Returns the structured settings used to launch this run.
* @summary Retrieve Run Settings
@ -453,17 +389,6 @@ export class RunInternalsApi extends BaseAPI {
return RunInternalsApiFp(this.configuration).retrieveRunCheckpoint(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns the key-value context map accumulated during the run. Empty if the run has not started.
* @summary Retrieve Run Context
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public retrieveRunContext(id: string, options?: RawAxiosRequestConfig) {
return RunInternalsApiFp(this.configuration).retrieveRunContext(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns the structured settings used to launch this run.
* @summary Retrieve Run Settings

119
lib/packages/fabro-api-client/src/api/runs-api.ts
View file

@ -22,13 +22,13 @@ import { DUMMY_BASE_URL, assertParamExists, setApiKeyToObject, setBasicAuthToObj
// @ts-ignore
import { BASE_PATH, COLLECTION_FORMATS, type RequestArgs, BaseAPI, RequiredError, operationServerMap } from '../base';
// @ts-ignore
import type { CreateRunRequest } from '../models';
// @ts-ignore
import type { ErrorResponse } from '../models';
// @ts-ignore
import type { PaginatedRunList } from '../models';
// @ts-ignore
import type { RunStatusResponse } from '../models';
// @ts-ignore
import type { StartRunRequest } from '../models';
/**
* RunsApi - axios parameter creator
*/
@ -75,6 +75,48 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
options: localVarRequestOptions,
};
},
/**
* Creates a new workflow run from a Graphviz graph source. The run is created in `submitted` status. Use `POST /api/v1/runs/{id}/start` to begin execution.
* @summary Create Run
* @param {CreateRunRequest} createRunRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
createRun: async (createRunRequest: CreateRunRequest, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'createRunRequest' is not null or undefined
assertParamExists('createRun', 'createRunRequest', createRunRequest)
const localVarPath = `/api/v1/runs`;
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
if (configuration) {
baseOptions = configuration.baseOptions;
}
const localVarRequestOptions = { method: 'POST', ...baseOptions, ...options};
const localVarHeaderParameter = {} as any;
const localVarQueryParameter = {} as any;
// authentication mTLS required
await setApiKeyToObject(localVarHeaderParameter, "X-mTLS-Client-CN", configuration)
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
localVarHeaderParameter['Content-Type'] = 'application/json';
localVarHeaderParameter['Accept'] = 'application/json';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
localVarRequestOptions.data = serializeDataIfNeeded(createRunRequest, localVarRequestOptions, configuration)
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Returns a paginated list of runs for the board view, ordered by recency.
* @summary List Runs
@ -246,16 +288,17 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
};
},
/**
* Queues a new workflow run from a Graphviz graph source. The run is created in `queued` status and will be picked up by the scheduler.
* Starts a submitted run, queuing it for execution. Returns 409 if the run is not in `submitted` status.
* @summary Start Run
* @param {StartRunRequest} startRunRequest
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
startRun: async (startRunRequest: StartRunRequest, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'startRunRequest' is not null or undefined
assertParamExists('startRun', 'startRunRequest', startRunRequest)
const localVarPath = `/api/v1/runs`;
startRun: async (id: string, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('startRun', 'id', id)
const localVarPath = `/api/v1/runs/{id}/start`
.replace(`{${"id"}}`, encodeURIComponent(String(id)));
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
@ -274,13 +317,11 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
localVarHeaderParameter['Content-Type'] = 'application/json';
localVarHeaderParameter['Accept'] = 'application/json';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
localVarRequestOptions.data = serializeDataIfNeeded(startRunRequest, localVarRequestOptions, configuration)
return {
url: toPathString(localVarUrlObj),
@ -391,6 +432,19 @@ export const RunsApiFp = function(configuration?: Configuration) {
const localVarOperationServerBasePath = operationServerMap['RunsApi.cancelRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Creates a new workflow run from a Graphviz graph source. The run is created in `submitted` status. Use `POST /api/v1/runs/{id}/start` to begin execution.
* @summary Create Run
* @param {CreateRunRequest} createRunRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async createRun(createRunRequest: CreateRunRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<RunStatusResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.createRun(createRunRequest, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.createRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns a paginated list of runs for the board view, ordered by recency.
* @summary List Runs
@ -445,14 +499,14 @@ export const RunsApiFp = function(configuration?: Configuration) {
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Queues a new workflow run from a Graphviz graph source. The run is created in `queued` status and will be picked up by the scheduler.
* Starts a submitted run, queuing it for execution. Returns 409 if the run is not in `submitted` status.
* @summary Start Run
* @param {StartRunRequest} startRunRequest
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async startRun(startRunRequest: StartRunRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<RunStatusResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.startRun(startRunRequest, options);
async startRun(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<RunStatusResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.startRun(id, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.startRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
@ -502,6 +556,16 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
cancelRun(id: string, options?: RawAxiosRequestConfig): AxiosPromise<RunStatusResponse> {
return localVarFp.cancelRun(id, options).then((request) => request(axios, basePath));
},
/**
* Creates a new workflow run from a Graphviz graph source. The run is created in `submitted` status. Use `POST /api/v1/runs/{id}/start` to begin execution.
* @summary Create Run
* @param {CreateRunRequest} createRunRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
createRun(createRunRequest: CreateRunRequest, options?: RawAxiosRequestConfig): AxiosPromise<RunStatusResponse> {
return localVarFp.createRun(createRunRequest, options).then((request) => request(axios, basePath));
},
/**
* Returns a paginated list of runs for the board view, ordered by recency.
* @summary List Runs
@ -544,14 +608,14 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
return localVarFp.retrieveRunGraph(id, options).then((request) => request(axios, basePath));
},
/**
* Queues a new workflow run from a Graphviz graph source. The run is created in `queued` status and will be picked up by the scheduler.
* Starts a submitted run, queuing it for execution. Returns 409 if the run is not in `submitted` status.
* @summary Start Run
* @param {StartRunRequest} startRunRequest
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
startRun(startRunRequest: StartRunRequest, options?: RawAxiosRequestConfig): AxiosPromise<RunStatusResponse> {
return localVarFp.startRun(startRunRequest, options).then((request) => request(axios, basePath));
startRun(id: string, options?: RawAxiosRequestConfig): AxiosPromise<RunStatusResponse> {
return localVarFp.startRun(id, options).then((request) => request(axios, basePath));
},
/**
* Opens a server-sent event (SSE) stream for real-time run updates. Returns 410 if the stream has been closed.
@ -591,6 +655,17 @@ export class RunsApi extends BaseAPI {
return RunsApiFp(this.configuration).cancelRun(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Creates a new workflow run from a Graphviz graph source. The run is created in `submitted` status. Use `POST /api/v1/runs/{id}/start` to begin execution.
* @summary Create Run
* @param {CreateRunRequest} createRunRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public createRun(createRunRequest: CreateRunRequest, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).createRun(createRunRequest, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns a paginated list of runs for the board view, ordered by recency.
* @summary List Runs
@ -637,14 +712,14 @@ export class RunsApi extends BaseAPI {
}
/**
* Queues a new workflow run from a Graphviz graph source. The run is created in `queued` status and will be picked up by the scheduler.
* Starts a submitted run, queuing it for execution. Returns 409 if the run is not in `submitted` status.
* @summary Start Run
* @param {StartRunRequest} startRunRequest
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public startRun(startRunRequest: StartRunRequest, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).startRun(startRunRequest, options).then((request) => request(this.axios, this.basePath));
public startRun(id: string, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).startRun(id, options).then((request) => request(this.axios, this.basePath));
}
/**

26
lib/packages/fabro-api-client/src/models/create-run-request.ts Normal file
View file

@ -0,0 +1,26 @@
/* 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.
*/
/**
* Request body for creating a new run from a Graphviz graph source.
*/
export interface CreateRunRequest {
/**
* Graphviz DOT language source defining the workflow graph.
*/
'dot_source': string;
}

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

@ -23,6 +23,7 @@ export * from './control-info';
export * from './control-performance';
export * from './control-reference';
export * from './create-completion-request';
export * from './create-run-request';
export * from './create-session-request';
export * from './create-session-response';
export * from './create-signoff-request';
@ -123,7 +124,6 @@ export * from './smoothness-rating';
export * from './stage-retro';
export * from './stage-status';
export * from './stage-turn';
export * from './start-run-request';
export * from './steer-request';
export * from './submit-answer-request';
export * from './system-stage-turn';

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

@ -19,6 +19,7 @@
*/
export const RunStatus = {
SUBMITTED: 'submitted',
QUEUED: 'queued',
STARTING: 'starting',
RUNNING: 'running',