flow execution is rejected when a flow is diabled

src/flow/mod.rs

@@ -19,6 +19,12 @@ pub fn flow_belongs_to_action(flow: &ValidationFlow, action_identifier: &str) ->
     flow.definition_source.as_deref() == Some(source.as_str())
 }
 
+/// Whether `flow` has been disabled, and if so, why. Callers must reject
+/// execution requests against a disabled flow rather than dispatching them.
+pub fn flow_disable_reason(flow: &ValidationFlow) -> Option<&str> {
+    flow.disable_reason.as_deref()
+}
+
 /// Projects a `ValidationFlow` down to the fields an action needs to
 /// execute/validate against it - an "action flow".
 pub fn to_action_flow(flow: &ValidationFlow) -> ActionFlow {
@@ -135,6 +141,21 @@ mod tests {
         assert!(!flow_belongs_to_action(&flow(None), "send-email"));
     }
 
+    #[test]
+    fn flow_disable_reason_returns_reason_when_disabled() {
+        let disabled = ValidationFlow {
+            disable_reason: Some("maintenance".to_string()),
+            ..Default::default()
+        };
+        let enabled = ValidationFlow {
+            disable_reason: None,
+            ..Default::default()
+        };
+
+        assert_eq!(flow_disable_reason(&disabled), Some("maintenance"));
+        assert_eq!(flow_disable_reason(&enabled), None);
+    }
+
     #[test]
     fn to_action_flow_projects_matching_fields() {
         let flow = ValidationFlow {

src/sagittarius/test_execution_client_impl/mod.rs

@@ -124,6 +124,39 @@ impl SagittariusTestExecutionServiceClient {
                                 }
                             };
 
+                        if let Some(reason) = flow::flow_disable_reason(&validation_flow) {
+                            log::warn!(
+                                "Rejecting Sagittarius execution request for a disabled flow requested_execution_id={} flow_id={} reason={}",
+                                request.execution_identifier,
+                                request.flow_id,
+                                reason
+                            );
+
+                            let execution_id = if request.execution_identifier.is_empty() {
+                                uuid::Uuid::new_v4().to_string()
+                            } else {
+                                request.execution_identifier.clone()
+                            };
+
+                            let rejection = validation::disabled_flow_rejection_result(
+                                execution_id,
+                                request.flow_id,
+                                reason,
+                            );
+
+                            if let Err(status) =
+                                self.response_sender.send_execution_result(rejection).await
+                            {
+                                log::error!(
+                                    "Failed to send disabled flow rejection result flow_id={} error={:?}",
+                                    request.flow_id,
+                                    status
+                                );
+                            }
+
+                            continue;
+                        }
+
                         if validation::is_rest_flow(&validation_flow) {
                             let input_schema = validation::extract_input_schema(&validation_flow);
                             if let Err(err) = validation::validate_body_against_schema(

src/server/action_transfer/nats_bridge.rs

@@ -113,6 +113,25 @@ pub(super) async fn handle_flow_execution(
         return;
     }
 
+    if let Some(reason) = flow::flow_disable_reason(&validation_flow) {
+        log::warn!(
+            "Rejected action flow execution request for a disabled flow action={} flow_id={} reason={}",
+            action_identifier,
+            flow_id,
+            reason
+        );
+        send_flow_execution_failure(
+            &tx,
+            execution_id,
+            format!(
+                "flow {} has been disabled for the reason: {}",
+                flow_id, reason
+            ),
+        )
+        .await;
+        return;
+    }
+
     if validation::is_rest_flow(&validation_flow) {
         let input_schema = validation::extract_input_schema(&validation_flow);
         if let Err(err) =

src/validation.rs

@@ -187,6 +187,31 @@ pub fn rejection_result(
     }
 }
 
+/// Synthesizes the [`ExecutionResult`] sent back in place of dispatching an
+/// execution request against a disabled flow.
+pub fn disabled_flow_rejection_result(
+    execution_identifier: String,
+    flow_id: i64,
+    reason: &str,
+) -> ExecutionResult {
+    let now = epoch_millis_now();
+    ExecutionResult {
+        execution_identifier,
+        flow_id,
+        started_at: now,
+        finished_at: now,
+        result: Some(execution_result::Result::Error(Error {
+            code: "A-VALIDATION-000002".to_string(),
+            category: "InvalidArgument".to_string(),
+            message: format!("flow {} has been disabled for the reason: {}", flow_id, reason),
+            timestamp: now,
+            version: crate::version::runtime_version().to_string(),
+            ..Default::default()
+        })),
+        ..Default::default()
+    }
+}
+
 pub(crate) fn epoch_millis_now() -> i64 {
     match SystemTime::now().duration_since(UNIX_EPOCH) {
         Ok(duration) => duration.as_millis() as i64,
@@ -321,4 +346,18 @@ mod tests {
             other => panic!("expected error result, got {:?}", other),
         }
     }
+
+    #[test]
+    fn disabled_flow_rejection_result_carries_reason() {
+        let result = disabled_flow_rejection_result("exec-1".to_string(), 42, "maintenance");
+
+        assert_eq!(result.execution_identifier, "exec-1");
+        assert_eq!(result.flow_id, 42);
+        match result.result {
+            Some(execution_result::Result::Error(err)) => {
+                assert!(err.message.contains("maintenance"));
+            }
+            other => panic!("expected error result, got {:?}", other),
+        }
+    }
 }