diff --git a/apps/fabro-web/app/data/runs.ts b/apps/fabro-web/app/data/runs.ts index eb9c30745..c0a87f6b3 100644 --- a/apps/fabro-web/app/data/runs.ts +++ b/apps/fabro-web/app/data/runs.ts @@ -29,15 +29,12 @@ export interface RunItem { sandboxId?: string; } -export type ColumnStatus = "working" | "initializing" | "review" | "merge" | "running" | "waiting" | "succeeded" | "failed"; +export type ColumnStatus = "initializing" | "running" | "blocked" | "succeeded" | "failed"; export const columnNames: Record = { - working: "Working", initializing: "Initializing", - review: "Verify", - merge: "Merge", running: "Running", - waiting: "Waiting", + blocked: "Blocked", succeeded: "Succeeded", failed: "Failed", }; @@ -112,20 +109,19 @@ export function deriveCiStatus(checks: CheckRun[]): CiStatus { } export const statusColors: Record = { - working: { dot: "bg-teal-500", text: "text-teal-500" }, initializing: { dot: "bg-amber", text: "text-amber" }, - review: { dot: "bg-mint", text: "text-mint" }, - merge: { dot: "bg-teal-300", text: "text-teal-300" }, running: { dot: "bg-teal-500", text: "text-teal-500" }, - waiting: { dot: "bg-amber", text: "text-amber" }, + blocked: { dot: "bg-amber", text: "text-amber" }, succeeded: { dot: "bg-teal-300", text: "text-teal-300" }, failed: { dot: "bg-coral", text: "text-coral" }, }; export type RunStatus = | "submitted" + | "queued" | "starting" | "running" + | "blocked" | "paused" | "removing" | "succeeded" @@ -134,8 +130,10 @@ export type RunStatus = export const runStatusDisplay: Record = { submitted: { label: "Submitted", dot: "bg-fg-muted", text: "text-fg-muted" }, + queued: { label: "Queued", dot: "bg-fg-muted", text: "text-fg-muted" }, starting: { label: "Starting", dot: "bg-amber", text: "text-amber" }, running: { label: "Running", dot: "bg-teal-500", text: "text-teal-500" }, + blocked: { label: "Blocked", dot: "bg-amber", text: "text-amber" }, paused: { label: "Paused", dot: "bg-amber", text: "text-amber" }, removing: { label: "Removing", dot: "bg-fg-muted", text: "text-fg-muted" }, succeeded: { label: "Succeeded", dot: "bg-mint", text: "text-mint" }, @@ -160,4 +158,4 @@ export const ciConfig: Record { @@ -803,4 +807,4 @@ export default function Runs({ loaderData }: any) { ); -} +} \ No newline at end of file diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index 3228cbefd..d610ddbd7 100644 --- a/docs/api-reference/fabro-api.yaml +++ b/docs/api-reference/fabro-api.yaml @@ -2089,10 +2089,18 @@ components: - queued - starting - running - - completed - - failed - - cancelled + - blocked - paused + - removing + - succeeded + - failed + - dead + + BlockedReason: + description: Reason a run is blocked. + type: string + enum: + - human_input_required RunManifest: description: Self-contained workflow run manifest. @@ -2481,6 +2489,10 @@ components: allOf: - $ref: "#/components/schemas/StatusReason" nullable: true + blocked_reason: + allOf: + - $ref: "#/components/schemas/BlockedReason" + nullable: true pending_control: allOf: - $ref: "#/components/schemas/RunControlAction" @@ -2871,19 +2883,6 @@ components: items: $ref: "#/components/schemas/RunArtifactEntry" - InternalRunStatus: - description: Internal event-sourced run status. - type: string - enum: - - submitted - - starting - - running - - paused - - removing - - succeeded - - failed - - dead - StatusReason: description: Optional reason attached to a run status transition. type: string @@ -2916,11 +2915,15 @@ components: - updated_at properties: status: - $ref: "#/components/schemas/InternalRunStatus" - reason: + $ref: "#/components/schemas/RunStatus" + status_reason: oneOf: - $ref: "#/components/schemas/StatusReason" - type: "null" + blocked_reason: + oneOf: + - $ref: "#/components/schemas/BlockedReason" + - type: "null" updated_at: type: string format: date-time @@ -3047,6 +3050,10 @@ components: type: object additionalProperties: true nullable: true + pending_interviews: + type: object + additionalProperties: true + description: Map from question ID to pending interview record. nodes: type: object description: Map from StageId (`node_id@visit`) to NodeState. @@ -3058,6 +3065,7 @@ components: type: object required: - run_id + - status - labels properties: run_id: @@ -3083,10 +3091,14 @@ components: format: date-time nullable: true status: - type: string - nullable: true + $ref: "#/components/schemas/RunStatus" status_reason: - type: string + allOf: + - $ref: "#/components/schemas/StatusReason" + nullable: true + blocked_reason: + allOf: + - $ref: "#/components/schemas/BlockedReason" nullable: true pending_control: allOf: @@ -3108,10 +3120,11 @@ components: description: Board column status for a run in the list view. type: string enum: - - working - initializing - - review - - merge + - running + - blocked + - succeeded + - failed CheckRunStatus: description: Status of a CI check run. @@ -4525,4 +4538,4 @@ components: login: type: string description: User's login identifier (e.g. GitHub username). - example: octocat + example: octocat \ No newline at end of file diff --git a/lib/crates/fabro-cli/src/commands/runs/list.rs b/lib/crates/fabro-cli/src/commands/runs/list.rs index dafe0d418..7aa67c30b 100644 --- a/lib/crates/fabro-cli/src/commands/runs/list.rs +++ b/lib/crates/fabro-cli/src/commands/runs/list.rs @@ -148,8 +148,10 @@ fn status_cell(status: RunStatus, use_color: bool) -> CellStruct { let color = match status { RunStatus::Succeeded => Some(Color::Green), RunStatus::Failed => Some(Color::Red), - RunStatus::Running | RunStatus::Starting | RunStatus::Submitted => Some(Color::Cyan), - RunStatus::Removing => Some(Color::Yellow), + RunStatus::Running | RunStatus::Starting | RunStatus::Submitted | RunStatus::Queued => { + Some(Color::Cyan) + } + RunStatus::Blocked | RunStatus::Removing => Some(Color::Yellow), RunStatus::Paused => Some(Color::Magenta), RunStatus::Dead => Some(Color::Ansi256(8)), }; diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 8700215ef..01f92758f 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -372,7 +372,8 @@ mod tests { fn sample_status() -> RunStatusRecord { RunStatusRecord { status: RunStatus::Running, - reason: Some(StatusReason::SandboxInitializing), + status_reason: Some(StatusReason::SandboxInitializing), + blocked_reason: None, updated_at: dt("2026-03-27T12:05:00Z"), } } @@ -526,7 +527,7 @@ mod tests { &run, &run_id, &Event::RunRunning { - reason: status_record.reason, + reason: status_record.status_reason, }, ) .await diff --git a/lib/crates/fabro-cli/src/server_runs.rs b/lib/crates/fabro-cli/src/server_runs.rs index f3ef05b79..d5a303332 100644 --- a/lib/crates/fabro-cli/src/server_runs.rs +++ b/lib/crates/fabro-cli/src/server_runs.rs @@ -63,7 +63,7 @@ impl ServerRunSummaryInfo { } pub(crate) fn status(&self) -> RunStatus { - self.summary.status.unwrap_or(RunStatus::Dead) + self.summary.status.unwrap_or(RunStatus::Submitted) } pub(crate) fn status_reason(&self) -> Option { diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index bfd19be7b..3db7b079d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -823,6 +823,15 @@ fn attach_json_errors_without_prompting_for_human_input() { "run_id": "[ULID]", "stage_id": "approve@1", "ts": "[TIMESTAMP]" + }, + { + "event": "run.blocked", + "id": "[EVENT_ID]", + "properties": { + "blocked_reason": "human_input_required" + }, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" } ] "#); diff --git a/lib/crates/fabro-cli/tests/it/cmd/runner.rs b/lib/crates/fabro-cli/tests/it/cmd/runner.rs index 9f822804d..8aa173f84 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/runner.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/runner.rs @@ -746,5 +746,5 @@ fn worker_exits_after_sigterm_cancel_even_when_stdin_stays_open() { .status .expect("cancelled run should have a status record"); assert_eq!(status_record.status.to_string(), "failed"); - assert_eq!(status_record.reason, Some(StatusReason::Cancelled)); + assert_eq!(status_record.status_reason, Some(StatusReason::Cancelled)); } diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index 869502184..02432db29 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -683,7 +683,7 @@ mod runs { workflow: WorkflowReference { slug: "implement".into(), }, - status: BoardColumn::Working, + status: BoardColumn::Running, pull_request: None, timings: Some(RunTimings { elapsed_secs: 420.0, @@ -705,7 +705,7 @@ mod runs { workflow: WorkflowReference { slug: "implement".into(), }, - status: BoardColumn::Working, + status: BoardColumn::Running, pull_request: None, timings: Some(RunTimings { elapsed_secs: 8100.0, @@ -727,7 +727,7 @@ mod runs { workflow: WorkflowReference { slug: "fix_build".into(), }, - status: BoardColumn::Working, + status: BoardColumn::Running, pull_request: None, timings: Some(RunTimings { elapsed_secs: 2700.0, @@ -809,7 +809,7 @@ mod runs { workflow: WorkflowReference { slug: "implement".into(), }, - status: BoardColumn::Review, + status: BoardColumn::Succeeded, pull_request: Some(RunPullRequest { number: 889, additions: Some(234), @@ -873,7 +873,7 @@ mod runs { workflow: WorkflowReference { slug: "implement".into(), }, - status: BoardColumn::Review, + status: BoardColumn::Succeeded, pull_request: Some(RunPullRequest { number: 156, additions: Some(412), @@ -927,7 +927,7 @@ mod runs { workflow: WorkflowReference { slug: "implement".into(), }, - status: BoardColumn::Merge, + status: BoardColumn::Succeeded, pull_request: Some(RunPullRequest { number: 1249, additions: Some(189), @@ -1016,7 +1016,7 @@ mod runs { workflow: WorkflowReference { slug: "expand".into(), }, - status: BoardColumn::Merge, + status: BoardColumn::Succeeded, pull_request: Some(RunPullRequest { number: 430, additions: Some(56), @@ -1075,7 +1075,7 @@ mod runs { workflow: WorkflowReference { slug: "sync_drift".into(), }, - status: BoardColumn::Merge, + status: BoardColumn::Succeeded, pull_request: Some(RunPullRequest { number: 76, additions: Some(34), diff --git a/lib/crates/fabro-server/src/serve.rs b/lib/crates/fabro-server/src/serve.rs index afff70b9c..baf5c9686 100644 --- a/lib/crates/fabro-server/src/serve.rs +++ b/lib/crates/fabro-server/src/serve.rs @@ -729,6 +729,7 @@ mod tests { sandbox: None, web: false, no_web: false, + watch_web: false, max_concurrent_runs: None, config: None, }; @@ -761,6 +762,7 @@ enabled = false sandbox: None, web: true, no_web: false, + watch_web: false, max_concurrent_runs: None, config: None, }; @@ -787,6 +789,7 @@ enabled = false sandbox: None, web: false, no_web: true, + watch_web: false, max_concurrent_runs: None, config: None, }; diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 8f6050f23..081b625b5 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -2078,7 +2078,7 @@ async fn list_run_stages( Some(managed_run) => { let active = !matches!( managed_run.status, - RunStatus::Completed | RunStatus::Failed | RunStatus::Cancelled + RunStatus::Succeeded | RunStatus::Failed | RunStatus::Dead ); (managed_run.checkpoint.clone(), active) } @@ -2533,9 +2533,11 @@ fn test_secret_store_path() -> PathBuf { fn board_column(status: WorkflowRunStatus) -> Option<&'static str> { match status { - WorkflowRunStatus::Submitted | WorkflowRunStatus::Starting => Some("initializing"), - WorkflowRunStatus::Running => Some("running"), - WorkflowRunStatus::Paused => Some("waiting"), + WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting => { + Some("initializing") + } + WorkflowRunStatus::Running | WorkflowRunStatus::Paused => Some("running"), + WorkflowRunStatus::Blocked => Some("blocked"), WorkflowRunStatus::Succeeded => Some("succeeded"), WorkflowRunStatus::Failed | WorkflowRunStatus::Dead => Some("failed"), WorkflowRunStatus::Removing => None, @@ -2546,7 +2548,7 @@ fn board_columns() -> serde_json::Value { serde_json::json!([ {"id": "initializing", "name": "Initializing"}, {"id": "running", "name": "Running"}, - {"id": "waiting", "name": "Waiting"}, + {"id": "blocked", "name": "Blocked"}, {"id": "succeeded", "name": "Succeeded"}, {"id": "failed", "name": "Failed"}, ]) @@ -2568,32 +2570,54 @@ async fn list_board_runs( .into_response(); } }; - let all_items: Vec = summaries - .into_iter() - .filter_map(|summary| { - let status = summary.status?; - let column = board_column(status)?; - let title = summary.goal.as_deref().unwrap_or("Untitled run"); - let workflow_slug = summary.workflow_slug.as_deref().unwrap_or("unknown"); - let workflow_name = summary.workflow_name.as_deref().unwrap_or(workflow_slug); - let repo_name = summary - .host_repo_path - .as_deref() - .and_then(|p| p.rsplit('/').next()) - .unwrap_or("unknown"); - let elapsed_secs = summary.duration_ms.map(|ms| ms as f64 / 1000.0); - let created_at = summary.run_id.created_at(); - Some(serde_json::json!({ - "id": summary.run_id.to_string(), - "title": title, - "repository": { "name": repo_name }, - "workflow": { "slug": workflow_slug, "name": workflow_name }, - "status": column, - "created_at": created_at.to_rfc3339(), - "timings": elapsed_secs.map(|s| serde_json::json!({ "elapsed_secs": s })), - })) - }) - .collect(); + let mut all_items: Vec = Vec::new(); + for summary in summaries { + let Some(status) = summary.status else { + continue; + }; + let Some(column) = board_column(status) else { + continue; + }; + let title = summary.goal.as_deref().unwrap_or("Untitled run"); + let workflow_slug = summary.workflow_slug.as_deref().unwrap_or("unknown"); + let workflow_name = summary.workflow_name.as_deref().unwrap_or(workflow_slug); + let repo_name = summary + .host_repo_path + .as_deref() + .and_then(|p| p.rsplit('/').next()) + .unwrap_or("unknown"); + let elapsed_secs = summary.duration_ms.map(|ms| ms as f64 / 1000.0); + let created_at = summary.run_id.created_at(); + + // Populate question text for blocked runs from pending interviews. + let question = if column == "blocked" { + match state.store.open_run_reader(&summary.run_id).await { + Ok(reader) => match reader.state().await { + Ok(proj) => proj + .pending_interviews + .values() + .filter_map(|pi| pi.started_at.map(|ts| (ts, &pi.question.text))) + .min_by_key(|(ts, _)| *ts) + .map(|(_, text)| serde_json::json!({ "text": text })), + Err(_) => None, + }, + Err(_) => None, + } + } else { + None + }; + + all_items.push(serde_json::json!({ + "id": summary.run_id.to_string(), + "title": title, + "repository": { "name": repo_name }, + "workflow": { "slug": workflow_slug, "name": workflow_name }, + "status": column, + "created_at": created_at.to_rfc3339(), + "timings": elapsed_secs.map(|s| serde_json::json!({ "elapsed_secs": s })), + "question": question, + })); + } let limit = pagination.limit.clamp(1, 100) as usize; let offset = pagination.offset as usize; let page: Vec<_> = all_items.into_iter().skip(offset).take(limit + 1).collect(); @@ -2942,8 +2966,10 @@ fn failure_for_incomplete_run( fn should_reconcile_run_on_startup(status: WorkflowRunStatus) -> bool { matches!( status, - WorkflowRunStatus::Starting + WorkflowRunStatus::Queued + | WorkflowRunStatus::Starting | WorkflowRunStatus::Running + | WorkflowRunStatus::Blocked | WorkflowRunStatus::Paused | WorkflowRunStatus::Removing ) @@ -3161,20 +3187,18 @@ fn managed_run( } } -fn api_status_from_workflow( - status: WorkflowRunStatus, - reason: Option, -) -> RunStatus { +fn api_status_from_workflow(status: WorkflowRunStatus) -> RunStatus { match status { WorkflowRunStatus::Submitted => RunStatus::Submitted, + WorkflowRunStatus::Queued => RunStatus::Queued, WorkflowRunStatus::Starting => RunStatus::Starting, - WorkflowRunStatus::Running | WorkflowRunStatus::Removing => RunStatus::Running, + WorkflowRunStatus::Running => RunStatus::Running, + WorkflowRunStatus::Blocked => RunStatus::Blocked, WorkflowRunStatus::Paused => RunStatus::Paused, - WorkflowRunStatus::Succeeded => RunStatus::Completed, - WorkflowRunStatus::Failed if reason == Some(WorkflowStatusReason::Cancelled) => { - RunStatus::Cancelled - } - WorkflowRunStatus::Failed | WorkflowRunStatus::Dead => RunStatus::Failed, + WorkflowRunStatus::Removing => RunStatus::Removing, + WorkflowRunStatus::Succeeded => RunStatus::Succeeded, + WorkflowRunStatus::Failed => RunStatus::Failed, + WorkflowRunStatus::Dead => RunStatus::Dead, } } @@ -3252,21 +3276,24 @@ fn update_live_run_from_event(state: &Arc, run_id: RunId, event: &RunE }; match &event.body { + EventBody::RunQueued(_) => managed_run.status = RunStatus::Queued, EventBody::RunStarting(_) => managed_run.status = RunStatus::Starting, EventBody::RunRunning(_) | EventBody::RunUnpaused(_) => { managed_run.status = RunStatus::Running; } + EventBody::RunBlocked(_) => managed_run.status = RunStatus::Blocked, + EventBody::RunUnblocked(_) => { + if managed_run.status == RunStatus::Blocked { + managed_run.status = RunStatus::Running; + } + } EventBody::RunPaused(_) => managed_run.status = RunStatus::Paused, EventBody::RunCompleted(_) => { - managed_run.status = RunStatus::Completed; + managed_run.status = RunStatus::Succeeded; managed_run.error = None; } EventBody::RunFailed(props) => { - managed_run.status = if props.reason == Some(WorkflowStatusReason::Cancelled) { - RunStatus::Cancelled - } else { - RunStatus::Failed - }; + managed_run.status = RunStatus::Failed; managed_run.error = Some(props.error.clone()); } _ => {} @@ -3707,6 +3734,7 @@ async fn create_run( error: None, queue_position: None, status_reason: None, + blocked_reason: None, pending_control: None, created_at, }), @@ -3930,6 +3958,7 @@ async fn start_run( error: None, queue_position: None, status_reason: None, + blocked_reason: None, pending_control: None, created_at: id.created_at(), }), @@ -4196,11 +4225,11 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { Ok(started) => match &started.finalized.outcome { Ok(_) => { info!(run_id = %run_id, "Run completed"); - managed_run.status = RunStatus::Completed; + managed_run.status = RunStatus::Succeeded; } Err(WorkflowError::Cancelled) => { info!(run_id = %run_id, "Run cancelled"); - managed_run.status = RunStatus::Cancelled; + managed_run.status = RunStatus::Failed; } Err(e) => { error!(run_id = %run_id, error = %e, "Run failed"); @@ -4210,7 +4239,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { }, Err(WorkflowError::Cancelled) => { info!(run_id = %run_id, "Run cancelled"); - managed_run.status = RunStatus::Cancelled; + managed_run.status = RunStatus::Failed; } Err(e) => { error!(run_id = %run_id, error = %e, "Run failed"); @@ -4220,7 +4249,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { }, ExecutionResult::CancelledBySignal => { info!(run_id = %run_id, "Run cancelled"); - managed_run.status = RunStatus::Cancelled; + managed_run.status = RunStatus::Failed; } } managed_run.checkpoint = checkpoint; @@ -4462,7 +4491,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { let mut runs = state.runs.lock().expect("runs lock poisoned"); if let Some(managed_run) = runs.get_mut(&run_id) { if let Some(status) = final_state.status.as_ref() { - managed_run.status = api_status_from_workflow(status.status, status.reason); + managed_run.status = api_status_from_workflow(status.status); } else if !wait_status.success() { managed_run.status = RunStatus::Failed; } @@ -5719,6 +5748,13 @@ async fn append_control_request( workflow_event::append_event(&run_store, &run_id, &event).await } +/// Append a `run.paused` event directly (for immediate pause from blocked). +async fn append_run_paused_event(state: &AppState, run_id: RunId) -> anyhow::Result<()> { + let run_store = state.store.open_run(&run_id).await?; + let event = workflow_event::Event::RunPaused; + workflow_event::append_event(&run_store, &run_id, &event).await +} + fn actor_from_subject(subject: &AuthenticatedSubject) -> Option { subject.login.clone().map(ActorRef::user) } @@ -5769,6 +5805,7 @@ async fn cancel_run( | RunStatus::Queued | RunStatus::Starting | RunStatus::Running + | RunStatus::Blocked | RunStatus::Paused => { let use_cancel_signal = !matches!( managed_run.answer_transport, @@ -5777,8 +5814,8 @@ async fn cancel_run( let persist_cancelled_status = matches!(managed_run.status, RunStatus::Submitted | RunStatus::Queued); let response_status = if persist_cancelled_status { - managed_run.status = RunStatus::Cancelled; - RunStatus::Cancelled + managed_run.status = RunStatus::Failed; + RunStatus::Failed } else { managed_run.status }; @@ -5854,6 +5891,7 @@ async fn cancel_run( error: None, queue_position: None, status_reason, + blocked_reason: None, pending_control, created_at, }), @@ -5877,11 +5915,18 @@ async fn pause_run( .into_response(); } }; - let (created_at, worker_pid) = { + let (created_at, current_status, worker_pid) = { let runs = state.runs.lock().expect("runs lock poisoned"); match runs.get(&id) { - Some(managed_run) if managed_run.status == RunStatus::Running => { - (managed_run.created_at, managed_run.worker_pid) + Some(managed_run) + if managed_run.status == RunStatus::Running + || managed_run.status == RunStatus::Blocked => + { + ( + managed_run.created_at, + managed_run.status, + managed_run.worker_pid, + ) } Some(_) => { return ApiError::new(StatusCode::CONFLICT, "Run is not pausable.").into_response(); @@ -5897,6 +5942,48 @@ async fn pause_run( ) .into_response(); } + + // Immediate pause from blocked: append pause.requested + paused directly + if current_status == RunStatus::Blocked { + if let Err(err) = append_control_request( + state.as_ref(), + id, + RunControlAction::Pause, + actor_from_subject(&subject), + ) + .await + { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + if let Err(err) = append_run_paused_event(state.as_ref(), id).await { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + if let Some(managed_run) = runs.get_mut(&id) { + managed_run.status = RunStatus::Paused; + } + } + let (status_reason, pending_control) = load_run_status_metadata(state.as_ref(), id).await; + return ( + StatusCode::OK, + Json(RunStatusResponse { + id: id.to_string(), + status: RunStatus::Paused, + error: None, + queue_position: None, + status_reason, + blocked_reason: None, + pending_control, + created_at, + }), + ) + .into_response(); + } + + // Cooperative pause from running let Some(worker_pid) = worker_pid else { return ApiError::new(StatusCode::CONFLICT, "Run worker is not available.").into_response(); }; @@ -5922,6 +6009,7 @@ async fn pause_run( error: None, queue_position: None, status_reason, + blocked_reason: None, pending_control, created_at, }), @@ -5990,6 +6078,7 @@ async fn unpause_run( error: None, queue_position: None, status_reason, + blocked_reason: None, pending_control, created_at, }), @@ -8568,7 +8657,7 @@ level = "debug" let run_store = state.store.open_run_reader(&run_id).await.unwrap(); let status = run_store.state().await.unwrap().status.unwrap(); assert_eq!(status.status, WorkflowRunStatus::Failed); - assert_eq!(status.reason, Some(WorkflowStatusReason::Cancelled)); + assert_eq!(status.status_reason, Some(WorkflowStatusReason::Cancelled)); } #[tokio::test] @@ -8778,7 +8867,10 @@ level = "debug" .unwrap(); let run_2_status = run_2.status.unwrap(); assert_eq!(run_2_status.status, WorkflowRunStatus::Failed); - assert_eq!(run_2_status.reason, Some(WorkflowStatusReason::Terminated)); + assert_eq!( + run_2_status.status_reason, + Some(WorkflowStatusReason::Terminated) + ); let run_3 = state .store @@ -8790,7 +8882,10 @@ level = "debug" .unwrap(); let run_3_status = run_3.status.unwrap(); assert_eq!(run_3_status.status, WorkflowRunStatus::Failed); - assert_eq!(run_3_status.reason, Some(WorkflowStatusReason::Cancelled)); + assert_eq!( + run_3_status.status_reason, + Some(WorkflowStatusReason::Cancelled) + ); assert_eq!(run_3.pending_control, None); } @@ -8866,7 +8961,10 @@ level = "debug" .unwrap(); let run_status = run_state.status.unwrap(); assert_eq!(run_status.status, WorkflowRunStatus::Failed); - assert_eq!(run_status.reason, Some(WorkflowStatusReason::Terminated)); + assert_eq!( + run_status.status_reason, + Some(WorkflowStatusReason::Terminated) + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -8906,7 +9004,7 @@ timeout = "30s" let runs = state.runs.lock().expect("runs lock poisoned"); let managed_run = runs.get(&run_id).expect("run should exist"); - assert_eq!(managed_run.status, RunStatus::Cancelled); + assert_eq!(managed_run.status, RunStatus::Failed); drop(runs); let run_store = state.store.open_run_reader(&run_id).await.unwrap(); @@ -8915,7 +9013,7 @@ timeout = "30s" for _ in 0..50 { if let Some(record) = run_store.state().await.unwrap().status { if record.status == WorkflowRunStatus::Failed - && record.reason == Some(WorkflowStatusReason::Cancelled) + && record.status_reason == Some(WorkflowStatusReason::Cancelled) { status_record = Some(record); break; @@ -8926,7 +9024,10 @@ timeout = "30s" let status_record = status_record.expect("status record should be persisted"); assert_eq!(status_record.status, WorkflowRunStatus::Failed); - assert_eq!(status_record.reason, Some(WorkflowStatusReason::Cancelled)); + assert_eq!( + status_record.status_reason, + Some(WorkflowStatusReason::Cancelled) + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -9267,7 +9368,7 @@ timeout = "30s" .iter() .find(|i| i["id"].as_str() == Some(&paused_id.to_string())) .expect("paused run should be on board"); - assert_eq!(paused_item["status"].as_str().unwrap(), "waiting"); + assert_eq!(paused_item["status"].as_str().unwrap(), "running"); let succeeded_item = data .iter() @@ -9278,7 +9379,7 @@ timeout = "30s" // Verify columns are included in the response let columns = body["columns"].as_array().expect("columns should be array"); assert!(columns.len() > 0); - assert!(columns.iter().any(|c| c["id"].as_str() == Some("waiting"))); + assert!(columns.iter().any(|c| c["id"].as_str() == Some("blocked"))); assert!( columns .iter() diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 4c73de2fc..9dc89faf4 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -132,13 +132,36 @@ impl RunProjection { self.pending_control = Some(RunControlAction::Unpause); } EventBody::RunPaused(_) => { - self.status = Some(run_status_record(RunStatus::Paused, None, ts)); + // Preserve blocked_reason while paused over an unresolved block. + let prev_blocked_reason = self.status.as_ref().and_then(|s| s.blocked_reason); + let mut record = run_status_record(RunStatus::Paused, None, ts); + record.blocked_reason = prev_blocked_reason; + self.status = Some(record); self.pending_control = None; } EventBody::RunUnpaused(_) => { self.status = Some(run_status_record(RunStatus::Running, None, ts)); self.pending_control = None; } + EventBody::RunQueued(props) => { + self.status = Some(run_status_record(RunStatus::Queued, props.reason, ts)); + } + EventBody::RunBlocked(props) => { + let mut record = run_status_record(RunStatus::Blocked, None, ts); + record.blocked_reason = Some(props.blocked_reason); + self.status = Some(record); + } + EventBody::RunUnblocked(_) => { + // Clear blocked_reason. If currently blocked, restore Running. + // If currently paused (block resolved while paused), keep Paused. + if let Some(ref mut status) = self.status { + status.blocked_reason = None; + if status.status == RunStatus::Blocked { + status.status = RunStatus::Running; + status.updated_at = ts; + } + } + } EventBody::RunCompleted(props) => { self.status = Some(run_status_record(RunStatus::Succeeded, props.reason, ts)); self.pending_control = None; @@ -385,8 +408,16 @@ impl RunProjection { .unwrap_or_default(), host_repo_path: self.run.as_ref().and_then(|run| run.host_repo_path.clone()), 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), + status: Some( + self.status + .as_ref() + .map_or(RunStatus::Submitted, |status| status.status), + ), + status_reason: self.status.as_ref().and_then(|status| status.status_reason), + blocked_reason: self + .status + .as_ref() + .and_then(|status| status.blocked_reason), pending_control: self.pending_control, duration_ms: self .conclusion @@ -436,7 +467,8 @@ fn run_status_record( ) -> RunStatusRecord { RunStatusRecord { status, - reason, + status_reason: reason, + blocked_reason: None, updated_at, } } @@ -856,4 +888,143 @@ mod tests { events[1].payload.as_value()["properties"]["definition_blob"] ); } + + #[test] + fn run_queued_sets_status() { + use fabro_types::run_event::RunStatusTransitionProps; + + let events = vec![test_event( + 1, + EventBody::RunQueued(RunStatusTransitionProps { reason: None }), + None, + )]; + let state = RunProjection::apply_events(&events).unwrap(); + assert_eq!( + state.status.as_ref().unwrap().status, + fabro_types::RunStatus::Queued + ); + } + + #[test] + fn run_blocked_sets_status_and_blocked_reason() { + use fabro_types::run_event::{RunBlockedProps, RunStatusTransitionProps}; + use fabro_types::{BlockedReason, RunStatus}; + + let events = vec![ + test_event( + 1, + EventBody::RunRunning(RunStatusTransitionProps { reason: None }), + None, + ), + test_event( + 2, + EventBody::RunBlocked(RunBlockedProps { + blocked_reason: BlockedReason::HumanInputRequired, + }), + None, + ), + ]; + let state = RunProjection::apply_events(&events).unwrap(); + let status = state.status.as_ref().unwrap(); + assert_eq!(status.status, RunStatus::Blocked); + assert_eq!( + status.blocked_reason, + Some(BlockedReason::HumanInputRequired) + ); + } + + #[test] + fn run_unblocked_clears_blocked_reason_and_restores_running() { + use fabro_types::run_event::{ + RunBlockedProps, RunStatusTransitionProps, RunUnblockedProps, + }; + use fabro_types::{BlockedReason, RunStatus}; + + let events = vec![ + test_event( + 1, + EventBody::RunRunning(RunStatusTransitionProps { reason: None }), + None, + ), + test_event( + 2, + EventBody::RunBlocked(RunBlockedProps { + blocked_reason: BlockedReason::HumanInputRequired, + }), + None, + ), + test_event(3, EventBody::RunUnblocked(RunUnblockedProps {}), None), + ]; + let state = RunProjection::apply_events(&events).unwrap(); + let status = state.status.as_ref().unwrap(); + assert_eq!(status.status, RunStatus::Running); + assert_eq!(status.blocked_reason, None); + } + + #[test] + fn paused_over_blocked_preserves_blocked_reason() { + use fabro_types::run_event::{ + RunBlockedProps, RunControlEffectProps, RunStatusTransitionProps, + }; + use fabro_types::{BlockedReason, RunStatus}; + + let events = vec![ + test_event( + 1, + EventBody::RunRunning(RunStatusTransitionProps { reason: None }), + None, + ), + test_event( + 2, + EventBody::RunBlocked(RunBlockedProps { + blocked_reason: BlockedReason::HumanInputRequired, + }), + None, + ), + test_event(3, EventBody::RunPaused(RunControlEffectProps {}), None), + ]; + let state = RunProjection::apply_events(&events).unwrap(); + let status = state.status.as_ref().unwrap(); + assert_eq!(status.status, RunStatus::Paused); + assert_eq!( + status.blocked_reason, + Some(BlockedReason::HumanInputRequired) + ); + } + + #[test] + fn unblocked_while_paused_clears_blocked_reason_keeps_paused() { + use fabro_types::run_event::{ + RunBlockedProps, RunControlEffectProps, RunStatusTransitionProps, RunUnblockedProps, + }; + use fabro_types::{BlockedReason, RunStatus}; + + let events = vec![ + test_event( + 1, + EventBody::RunRunning(RunStatusTransitionProps { reason: None }), + None, + ), + test_event( + 2, + EventBody::RunBlocked(RunBlockedProps { + blocked_reason: BlockedReason::HumanInputRequired, + }), + None, + ), + test_event(3, EventBody::RunPaused(RunControlEffectProps {}), None), + test_event(4, EventBody::RunUnblocked(RunUnblockedProps {}), None), + ]; + let state = RunProjection::apply_events(&events).unwrap(); + let status = state.status.as_ref().unwrap(); + assert_eq!(status.status, RunStatus::Paused); + assert_eq!(status.blocked_reason, None); + } + + #[test] + fn missing_lifecycle_status_synthesizes_submitted() { + let state = RunProjection::default(); + let summary = state.build_summary(&fixtures::RUN_1); + assert_eq!(summary.status, Some(fabro_types::RunStatus::Submitted)); + } } diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index fe3fae6a8..704cff4dd 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use chrono::{DateTime, Utc}; -use fabro_types::{RunControlAction, RunEvent, RunId, RunStatus, StatusReason}; +use fabro_types::{BlockedReason, RunControlAction, RunEvent, RunId, RunStatus, StatusReason}; use serde::{Deserialize, Serialize}; use crate::{Error, Result}; @@ -17,6 +17,7 @@ pub struct RunSummary { pub start_time: Option>, pub status: Option, pub status_reason: Option, + pub blocked_reason: Option, pub pending_control: Option, pub duration_ms: Option, pub total_usd_micros: Option, diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 1b994278c..599df8d45 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -53,6 +53,6 @@ pub use sandbox_record::SandboxRecord; pub use stage_id::{ParallelBranchId, StageId}; pub use start::StartRecord; pub use status::{ - InvalidTransition, ParseRunStatusError, RunControlAction, RunStatus, RunStatusRecord, - StatusReason, + BlockedReason, InvalidTransition, ParseRunStatusError, RunControlAction, RunStatus, + RunStatusRecord, StatusReason, }; diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index 1496563e4..90d498dfe 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -106,6 +106,12 @@ pub enum EventBody { RunPaused(RunControlEffectProps), #[serde(rename = "run.unpaused")] RunUnpaused(RunControlEffectProps), + #[serde(rename = "run.queued")] + RunQueued(RunStatusTransitionProps), + #[serde(rename = "run.blocked")] + RunBlocked(RunBlockedProps), + #[serde(rename = "run.unblocked")] + RunUnblocked(RunUnblockedProps), #[serde(rename = "run.rewound")] RunRewound(RunRewoundProps), #[serde(rename = "run.completed")] @@ -365,6 +371,9 @@ impl EventBody { Self::RunUnpauseRequested(_) => "run.unpause.requested", Self::RunPaused(_) => "run.paused", Self::RunUnpaused(_) => "run.unpaused", + Self::RunQueued(_) => "run.queued", + Self::RunBlocked(_) => "run.blocked", + Self::RunUnblocked(_) => "run.unblocked", Self::RunRewound(_) => "run.rewound", Self::RunCompleted(_) => "run.completed", Self::RunFailed(_) => "run.failed", diff --git a/lib/crates/fabro-types/src/run_event/run.rs b/lib/crates/fabro-types/src/run_event/run.rs index 90fdb2fda..8dd8fdea0 100644 --- a/lib/crates/fabro-types/src/run_event/run.rs +++ b/lib/crates/fabro-types/src/run_event/run.rs @@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize}; use super::{BilledTokenCounts, RunNoticeLevel}; use crate::settings::SettingsLayer; -use crate::{Graph, RunBlobId, RunControlAction, RunProvenance, StatusReason}; +use crate::{BlockedReason, Graph, RunBlobId, RunControlAction, RunProvenance, StatusReason}; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct RunCreatedProps { @@ -116,3 +116,12 @@ pub struct RunNoticeProps { pub code: String, pub message: String, } + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct RunBlockedProps { + pub blocked_reason: BlockedReason, +} + +#[allow(clippy::empty_structs_with_brackets)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +pub struct RunUnblockedProps {} diff --git a/lib/crates/fabro-types/src/status.rs b/lib/crates/fabro-types/src/status.rs index 7b6f9189e..9b18d11b0 100644 --- a/lib/crates/fabro-types/src/status.rs +++ b/lib/crates/fabro-types/src/status.rs @@ -8,8 +8,10 @@ use serde::{Deserialize, Serialize}; #[serde(rename_all = "snake_case")] pub enum RunStatus { Submitted, + Queued, Starting, Running, + Blocked, Paused, Removing, Succeeded, @@ -25,7 +27,13 @@ impl RunStatus { pub fn is_active(self) -> bool { matches!( self, - Self::Submitted | Self::Starting | Self::Running | Self::Paused | Self::Removing + Self::Submitted + | Self::Queued + | Self::Starting + | Self::Running + | Self::Blocked + | Self::Paused + | Self::Removing ) } @@ -38,16 +46,18 @@ impl RunStatus { } matches!( (self, to), - (Self::Submitted, Self::Starting) + (Self::Submitted, Self::Queued | Self::Starting) + | (Self::Queued, Self::Starting) | (Self::Starting | Self::Paused, Self::Running) + | ( + Self::Running, + Self::Blocked | Self::Succeeded | Self::Paused | Self::Removing + ) + | (Self::Blocked, Self::Running | Self::Paused | Self::Failed) | ( Self::Starting | Self::Running | Self::Paused | Self::Removing, Self::Failed ) - | ( - Self::Running, - Self::Succeeded | Self::Paused | Self::Removing - ) | (Self::Paused, Self::Removing) ) } @@ -65,8 +75,10 @@ impl fmt::Display for RunStatus { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let s = match self { Self::Submitted => "submitted", + Self::Queued => "queued", Self::Starting => "starting", Self::Running => "running", + Self::Blocked => "blocked", Self::Paused => "paused", Self::Removing => "removing", Self::Succeeded => "succeeded", @@ -83,8 +95,10 @@ impl FromStr for RunStatus { fn from_str(s: &str) -> Result { match s { "submitted" => Ok(Self::Submitted), + "queued" => Ok(Self::Queued), "starting" => Ok(Self::Starting), "running" => Ok(Self::Running), + "blocked" => Ok(Self::Blocked), "paused" => Ok(Self::Paused), "removing" => Ok(Self::Removing), "succeeded" => Ok(Self::Succeeded), @@ -136,6 +150,12 @@ pub enum StatusReason { SandboxInitializing, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum BlockedReason { + HumanInputRequired, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum RunControlAction { @@ -148,16 +168,90 @@ pub enum RunControlAction { pub struct RunStatusRecord { pub status: RunStatus, #[serde(default, skip_serializing_if = "Option::is_none")] - pub reason: Option, + pub status_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub blocked_reason: Option, pub updated_at: DateTime, } impl RunStatusRecord { - pub fn new(status: RunStatus, reason: Option) -> Self { + pub fn new(status: RunStatus, status_reason: Option) -> Self { Self { status, - reason, + status_reason, + blocked_reason: None, updated_at: Utc::now(), } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transition_submitted_to_queued() { + assert!(RunStatus::Submitted.can_transition_to(RunStatus::Queued)); + } + + #[test] + fn transition_queued_to_starting() { + assert!(RunStatus::Queued.can_transition_to(RunStatus::Starting)); + } + + #[test] + fn transition_running_to_blocked() { + assert!(RunStatus::Running.can_transition_to(RunStatus::Blocked)); + } + + #[test] + fn transition_blocked_to_running() { + assert!(RunStatus::Blocked.can_transition_to(RunStatus::Running)); + } + + #[test] + fn transition_blocked_to_paused() { + assert!(RunStatus::Blocked.can_transition_to(RunStatus::Paused)); + } + + #[test] + fn transition_blocked_to_failed() { + assert!(RunStatus::Blocked.can_transition_to(RunStatus::Failed)); + } + + #[test] + fn no_direct_paused_to_blocked() { + assert!(!RunStatus::Paused.can_transition_to(RunStatus::Blocked)); + } + + #[test] + fn display_and_from_str_queued() { + let s = RunStatus::Queued.to_string(); + assert_eq!(s, "queued"); + assert_eq!(RunStatus::from_str(&s).unwrap(), RunStatus::Queued); + } + + #[test] + fn display_and_from_str_blocked() { + let s = RunStatus::Blocked.to_string(); + assert_eq!(s, "blocked"); + assert_eq!(RunStatus::from_str(&s).unwrap(), RunStatus::Blocked); + } + + #[test] + fn blocked_reason_serde_round_trip() { + let reason = BlockedReason::HumanInputRequired; + let json = serde_json::to_string(&reason).unwrap(); + assert_eq!(json, r#""human_input_required""#); + let parsed: BlockedReason = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed, reason); + } + + #[test] + fn queued_and_blocked_are_active() { + assert!(RunStatus::Queued.is_active()); + assert!(RunStatus::Blocked.is_active()); + assert!(!RunStatus::Queued.is_terminal()); + assert!(!RunStatus::Blocked.is_terminal()); + } +} diff --git a/lib/crates/fabro-workflow/src/blocked_state.rs b/lib/crates/fabro-workflow/src/blocked_state.rs new file mode 100644 index 000000000..3ccbcfe87 --- /dev/null +++ b/lib/crates/fabro-workflow/src/blocked_state.rs @@ -0,0 +1,156 @@ +use std::sync::Mutex; + +use fabro_types::BlockedReason; + +use crate::event::{Emitter, Event}; + +/// Tracks the number of unresolved human interview questions for a run. +/// +/// Emits `run.blocked` on the `0 -> 1` transition and `run.unblocked` on +/// the `1 -> 0` transition. Thread-safe for parallel human stages. +pub struct BlockedStateTracker { + state: Mutex, + emitter: std::sync::Arc, +} + +struct BlockedState { + unresolved_count: usize, +} + +impl BlockedStateTracker { + pub fn new(emitter: std::sync::Arc) -> Self { + Self { + state: Mutex::new(BlockedState { + unresolved_count: 0, + }), + emitter, + } + } + + /// Called when a new interview question is started. If this is the first + /// unresolved question (0 -> 1), emits `run.blocked`. + pub fn on_interview_started(&self) { + let mut state = self.state.lock().expect("blocked state lock poisoned"); + let was_zero = state.unresolved_count == 0; + state.unresolved_count += 1; + if was_zero { + drop(state); + self.emitter.emit(&Event::RunBlocked { + blocked_reason: BlockedReason::HumanInputRequired, + }); + } + } + + /// Called when an interview question is resolved (completed, timed out, or + /// interrupted). If this was the last unresolved question (1 -> 0), emits + /// `run.unblocked`. + pub fn on_interview_resolved(&self) { + let mut state = self.state.lock().expect("blocked state lock poisoned"); + state.unresolved_count = state.unresolved_count.saturating_sub(1); + let now_zero = state.unresolved_count == 0; + drop(state); + if now_zero { + self.emitter.emit(&Event::RunUnblocked); + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use fabro_types::RunId; + + use super::*; + + fn test_emitter() -> Arc { + let run_id = RunId::new(); + Arc::new(Emitter::new(run_id)) + } + + #[test] + fn first_interview_emits_blocked() { + let emitter = test_emitter(); + let events = Arc::new(Mutex::new(Vec::new())); + let events_clone = Arc::clone(&events); + emitter.on_event(move |event| { + events_clone + .lock() + .unwrap() + .push(event.body.event_name().to_string()); + }); + let tracker = BlockedStateTracker::new(emitter); + tracker.on_interview_started(); + let names: Vec = events.lock().unwrap().clone(); + assert!(names.contains(&"run.blocked".to_string())); + } + + #[test] + fn second_interview_does_not_emit_blocked() { + let emitter = test_emitter(); + let events = Arc::new(Mutex::new(Vec::new())); + let events_clone = Arc::clone(&events); + emitter.on_event(move |event| { + events_clone + .lock() + .unwrap() + .push(event.body.event_name().to_string()); + }); + let tracker = BlockedStateTracker::new(emitter); + tracker.on_interview_started(); + tracker.on_interview_started(); + let names: Vec = events.lock().unwrap().clone(); + assert_eq!( + names.iter().filter(|n| *n == "run.blocked").count(), + 1, + "should emit run.blocked exactly once" + ); + } + + #[test] + fn last_resolution_emits_unblocked() { + let emitter = test_emitter(); + let events = Arc::new(Mutex::new(Vec::new())); + let events_clone = Arc::clone(&events); + emitter.on_event(move |event| { + events_clone + .lock() + .unwrap() + .push(event.body.event_name().to_string()); + }); + let tracker = BlockedStateTracker::new(emitter); + tracker.on_interview_started(); + tracker.on_interview_started(); + tracker.on_interview_resolved(); // 2 -> 1, no unblocked + let names: Vec = events.lock().unwrap().clone(); + assert!( + !names.contains(&"run.unblocked".to_string()), + "should not emit run.unblocked with 1 remaining" + ); + tracker.on_interview_resolved(); // 1 -> 0, unblocked + let names: Vec = events.lock().unwrap().clone(); + assert!( + names.contains(&"run.unblocked".to_string()), + "should emit run.unblocked on last resolution" + ); + } + + #[test] + fn exactly_one_blocked_and_one_unblocked_for_single_interview() { + let emitter = test_emitter(); + let events = Arc::new(Mutex::new(Vec::new())); + let events_clone = Arc::clone(&events); + emitter.on_event(move |event| { + events_clone + .lock() + .unwrap() + .push(event.body.event_name().to_string()); + }); + let tracker = BlockedStateTracker::new(emitter); + tracker.on_interview_started(); + tracker.on_interview_resolved(); + let names: Vec = events.lock().unwrap().clone(); + assert_eq!(names.iter().filter(|n| *n == "run.blocked").count(), 1); + assert_eq!(names.iter().filter(|n| *n == "run.unblocked").count(), 1); + } +} diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f339098b9..e5861ca19 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -5,8 +5,8 @@ use std::sync::Arc; use std::sync::atomic::{AtomicI64, Ordering}; use ::fabro_types::{ - ActorRef, BilledTokenCounts, ParallelBranchId, RunBlobId, RunControlAction, RunEvent, RunId, - RunProvenance, StageId, StageStatus, StatusReason, run_event as fabro_types, + ActorRef, BilledTokenCounts, BlockedReason, ParallelBranchId, RunBlobId, RunControlAction, + RunEvent, RunId, RunProvenance, StageId, StageStatus, StatusReason, run_event as fabro_types, }; use anyhow::{Context, Result}; use chrono::Utc; @@ -104,6 +104,14 @@ pub enum Event { }, RunPaused, RunUnpaused, + RunQueued { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + RunBlocked { + blocked_reason: BlockedReason, + }, + RunUnblocked, RunRewound { target_checkpoint_ordinal: usize, target_node_id: String, @@ -598,6 +606,15 @@ impl Event { Self::RunUnpaused => { info!("Run unpaused"); } + Self::RunQueued { .. } => { + info!("Run queued"); + } + Self::RunBlocked { blocked_reason } => { + info!(?blocked_reason, "Run blocked"); + } + Self::RunUnblocked => { + info!("Run unblocked"); + } Self::RunRewound { target_checkpoint_ordinal, target_node_id, @@ -1154,6 +1171,9 @@ pub fn event_name(event: &Event) -> &'static str { Event::RunUnpauseRequested { .. } => "run.unpause.requested", Event::RunPaused => "run.paused", Event::RunUnpaused => "run.unpaused", + Event::RunQueued { .. } => "run.queued", + Event::RunBlocked { .. } => "run.blocked", + Event::RunUnblocked => "run.unblocked", Event::RunRewound { .. } => "run.rewound", Event::WorkflowRunCompleted { .. } => "run.completed", Event::WorkflowRunFailed { .. } => "run.failed", @@ -1544,6 +1564,15 @@ fn event_body_from_event(event: &Event) -> EventBody { } Event::RunPaused => EventBody::RunPaused(fabro_types::RunControlEffectProps::default()), Event::RunUnpaused => EventBody::RunUnpaused(fabro_types::RunControlEffectProps::default()), + Event::RunQueued { reason } => { + EventBody::RunQueued(fabro_types::RunStatusTransitionProps { reason: *reason }) + } + Event::RunBlocked { blocked_reason } => { + EventBody::RunBlocked(fabro_types::RunBlockedProps { + blocked_reason: *blocked_reason, + }) + } + Event::RunUnblocked => EventBody::RunUnblocked(fabro_types::RunUnblockedProps::default()), Event::RunRewound { target_checkpoint_ordinal, target_node_id, diff --git a/lib/crates/fabro-workflow/src/handler/human.rs b/lib/crates/fabro-workflow/src/handler/human.rs index d0491a252..54c4a7759 100644 --- a/lib/crates/fabro-workflow/src/handler/human.rs +++ b/lib/crates/fabro-workflow/src/handler/human.rs @@ -229,6 +229,9 @@ impl Handler for HumanHandler { }, &stage_scope, ); + if let Some(tracker) = &services.blocked_state_tracker { + tracker.on_interview_started(); + } let interview_start = Instant::now(); let answer = self.interviewer.ask(question).await; @@ -244,6 +247,9 @@ impl Handler for HumanHandler { }, &stage_scope, ); + if let Some(tracker) = &services.blocked_state_tracker { + tracker.on_interview_resolved(); + } let default_choice = node .attrs .get("human.default_choice") @@ -259,6 +265,8 @@ impl Handler for HumanHandler { } if answer.value == AnswerValue::Cancelled { + // Don't emit run.unblocked on cancellation; terminal events + // end the blocked condition implicitly. return Err(Error::Cancelled); } @@ -282,6 +290,9 @@ impl Handler for HumanHandler { }, &stage_scope, ); + if let Some(tracker) = &services.blocked_state_tracker { + tracker.on_interview_resolved(); + } return Ok(unanswered_human_gate( "human interaction interrupted before an answer was provided", )); @@ -297,6 +308,9 @@ impl Handler for HumanHandler { }, &stage_scope, ); + if let Some(tracker) = &services.blocked_state_tracker { + tracker.on_interview_resolved(); + } return Ok(unanswered_human_gate("human skipped interaction")); } @@ -311,6 +325,9 @@ impl Handler for HumanHandler { }, &stage_scope, ); + if let Some(tracker) = &services.blocked_state_tracker { + tracker.on_interview_resolved(); + } // 6. Try fixed-choice match if let Some(selected) = find_choice_match(&answer, &choices) { diff --git a/lib/crates/fabro-workflow/src/handler/mod.rs b/lib/crates/fabro-workflow/src/handler/mod.rs index d9e59c2a8..ea2c6a302 100644 --- a/lib/crates/fabro-workflow/src/handler/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/mod.rs @@ -31,6 +31,7 @@ use object_store::memory::InMemory; use tokio::time; use tokio_util::sync::CancellationToken; +use crate::blocked_state::BlockedStateTracker; use crate::context::Context; use crate::error::Error; use crate::event::Emitter; @@ -66,6 +67,8 @@ pub struct EngineServices { pub workflow_path: Option, /// Bundled workflows available for child-workflow resolution. pub workflow_bundle: Option>, + /// Run-scoped blocked-state tracker for human interview coordination. + pub blocked_state_tracker: Option>, } impl EngineServices { @@ -139,6 +142,7 @@ impl EngineServices { provider: Provider::Anthropic, workflow_path: None, workflow_bundle: None, + blocked_state_tracker: None, } } } diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 87afb627c..b7351f70c 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -294,6 +294,7 @@ impl Handler for ParallelHandler { let provider = services.provider; let workflow_path = services.workflow_path.clone(); let workflow_bundle = services.workflow_bundle.clone(); + let blocked_state_tracker = services.blocked_state_tracker.clone(); let graph = graph.clone(); let run_dir = run_dir.to_path_buf(); let sem = Arc::clone(&semaphore); @@ -367,6 +368,7 @@ impl Handler for ParallelHandler { provider, workflow_path, workflow_bundle, + blocked_state_tracker, }; let handler = registry.resolve(target_node); let outcome = super::dispatch_handler( diff --git a/lib/crates/fabro-workflow/src/lib.rs b/lib/crates/fabro-workflow/src/lib.rs index 9eda2b32c..2437d9958 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/lib/crates/fabro-workflow/src/lib.rs @@ -116,6 +116,7 @@ pub fn extract_stage_durations_from_events(events: &[EventEnvelope]) -> HashMap< pub mod artifact; pub mod artifact_snapshot; pub mod artifact_upload; +pub mod blocked_state; pub(crate) mod condition; pub mod context; pub mod devcontainer_bridge; diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index 1d305030d..2277c407c 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -9,6 +9,7 @@ use tokio_util::sync::CancellationToken; use super::types::{Executed, Initialized}; use crate::artifact; +use crate::blocked_state; use crate::context::{self, Context}; use crate::error::Error; use crate::event::Event; @@ -81,6 +82,9 @@ pub async fn execute(init: Initialized) -> Executed { })) }); + let blocked_state_tracker = Some(Arc::new(blocked_state::BlockedStateTracker::new( + Arc::clone(&emitter), + ))); let shared_services = Arc::new(EngineServices { registry, emitter: Arc::clone(&emitter), @@ -95,6 +99,7 @@ pub async fn execute(init: Initialized) -> Executed { provider, workflow_path, workflow_bundle, + blocked_state_tracker, }); let handler = Arc::new(WorkflowNodeHandler { diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index c63b06664..db3603ef3 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -788,7 +788,7 @@ async fn execute_cancelled_mid_run_persists_cancelled_status() { assert!(matches!(executed.outcome, Err(Error::Cancelled))); let status = executed.run_store.state().await.unwrap().status.unwrap(); assert_eq!(status.status, RunStatus::Failed); - assert_eq!(status.reason, Some(StatusReason::Cancelled)); + assert_eq!(status.status_reason, Some(StatusReason::Cancelled)); } #[tokio::test] diff --git a/lib/packages/fabro-api-client/src/models/blocked-reason.ts b/lib/packages/fabro-api-client/src/models/blocked-reason.ts new file mode 100644 index 000000000..784fdcdc1 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/blocked-reason.ts @@ -0,0 +1,27 @@ +/* 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. + */ + + + +/** + * Reason a run is blocked. + */ + +export const BlockedReason = { + HUMAN_INPUT_REQUIRED: 'human_input_required' +} as const; + +export type BlockedReason = typeof BlockedReason[keyof typeof BlockedReason]; + + diff --git a/lib/packages/fabro-api-client/src/models/board-column.ts b/lib/packages/fabro-api-client/src/models/board-column.ts index 24385ce4b..6462d1aef 100644 --- a/lib/packages/fabro-api-client/src/models/board-column.ts +++ b/lib/packages/fabro-api-client/src/models/board-column.ts @@ -19,13 +19,13 @@ */ export const BoardColumn = { - WORKING: 'working', INITIALIZING: 'initializing', - REVIEW: 'review', - MERGE: 'merge' + RUNNING: 'running', + BLOCKED: 'blocked', + SUCCEEDED: 'succeeded', + FAILED: 'failed' } as const; export type BoardColumn = typeof BoardColumn[keyof typeof BoardColumn]; - diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index ef42d8cb7..c79550ea3 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -13,6 +13,7 @@ export * from './assistant-stage-turn'; export * from './billed-token-counts'; export * from './billing-by-model'; export * from './billing-stage-ref'; +export * from './blocked-reason'; export * from './board-column'; export * from './check-run'; export * from './check-run-status'; @@ -143,4 +144,4 @@ export * from './tool-use'; export * from './user-response'; export * from './workflow-diagnostic'; export * from './workflow-reference'; -export * from './write-blob-response'; +export * from './write-blob-response'; \ No newline at end of file diff --git a/lib/packages/fabro-api-client/src/models/run-status-record.ts b/lib/packages/fabro-api-client/src/models/run-status-record.ts index 8a74e7576..c517f2afd 100644 --- a/lib/packages/fabro-api-client/src/models/run-status-record.ts +++ b/lib/packages/fabro-api-client/src/models/run-status-record.ts @@ -15,7 +15,10 @@ // May contain unused imports in some cases // @ts-ignore -import type { InternalRunStatus } from './internal-run-status'; +import type { BlockedReason } from './blocked-reason'; +// 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'; @@ -24,10 +27,10 @@ import type { StatusReason } from './status-reason'; * Internal run status record from the event projection. */ export interface RunStatusRecord { - 'status': InternalRunStatus; - 'reason'?: StatusReason | null; + 'status': RunStatus; + 'status_reason'?: StatusReason | null; + 'blocked_reason'?: BlockedReason | null; 'updated_at': string; } - diff --git a/lib/packages/fabro-api-client/src/models/run-status-response.ts b/lib/packages/fabro-api-client/src/models/run-status-response.ts index 2c7c0a17c..e5f6ee53b 100644 --- a/lib/packages/fabro-api-client/src/models/run-status-response.ts +++ b/lib/packages/fabro-api-client/src/models/run-status-response.ts @@ -13,6 +13,9 @@ */ +// May contain unused imports in some cases +// @ts-ignore +import type { BlockedReason } from './blocked-reason'; // May contain unused imports in some cases // @ts-ignore import type { RunControlAction } from './run-control-action'; @@ -41,6 +44,7 @@ export interface RunStatusResponse { */ 'queue_position'?: number; 'status_reason'?: StatusReason; + 'blocked_reason'?: BlockedReason; 'pending_control'?: RunControlAction; /** * Timestamp when the run was created. @@ -49,4 +53,3 @@ export interface RunStatusResponse { } - diff --git a/lib/packages/fabro-api-client/src/models/run-status.ts b/lib/packages/fabro-api-client/src/models/run-status.ts index 61b235092..a2484d440 100644 --- a/lib/packages/fabro-api-client/src/models/run-status.ts +++ b/lib/packages/fabro-api-client/src/models/run-status.ts @@ -23,13 +23,14 @@ export const RunStatus = { QUEUED: 'queued', STARTING: 'starting', RUNNING: 'running', - COMPLETED: 'completed', + BLOCKED: 'blocked', + PAUSED: 'paused', + REMOVING: 'removing', + SUCCEEDED: 'succeeded', FAILED: 'failed', - CANCELLED: 'cancelled', - PAUSED: 'paused' + DEAD: 'dead' } as const; export type RunStatus = typeof RunStatus[keyof typeof RunStatus]; - diff --git a/lib/packages/fabro-api-client/src/models/store-run-summary.ts b/lib/packages/fabro-api-client/src/models/store-run-summary.ts index e5d68c84a..40b9f80de 100644 --- a/lib/packages/fabro-api-client/src/models/store-run-summary.ts +++ b/lib/packages/fabro-api-client/src/models/store-run-summary.ts @@ -13,9 +13,18 @@ */ +// May contain unused imports in some cases +// @ts-ignore +import type { BlockedReason } from './blocked-reason'; // 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 { RunStatus } from './run-status'; +// May contain unused imports in some cases +// @ts-ignore +import type { StatusReason } from './status-reason'; /** * Durable run summary derived from the backing store. @@ -28,12 +37,12 @@ export interface StoreRunSummary { 'labels': { [key: string]: string; }; 'host_repo_path'?: string; 'start_time'?: string; - 'status'?: string; - 'status_reason'?: string; + 'status': RunStatus; + 'status_reason'?: StatusReason; + 'blocked_reason'?: BlockedReason; 'pending_control'?: RunControlAction; 'duration_ms'?: number; 'total_usd_micros'?: number; } -