miyabi_a2a/http/
routes.rs

1//! API route handlers
2
3use axum::{
4    extract::{Path, State},
5    http::StatusCode as AxumStatusCode,
6    response::IntoResponse,
7    Json,
8};
9use serde::{Deserialize, Serialize};
10
11use super::server::AppState;
12use super::websocket::{DashboardUpdate, TaskCancelEvent, TaskRetryEvent};
13use crate::storage::TaskUpdate;
14use crate::task::TaskStatus;
15use chrono::Utc;
16
17// Type alias to avoid conflicts with reqwest::StatusCode
18type StatusCode = AxumStatusCode;
19
20/// Maximum number of retry attempts allowed for a failed task
21const MAX_RETRY_COUNT: u32 = 3;
22
23/// Base delay for exponential backoff (in seconds)
24const BASE_RETRY_DELAY_SECS: u64 = 10;
25
26/// Custom error response wrapper to avoid type ambiguity with reqwest::StatusCode
27#[derive(Debug, Clone)]
28pub struct ApiError(AxumStatusCode);
29
30impl IntoResponse for ApiError {
31    fn into_response(self) -> axum::response::Response {
32        self.0.into_response()
33    }
34}
35
36impl From<AxumStatusCode> for ApiError {
37    fn from(status: AxumStatusCode) -> Self {
38        ApiError(status)
39    }
40}
41
42/// Structured error response for API endpoints
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub struct ErrorResponse {
45    /// HTTP status code
46    pub status: u16,
47    /// Machine-readable error code
48    pub error_code: String,
49    /// Human-readable error message
50    pub message: String,
51    /// Optional error details
52    #[serde(skip_serializing_if = "Option::is_none")]
53    pub details: Option<String>,
54    /// Optional request ID for tracing
55    #[serde(skip_serializing_if = "Option::is_none")]
56    pub request_id: Option<String>,
57    /// Timestamp when error occurred
58    pub timestamp: chrono::DateTime<Utc>,
59}
60
61impl ErrorResponse {
62    /// Create a new error response
63    pub fn new(
64        status: StatusCode,
65        error_code: impl Into<String>,
66        message: impl Into<String>,
67    ) -> Self {
68        Self {
69            status: status.as_u16(),
70            error_code: error_code.into(),
71            message: message.into(),
72            details: None,
73            request_id: None,
74            timestamp: Utc::now(),
75        }
76    }
77
78    /// Add error details
79    pub fn with_details(mut self, details: impl Into<String>) -> Self {
80        self.details = Some(details.into());
81        self
82    }
83
84    /// Add request ID for tracing
85    pub fn with_request_id(mut self, request_id: impl Into<String>) -> Self {
86        self.request_id = Some(request_id.into());
87        self
88    }
89}
90
91impl IntoResponse for ErrorResponse {
92    fn into_response(self) -> axum::response::Response {
93        let status = StatusCode::from_u16(self.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
94        (status, Json(self)).into_response()
95    }
96}
97
98/// Agent data structure
99#[derive(Debug, Clone, Serialize, Deserialize)]
100pub struct Agent {
101    pub id: u32,
102    pub name: String,
103    pub role: String,
104    pub category: AgentCategory,
105    pub status: AgentStatus,
106    pub tasks: u32,
107    pub color: AgentColor,
108    pub description: String,
109}
110
111#[derive(Debug, Clone, Serialize, Deserialize)]
112#[serde(rename_all = "lowercase")]
113pub enum AgentCategory {
114    Coding,
115    Business,
116}
117
118#[derive(Debug, Clone, Serialize, Deserialize)]
119#[serde(rename_all = "lowercase")]
120pub enum AgentStatus {
121    Active,
122    Working,
123    Idle,
124    Error,
125}
126
127#[derive(Debug, Clone, Serialize, Deserialize)]
128#[serde(rename_all = "lowercase")]
129pub enum AgentColor {
130    Leader,
131    Executor,
132    Analyst,
133    Support,
134}
135
136/// System status structure
137#[derive(Debug, Clone, Serialize, Deserialize)]
138pub struct SystemStatus {
139    pub status: String,
140    pub active_agents: u32,
141    pub total_agents: u32,
142    pub active_tasks: u32,
143    pub queued_tasks: u32,
144    pub task_throughput: f64,
145    pub avg_completion_time: f64,
146}
147
148/// Timeline event structure
149#[derive(Debug, Clone, Serialize, Deserialize)]
150pub struct TimelineEvent {
151    pub id: String,
152    #[serde(rename = "type")]
153    pub event_type: String,
154    pub message: String,
155    pub timestamp: String,
156    #[serde(skip_serializing_if = "Option::is_none")]
157    pub task_id: Option<String>,
158    #[serde(skip_serializing_if = "Option::is_none")]
159    pub agent_id: Option<String>,
160    #[serde(skip_serializing_if = "Option::is_none")]
161    pub agent_name: Option<String>,
162    #[serde(skip_serializing_if = "Option::is_none")]
163    pub agent_type: Option<String>,
164}
165
166/// Health check endpoint
167pub async fn health_check() -> (StatusCode, &'static str) {
168    (StatusCode::OK, "OK")
169}
170
171/// Get all agents endpoint
172pub async fn get_agents() -> Json<Vec<Agent>> {
173    // Fetch real data from GitHub API
174    match crate::http::fetch_real_agents().await {
175        Ok(agents) => Json(agents),
176        Err(e) => {
177            tracing::error!("Failed to fetch real agents: {}", e);
178            // Return empty array on error
179            Json(vec![])
180        },
181    }
182}
183
184/// Get system status endpoint
185pub async fn get_system_status() -> Json<SystemStatus> {
186    // Fetch real system status from GitHub API
187    match crate::http::fetch_real_system_status().await {
188        Ok(status) => Json(status),
189        Err(e) => {
190            tracing::error!("Failed to fetch real system status: {}", e);
191            // Return default status on error
192            Json(SystemStatus {
193                status: "error".to_string(),
194                active_agents: 0,
195                total_agents: 21,
196                active_tasks: 0,
197                queued_tasks: 0,
198                task_throughput: 0.0,
199                avg_completion_time: 0.0,
200            })
201        },
202    }
203}
204
205/// Get timeline events endpoint
206pub async fn get_events() -> Json<Vec<TimelineEvent>> {
207    // Fetch real events from GitHub API
208    match crate::http::fetch_real_events().await {
209        Ok(events) => Json(events),
210        Err(e) => {
211            tracing::error!("Failed to fetch real events: {}", e);
212            // Return empty array on error
213            Json(vec![])
214        },
215    }
216}
217
218/// DAG node structure for frontend
219#[derive(Debug, Clone, Serialize, Deserialize)]
220pub struct DagNode {
221    pub id: String,
222    pub label: String,
223    pub status: String, // "pending" | "working" | "completed" | "failed"
224    pub agent: String,
225    #[serde(rename = "agentType")]
226    pub agent_type: String,
227    // 🆕 Extended fields for full parameter mapping
228    pub priority: String, // "P0" | "P1" | "P2" | "P3"
229    #[serde(rename = "estimatedMinutes")]
230    pub estimated_minutes: u32, // Estimated task duration
231    pub description: String, // Task description (from issue body)
232    pub module: String,   // Module name (e.g., "Miyabi Agents", "Miyabi Core")
233    pub layer: String,    // Layer: "ui" | "logic" | "data" | "infra"
234}
235
236/// DAG edge structure for frontend
237#[derive(Debug, Clone, Serialize, Deserialize)]
238pub struct DagEdge {
239    pub from: String,
240    pub to: String,
241    #[serde(rename = "type")]
242    pub edge_type: String, // "depends_on"
243}
244
245/// DAG data structure for frontend
246#[derive(Debug, Clone, Serialize, Deserialize)]
247pub struct DagData {
248    #[serde(rename = "workflowId")]
249    pub workflow_id: String,
250    pub nodes: Vec<DagNode>,
251    pub edges: Vec<DagEdge>,
252}
253
254/// Get workflow DAG endpoint
255pub async fn get_workflow_dag() -> Json<DagData> {
256    // Fetch real DAG from GitHub API or internal state
257    // Note: fetch_real_workflow_dag already has fallback to sample DAG built in
258    match crate::http::fetch_real_workflow_dag().await {
259        Ok(dag_data) => Json(dag_data),
260        Err(e) => {
261            tracing::error!("Failed to fetch workflow DAG: {}", e);
262            // This should not happen since fetch_real_workflow_dag has internal fallback
263            // But as a last resort, return sample DAG
264            Json(crate::http::create_sample_dag_public())
265        },
266    }
267}
268
269// ===== Task Recovery Endpoints =====
270
271/// Task retry request payload
272#[derive(Debug, Clone, Serialize, Deserialize)]
273pub struct TaskRetryRequest {
274    /// Optional reason for retry
275    pub reason: Option<String>,
276}
277
278/// Task retry response
279#[derive(Debug, Clone, Serialize, Deserialize)]
280pub struct TaskRetryResponse {
281    /// Task ID that was retried
282    pub task_id: String,
283    /// Current task status after retry
284    pub status: String,
285    /// Response message
286    pub message: String,
287    /// Number of retry attempts
288    pub retry_count: u32,
289}
290
291/// Task cancel response
292#[derive(Debug, Clone, Serialize, Deserialize)]
293pub struct TaskCancelResponse {
294    /// Task ID that was cancelled
295    pub task_id: String,
296    /// Current task status after cancellation
297    pub status: String,
298    /// Response message
299    pub message: String,
300}
301
302/// Retry a failed task
303///
304/// # Arguments
305/// * `task_id` - The ID of the task to retry
306/// * `payload` - Optional retry reason
307///
308/// # Returns
309/// * `TaskRetryResponse` with task status and retry count
310///
311/// # Errors
312/// * `400 Bad Request` - Invalid task ID format
313/// * `404 Not Found` - Task does not exist
314/// * `409 Conflict` - Task is not in failed state
315/// * `429 Too Many Requests` - Retry limit exceeded (max: 3 attempts)
316/// * `500 Internal Server Error` - Storage operation failed
317pub async fn retry_task(
318    State(state): State<AppState>,
319    Path(task_id): Path<String>,
320    Json(payload): Json<TaskRetryRequest>,
321) -> Result<Json<TaskRetryResponse>, ErrorResponse> {
322    tracing::info!("Retrying task: {} (reason: {:?})", task_id, payload.reason);
323
324    // 1. Parse task_id as u64
325    let id = task_id.parse::<u64>().map_err(|e| {
326        tracing::error!("Invalid task ID format: {} - {}", task_id, e);
327        ErrorResponse::new(StatusCode::BAD_REQUEST, "INVALID_TASK_ID", "Invalid task ID format")
328            .with_details(format!("Task ID '{}' must be a valid integer: {}", task_id, e))
329    })?;
330
331    // 2. Get task from storage
332    let task = state.storage.get_task(id).await.map_err(|e| {
333        tracing::error!("Failed to fetch task {}: {}", id, e);
334        ErrorResponse::new(
335            StatusCode::INTERNAL_SERVER_ERROR,
336            "STORAGE_ERROR",
337            "Failed to retrieve task from storage",
338        )
339        .with_details(format!("Storage error for task {}: {}", id, e))
340    })?;
341
342    // 3. Check if task exists
343    let mut task = task.ok_or_else(|| {
344        tracing::warn!("Task {} not found", id);
345        ErrorResponse::new(
346            StatusCode::NOT_FOUND,
347            "TASK_NOT_FOUND",
348            format!("Task {} does not exist", id),
349        )
350    })?;
351
352    // 4. Check if task is in failed state
353    if task.status != TaskStatus::Failed {
354        tracing::warn!("Task {} is not in failed state (current: {:?})", id, task.status);
355        return Err(ErrorResponse::new(
356            StatusCode::CONFLICT,
357            "INVALID_TASK_STATE",
358            "Task must be in failed state to retry",
359        )
360        .with_details(format!("Current task status: {:?}", task.status)));
361    }
362
363    // 5. Check retry count limit
364    if task.retry_count >= MAX_RETRY_COUNT {
365        tracing::warn!(
366            "Task {} has reached max retry limit ({}/{})",
367            id,
368            task.retry_count,
369            MAX_RETRY_COUNT
370        );
371        return Err(ErrorResponse::new(
372            StatusCode::TOO_MANY_REQUESTS,
373            "MAX_RETRIES_EXCEEDED",
374            format!("Maximum retry limit of {} attempts reached", MAX_RETRY_COUNT),
375        )
376        .with_details(format!("Current retry count: {}/{}", task.retry_count, MAX_RETRY_COUNT)));
377    }
378
379    // 6. Increment retry_count
380    task.retry_count += 1;
381
382    // 7. Prepare retry reason (clone for WebSocket event later)
383    let retry_reason = payload.reason.clone();
384    let description = payload
385        .reason
386        .or(Some(format!("Retry attempt {} - Previous failure", task.retry_count)));
387
388    // 8. Update task status to Submitted for retry
389    let update = TaskUpdate {
390        status: Some(TaskStatus::Submitted),
391        description,
392        agent: None,
393        priority: None,
394        retry_count: Some(task.retry_count),
395    };
396
397    state.storage.update_task(id, update).await.map_err(|e| {
398        tracing::error!("Failed to update task {}: {}", id, e);
399        ErrorResponse::new(
400            StatusCode::INTERNAL_SERVER_ERROR,
401            "STORAGE_UPDATE_ERROR",
402            "Failed to update task status",
403        )
404        .with_details(format!("Storage update error for task {}: {}", id, e))
405    })?;
406
407    // 9. Calculate exponential backoff delay: base_delay * 2^retry_count
408    let delay_secs = BASE_RETRY_DELAY_SECS * 2u64.pow(task.retry_count - 1); // retry_count is already incremented
409    let next_retry_at = Utc::now() + chrono::Duration::seconds(delay_secs as i64);
410
411    // 10. Broadcast retry event via WebSocket
412    let retry_event = DashboardUpdate::TaskRetry {
413        event: TaskRetryEvent {
414            task_id: task_id.clone(),
415            retry_count: task.retry_count,
416            reason: retry_reason,
417            next_retry_at: Some(next_retry_at),
418            timestamp: Utc::now(),
419        },
420    };
421
422    if let Err(e) = state.ws_state.tx.send(retry_event) {
423        // Log error but don't fail the request - WebSocket broadcasting is not critical
424        tracing::warn!("Failed to broadcast retry event for task {}: {}", id, e);
425    }
426
427    tracing::info!(
428        "Task {} queued for retry (attempt {}, next retry at {})",
429        id,
430        task.retry_count,
431        next_retry_at
432    );
433
434    Ok(Json(TaskRetryResponse {
435        task_id: task_id.clone(),
436        status: "submitted".to_string(),
437        message: format!("Task {} has been queued for retry", task_id),
438        retry_count: task.retry_count,
439    }))
440}
441
442/// Cancel a running task
443///
444/// # Arguments
445/// * `task_id` - The ID of the task to cancel
446///
447/// # Returns
448/// * `TaskCancelResponse` with task status
449///
450/// # Errors
451/// * `400 Bad Request` - Invalid task ID format
452/// * `404 Not Found` - Task does not exist
453/// * `409 Conflict` - Task is not in cancellable state (must be Submitted or Working)
454/// * `500 Internal Server Error` - Storage operation failed
455pub async fn cancel_task(
456    State(state): State<AppState>,
457    Path(task_id): Path<String>,
458) -> Result<Json<TaskCancelResponse>, ErrorResponse> {
459    tracing::info!("Cancelling task: {}", task_id);
460
461    // 1. Parse task_id as u64
462    let id = task_id.parse::<u64>().map_err(|e| {
463        tracing::error!("Invalid task ID format: {} - {}", task_id, e);
464        ErrorResponse::new(StatusCode::BAD_REQUEST, "INVALID_TASK_ID", "Invalid task ID format")
465            .with_details(format!("Task ID '{}' must be a valid integer: {}", task_id, e))
466    })?;
467
468    // 2. Get task from storage
469    let task = state.storage.get_task(id).await.map_err(|e| {
470        tracing::error!("Failed to fetch task {}: {}", id, e);
471        ErrorResponse::new(
472            StatusCode::INTERNAL_SERVER_ERROR,
473            "STORAGE_ERROR",
474            "Failed to retrieve task from storage",
475        )
476        .with_details(format!("Storage error for task {}: {}", id, e))
477    })?;
478
479    // 3. Check if task exists
480    let task = task.ok_or_else(|| {
481        tracing::warn!("Task {} not found", id);
482        ErrorResponse::new(
483            StatusCode::NOT_FOUND,
484            "TASK_NOT_FOUND",
485            format!("Task {} does not exist", id),
486        )
487    })?;
488
489    // 4. Check if task is cancellable (must be in Submitted or Working state)
490    if task.status != TaskStatus::Submitted && task.status != TaskStatus::Working {
491        tracing::warn!("Task {} is not cancellable (current state: {:?})", id, task.status);
492        return Err(ErrorResponse::new(
493            StatusCode::CONFLICT,
494            "INVALID_TASK_STATE",
495            "Task must be in Submitted or Working state to cancel",
496        )
497        .with_details(format!(
498            "Current task status: {:?}. Only Submitted or Working tasks can be cancelled.",
499            task.status
500        )));
501    }
502
503    // 5. Update task status to Cancelled
504    let update = TaskUpdate {
505        status: Some(TaskStatus::Cancelled),
506        description: Some(format!("Task cancelled by user at {}", chrono::Utc::now())),
507        agent: None,
508        priority: None,
509        retry_count: None,
510    };
511
512    state.storage.update_task(id, update).await.map_err(|e| {
513        tracing::error!("Failed to update task {}: {}", id, e);
514        ErrorResponse::new(
515            StatusCode::INTERNAL_SERVER_ERROR,
516            "STORAGE_UPDATE_ERROR",
517            "Failed to update task status",
518        )
519        .with_details(format!("Storage update error for task {}: {}", id, e))
520    })?;
521
522    // 6. Send cancellation signal to task executor (future enhancement)
523    // TODO: Send cancellation signal to running agent via inter-process communication
524
525    // 7. Broadcast cancellation event via WebSocket
526    let cancel_event = DashboardUpdate::TaskCancel {
527        event: TaskCancelEvent {
528            task_id: task_id.clone(),
529            reason: format!("Task cancelled by user at {}", Utc::now()),
530            timestamp: Utc::now(),
531        },
532    };
533
534    if let Err(e) = state.ws_state.tx.send(cancel_event) {
535        // Log error but don't fail the request - WebSocket broadcasting is not critical
536        tracing::warn!("Failed to broadcast cancel event for task {}: {}", id, e);
537    }
538
539    tracing::info!("Task {} has been cancelled", id);
540
541    Ok(Json(TaskCancelResponse {
542        task_id: task_id.clone(),
543        status: "cancelled".to_string(),
544        message: format!("Task {} has been cancelled", task_id),
545    }))
546}
miyabi_a2a/http/routes.rs

miyabi_a2a/http/
routes.rs
