bamboo_server/handlers/
workflow_runs.rs

1//! Safe HTTP facade for durable catalog-pinned workflow runs.
2
3use actix_web::{web, HttpResponse};
4use bamboo_engine::WorkflowRunError;
5use serde::Deserialize;
6use serde_json::Value;
7
8use crate::app_state::AppState;
9use crate::workflow::public_workflow_snapshot;
10
11#[derive(Debug, Deserialize)]
12#[serde(deny_unknown_fields)]
13pub struct StartWorkflowRunRequest {
14    pub workflow_id: String,
15    pub revision: u64,
16    #[serde(default = "empty_object")]
17    pub args: Value,
18    #[serde(default)]
19    pub budget: Option<bamboo_domain::WorkflowBudgets>,
20}
21
22fn empty_object() -> Value {
23    serde_json::json!({})
24}
25
26#[derive(Debug, Default, Deserialize)]
27#[serde(deny_unknown_fields)]
28pub struct WorkflowEventsQuery {
29    #[serde(default)]
30    pub since: u64,
31}
32
33pub async fn start(
34    state: web::Data<AppState>,
35    session_id: web::Path<String>,
36    body: web::Json<StartWorkflowRunRequest>,
37) -> HttpResponse {
38    match state
39        .workflow_runs
40        .start(
41            &session_id,
42            &body.workflow_id,
43            body.revision,
44            body.args.clone(),
45            body.budget.clone(),
46        )
47        .await
48    {
49        Ok(snapshot) => HttpResponse::Accepted().json(public_workflow_snapshot(snapshot)),
50        Err(error) => workflow_error(error),
51    }
52}
53
54pub async fn list(state: web::Data<AppState>, session_id: web::Path<String>) -> HttpResponse {
55    match state.workflow_runs.list_for_session(&session_id).await {
56        Ok(snapshots) => HttpResponse::Ok().json(
57            snapshots
58                .into_iter()
59                .map(public_workflow_snapshot)
60                .collect::<Vec<_>>(),
61        ),
62        Err(error) => workflow_error(error),
63    }
64}
65
66pub async fn get(state: web::Data<AppState>, path: web::Path<(String, String)>) -> HttpResponse {
67    let (session_id, run_id) = path.into_inner();
68    match state
69        .workflow_runs
70        .progress_for_session(&session_id, &run_id, u64::MAX)
71        .await
72    {
73        Ok(progress) => HttpResponse::Ok().json(public_workflow_snapshot(progress.snapshot)),
74        Err(error) => workflow_error(error),
75    }
76}
77
78pub async fn events(
79    state: web::Data<AppState>,
80    path: web::Path<(String, String)>,
81    query: web::Query<WorkflowEventsQuery>,
82) -> HttpResponse {
83    let (session_id, run_id) = path.into_inner();
84    match state
85        .workflow_runs
86        .progress_for_session(&session_id, &run_id, query.since)
87        .await
88    {
89        Ok(progress) => HttpResponse::Ok().json(progress.events),
90        Err(error) => workflow_error(error),
91    }
92}
93
94pub async fn cancel(state: web::Data<AppState>, path: web::Path<(String, String)>) -> HttpResponse {
95    let (session_id, run_id) = path.into_inner();
96    match state
97        .workflow_runs
98        .cancel_for_session(&session_id, &run_id)
99        .await
100    {
101        Ok(snapshot) => HttpResponse::Ok().json(public_workflow_snapshot(snapshot)),
102        Err(error) => workflow_error(error),
103    }
104}
105
106pub async fn restart(
107    state: web::Data<AppState>,
108    path: web::Path<(String, String)>,
109) -> HttpResponse {
110    let (session_id, run_id) = path.into_inner();
111    match state
112        .workflow_runs
113        .restart_for_session(&session_id, &run_id)
114        .await
115    {
116        Ok(snapshot) => HttpResponse::Accepted().json(public_workflow_snapshot(snapshot)),
117        Err(error) => workflow_error(error),
118    }
119}
120
121fn workflow_error(error: WorkflowRunError) -> HttpResponse {
122    let message = error.to_string();
123    match error {
124        WorkflowRunError::NotFound => HttpResponse::NotFound().json(serde_json::json!({
125            "error": crate::error::error_value(message)
126        })),
127        WorkflowRunError::Terminal => HttpResponse::Conflict().json(serde_json::json!({
128            "error": crate::error::error_value(message)
129        })),
130        WorkflowRunError::Storage(details) => {
131            let recovery_run_id = recovery_run_id_from_storage_details(&details);
132            HttpResponse::InternalServerError().json(match recovery_run_id {
133                Some(run_id) => serde_json::json!({
134                    "error": crate::error::error_value(
135                        "workflow storage unavailable; run recovery is required"
136                    ),
137                    "recovery_run_id": run_id,
138                }),
139                None => serde_json::json!({
140                    "error": crate::error::error_value("workflow storage unavailable")
141                }),
142            })
143        }
144        WorkflowRunError::Compile(_)
145        | WorkflowRunError::InvalidInput(_)
146        | WorkflowRunError::Preflight(_) => HttpResponse::BadRequest()
147            .json(serde_json::json!({ "error": crate::error::error_value(message) })),
148    }
149}
150
151fn recovery_run_id_from_storage_details(details: &str) -> Option<&str> {
152    details
153        .split("orphan run ")
154        .nth(1)
155        .and_then(|tail| tail.split_whitespace().next())
156        .filter(|value| {
157            !value.is_empty()
158                && value.len() <= 64
159                && value
160                    .chars()
161                    .all(|character| character.is_ascii_alphanumeric() || character == '-')
162        })
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168
169    #[test]
170    fn start_request_rejects_spoofed_trust_and_capabilities() {
171        for field in ["workspace_trusted", "allowed_capabilities"] {
172            let mut value = serde_json::json!({
173                "workflow_id": "safe",
174                "revision": 1
175            });
176            value[field] = serde_json::json!(true);
177            assert!(serde_json::from_value::<StartWorkflowRunRequest>(value).is_err());
178        }
179        assert!(
180            serde_json::from_value::<StartWorkflowRunRequest>(serde_json::json!({
181                "workflow_id": "safe",
182                "revision": 1,
183                "session_id": "caller-controlled"
184            }))
185            .is_err()
186        );
187    }
188
189    #[actix_web::test]
190    async fn orphan_storage_error_returns_safe_recovery_handle() {
191        let response = workflow_error(WorkflowRunError::Storage(
192            "run index persistence failed; orphan run 123e4567-e89b-12d3-a456-426614174000 could not be cancelled (failure)"
193                .to_string(),
194        ));
195        assert_eq!(
196            response.status(),
197            actix_web::http::StatusCode::INTERNAL_SERVER_ERROR
198        );
199        let body = actix_web::body::to_bytes(response.into_body())
200            .await
201            .expect("body");
202        let value: serde_json::Value = serde_json::from_slice(&body).expect("json");
203        assert_eq!(
204            value["recovery_run_id"],
205            "123e4567-e89b-12d3-a456-426614174000"
206        );
207        assert_eq!(value["error"]["type"], "api_error");
208        assert_eq!(
209            value["error"]["message"],
210            "workflow storage unavailable; run recovery is required"
211        );
212        assert!(!value["error"]["message"]
213            .as_str()
214            .unwrap_or_default()
215            .contains("failure"));
216        assert_eq!(recovery_run_id_from_storage_details("disk /secret"), None);
217    }
218
219    #[actix_web::test]
220    async fn workflow_errors_preserve_status_and_use_canonical_envelope() {
221        let cases = [
222            (
223                WorkflowRunError::NotFound,
224                actix_web::http::StatusCode::NOT_FOUND,
225            ),
226            (
227                WorkflowRunError::Terminal,
228                actix_web::http::StatusCode::CONFLICT,
229            ),
230            (
231                WorkflowRunError::InvalidInput("bad args".to_string()),
232                actix_web::http::StatusCode::BAD_REQUEST,
233            ),
234        ];
235
236        for (error, expected_status) in cases {
237            let expected_message = error.to_string();
238            let response = workflow_error(error);
239            assert_eq!(response.status(), expected_status);
240            let body = actix_web::body::to_bytes(response.into_body())
241                .await
242                .expect("body");
243            let value: serde_json::Value = serde_json::from_slice(&body).expect("json");
244            assert_eq!(value["error"]["message"], expected_message);
245            assert_eq!(value["error"]["type"], "api_error");
246        }
247    }
248}
bamboo_server/handlers/workflow_runs.rs

bamboo_server/handlers/
workflow_runs.rs
