diff --git a/src/components/microservices/code-queue-mgr/src-rs/main.rs b/src/components/microservices/code-queue-mgr/src-rs/main.rs index dbf24581..897ad85a 100644 --- a/src/components/microservices/code-queue-mgr/src-rs/main.rs +++ b/src/components/microservices/code-queue-mgr/src-rs/main.rs @@ -20,6 +20,7 @@ const MAX_OUTPUT_PAGE_LIMIT: usize = 500; const MAX_PREVIEW_CHARS: usize = 6000; const WORKDIR_MAX_CHARS: usize = 512; const SCHEDULER_HEARTBEAT_STALE_MS: i64 = 5 * 60 * 1000; +const CODEX_STATS_TIME_ZONE: &str = "Asia/Shanghai"; #[derive(Clone)] struct Config { @@ -86,6 +87,19 @@ struct TaskMeta { task_json: Option, } +#[derive(Clone, Debug)] +struct DailyTaskStatsRow { + date: String, + executed_tasks: i64, + completed_tasks: i64, + retry_attempts: i64, + succeeded_tasks: i64, + failed_tasks: i64, + canceled_tasks: i64, + total_duration_ms: i64, + duration_samples: i64, +} + #[derive(Clone, Debug)] struct RequestUrl { path: String, @@ -593,6 +607,174 @@ fn timing_json(task: &TaskMeta) -> Value { }) } +fn codex_stats_days_from_url(url: &RequestUrl) -> i64 { + parse_i64(query_param(url, "days"), 14).clamp(1, 90) +} + +fn codex_stats_queue_id_from_url(url: &RequestUrl) -> Option { + query_param(url, "queueId") + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .map(|value| safe_queue_id(&value)) +} + +fn codex_stats_range_sql() -> &'static str { + " + WITH stats_range AS ( + SELECT + $2::int AS days, + to_char(now() AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS end_date, + (date_trunc('day', now() AT TIME ZONE 'Asia/Shanghai') - (($2::int - 1) * INTERVAL '1 day')) AT TIME ZONE 'Asia/Shanghai' AS start_at, + (date_trunc('day', now() AT TIME ZONE 'Asia/Shanghai') + INTERVAL '1 day') AT TIME ZONE 'Asia/Shanghai' AS end_at + ), + day_series AS ( + SELECT to_char(day::date, 'YYYY-MM-DD') AS date + FROM stats_range stats, + generate_series( + (stats.start_at AT TIME ZONE 'Asia/Shanghai')::date, + (stats.end_at AT TIME ZONE 'Asia/Shanghai')::date - 1, + INTERVAL '1 day' + ) AS day + ), + executed AS ( + SELECT to_char(started_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS date, COUNT(*)::bigint AS count + FROM unidesk_code_queue_tasks, stats_range + WHERE ($1::text IS NULL OR queue_id = $1) + AND started_at >= stats_range.start_at + AND started_at < stats_range.end_at + GROUP BY date + ), + completed AS ( + SELECT + to_char(COALESCE(finished_at, updated_at) AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS date, + COUNT(*)::bigint AS completed_tasks, + COUNT(*) FILTER (WHERE status = 'succeeded')::bigint AS succeeded_tasks, + COUNT(*) FILTER (WHERE status = 'failed')::bigint AS failed_tasks, + COUNT(*) FILTER (WHERE status = 'canceled')::bigint AS canceled_tasks, + COALESCE(SUM(GREATEST(EXTRACT(EPOCH FROM (COALESCE(finished_at, updated_at) - COALESCE(started_at, created_at))) * 1000, 0)), 0)::bigint AS total_duration_ms, + COUNT(CASE WHEN COALESCE(started_at, created_at) IS NOT NULL THEN 1 END)::bigint AS duration_samples + FROM unidesk_code_queue_tasks, stats_range + WHERE ($1::text IS NULL OR queue_id = $1) + AND status IN ('succeeded', 'failed', 'canceled') + AND COALESCE(finished_at, updated_at) >= stats_range.start_at + AND COALESCE(finished_at, updated_at) < stats_range.end_at + GROUP BY date + ), + retries AS ( + SELECT + to_char(updated_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS date, + SUM(GREATEST(current_attempt - 1, attempt_count - 1, 0))::bigint AS count + FROM unidesk_code_queue_tasks, stats_range + WHERE ($1::text IS NULL OR queue_id = $1) + AND updated_at >= stats_range.start_at + AND updated_at < stats_range.end_at + AND GREATEST(current_attempt - 1, attempt_count - 1, 0) > 0 + GROUP BY date + ) + SELECT + day_series.date, + COALESCE(executed.count, 0)::bigint AS executed_tasks, + COALESCE(completed.completed_tasks, 0)::bigint AS completed_tasks, + COALESCE(retries.count, 0)::bigint AS retry_attempts, + COALESCE(completed.succeeded_tasks, 0)::bigint AS succeeded_tasks, + COALESCE(completed.failed_tasks, 0)::bigint AS failed_tasks, + COALESCE(completed.canceled_tasks, 0)::bigint AS canceled_tasks, + COALESCE(completed.total_duration_ms, 0)::bigint AS total_duration_ms, + COALESCE(completed.duration_samples, 0)::bigint AS duration_samples, + (SELECT end_date FROM stats_range) AS end_date + FROM day_series + LEFT JOIN executed ON executed.date = day_series.date + LEFT JOIN completed ON completed.date = day_series.date + LEFT JOIN retries ON retries.date = day_series.date + ORDER BY day_series.date ASC + " +} + +fn rounded_average_ms(total_duration_ms: i64, duration_samples: i64) -> Value { + if duration_samples <= 0 { + return Value::Null; + } + json!((total_duration_ms as f64 / duration_samples as f64).round() as i64) +} + +fn codex_stats_summary_from_rows(days: i64, generated_at: &str, rows: &[DailyTaskStatsRow]) -> Value { + let mut total_executed = 0_i64; + let mut total_completed = 0_i64; + let mut total_retries = 0_i64; + let mut total_succeeded = 0_i64; + let mut total_failed = 0_i64; + let mut total_canceled = 0_i64; + let mut total_duration_ms = 0_i64; + let mut duration_samples = 0_i64; + let daily = rows + .iter() + .map(|row| { + total_executed += row.executed_tasks; + total_completed += row.completed_tasks; + total_retries += row.retry_attempts; + total_succeeded += row.succeeded_tasks; + total_failed += row.failed_tasks; + total_canceled += row.canceled_tasks; + total_duration_ms += row.total_duration_ms; + duration_samples += row.duration_samples; + json!({ + "date": row.date, + "executedTasks": row.executed_tasks, + "completedTasks": row.completed_tasks, + "retryAttempts": row.retry_attempts, + "succeededTasks": row.succeeded_tasks, + "failedTasks": row.failed_tasks, + "canceledTasks": row.canceled_tasks, + "totalDurationMs": row.total_duration_ms, + "durationSamples": row.duration_samples, + "avgDurationMs": rounded_average_ms(row.total_duration_ms, row.duration_samples) + }) + }) + .collect::>(); + let start_date = rows.first().map(|row| row.date.clone()).unwrap_or_default(); + let end_date = rows.last().map(|row| row.date.clone()).unwrap_or_default(); + json!({ + "generatedAt": generated_at, + "timezone": CODEX_STATS_TIME_ZONE, + "days": days, + "range": { "startDate": start_date, "endDate": end_date }, + "totals": { + "executedTasks": total_executed, + "completedTasks": total_completed, + "retryAttempts": total_retries, + "succeededTasks": total_succeeded, + "failedTasks": total_failed, + "canceledTasks": total_canceled, + "totalDurationMs": total_duration_ms, + "durationSamples": duration_samples, + "avgDurationMs": rounded_average_ms(total_duration_ms, duration_samples) + }, + "daily": daily + }) +} + +fn task_statistics_summary(state: &AppState, queue_id: Option, days: i64) -> Result { + let mut client = db_client(state, "unidesk-code-queue-mgr-rust-stats")?; + let rows = client + .query(codex_stats_range_sql(), &[&queue_id, &days]) + .map_err(|error| error.to_string())?; + let stats_rows = rows + .iter() + .map(|row| DailyTaskStatsRow { + date: row.get("date"), + executed_tasks: row.get("executed_tasks"), + completed_tasks: row.get("completed_tasks"), + retry_attempts: row.get("retry_attempts"), + succeeded_tasks: row.get("succeeded_tasks"), + failed_tasks: row.get("failed_tasks"), + canceled_tasks: row.get("canceled_tasks"), + total_duration_ms: row.get("total_duration_ms"), + duration_samples: row.get("duration_samples"), + }) + .collect::>(); + Ok(codex_stats_summary_from_rows(days, &now_iso(), &stats_rows)) +} + fn json_array_len(value: Option<&Value>, key: &str, fallback: i64) -> i64 { value .and_then(|json| json.get(key)) @@ -2316,8 +2498,15 @@ fn tasks_overview(state: &AppState, url: &RequestUrl) -> Result<(Value, u16), (V let selected_enabled = query_param(url, "selected").as_deref() != Some("0"); let prefer_id = query_param(url, "preferId").unwrap_or_default(); let include_selected_json = selected_enabled && !prefer_id.is_empty(); + let include_statistics = query_param(url, "stats").as_deref() != Some("0"); let (tasks, total, limit, offset) = load_tasks_for_list(state, url, false).map_err(|error| (json!({ "ok": false, "error": error }), 500))?; let summary = queue_summary(state).map_err(|error| (json!({ "ok": false, "error": error }), 500))?; + let statistics = if include_statistics { + task_statistics_summary(state, codex_stats_queue_id_from_url(url), codex_stats_days_from_url(url)) + .map_err(|error| (json!({ "ok": false, "error": error }), 500))? + } else { + Value::Null + }; let selected = if include_selected_json { let selected_id = if prefer_id.is_empty() { tasks.first().map(|task| task.id.clone()).unwrap_or_default() } else { prefer_id }; if selected_id.is_empty() { @@ -2358,7 +2547,7 @@ fn tasks_overview(state: &AppState, url: &RequestUrl) -> Result<(Value, u16), (V json!({ "ok": true, "queue": summary, - "statistics": { "days": 14, "source": "code-queue-mgr-rust", "skipped": true }, + "statistics": statistics, "tasks": tasks.iter().map(|task| task_list_response(state, task, true)).collect::>(), "selected": selected, "pagination": { @@ -2783,9 +2972,12 @@ fn handle_request(state: &AppState, request: &mut Request) -> (Value, u16) { (Method::Post, "/api/queues") => return create_queue(state, request).unwrap_or_else(|response| response), (Method::Post, "/api/queues/merge") => (json!({ "ok": false, "error": "queue merge requires /api/queues/:target/merge on rust mgr" }), 400), (Method::Get, "/api/tasks/overview") => return tasks_overview(state, &url).unwrap_or_else(|response| response), - (Method::Get, "/api/tasks/stats") => match queue_summary(state) { - Ok(summary) => (json!({ "ok": true, "statistics": { "days": 14, "source": "code-queue-mgr-rust", "skipped": true }, "queue": summary }), 200), - Err(error) => (json!({ "ok": false, "error": error }), 500), + (Method::Get, "/api/tasks/stats") => match ( + task_statistics_summary(state, codex_stats_queue_id_from_url(&url), codex_stats_days_from_url(&url)), + queue_summary(state), + ) { + (Ok(statistics), Ok(summary)) => (json!({ "ok": true, "statistics": statistics, "queue": summary }), 200), + (Err(error), _) | (_, Err(error)) => (json!({ "ok": false, "error": error }), 500), }, (Method::Get, "/api/tasks") => { match load_tasks_for_list(state, &url, false).and_then(|(tasks, total, limit, offset)| { @@ -3177,6 +3369,50 @@ mod tests { assert!(sql.contains("'{nextMode}', to_jsonb('retry'::text)"), "{sql}"); } + #[test] + fn stats_summary_contract_exposes_daily_buckets_and_totals() { + let stats = codex_stats_summary_from_rows( + 2, + "2026-05-20T00:00:00.000Z", + &[ + DailyTaskStatsRow { + date: "2026-05-19".to_string(), + executed_tasks: 2, + completed_tasks: 1, + retry_attempts: 1, + succeeded_tasks: 1, + failed_tasks: 0, + canceled_tasks: 0, + total_duration_ms: 120_000, + duration_samples: 1, + }, + DailyTaskStatsRow { + date: "2026-05-20".to_string(), + executed_tasks: 3, + completed_tasks: 2, + retry_attempts: 2, + succeeded_tasks: 1, + failed_tasks: 1, + canceled_tasks: 0, + total_duration_ms: 480_000, + duration_samples: 2, + }, + ], + ); + + assert_eq!(stats.get("timezone").and_then(Value::as_str), Some("Asia/Shanghai")); + assert_eq!(stats.pointer("/range/startDate").and_then(Value::as_str), Some("2026-05-19")); + assert_eq!(stats.pointer("/range/endDate").and_then(Value::as_str), Some("2026-05-20")); + assert_eq!(stats.pointer("/totals/executedTasks").and_then(Value::as_i64), Some(5)); + assert_eq!(stats.pointer("/totals/completedTasks").and_then(Value::as_i64), Some(3)); + assert_eq!(stats.pointer("/totals/retryAttempts").and_then(Value::as_i64), Some(3)); + assert_eq!(stats.pointer("/totals/failedTasks").and_then(Value::as_i64), Some(1)); + assert_eq!(stats.pointer("/totals/avgDurationMs").and_then(Value::as_i64), Some(200_000)); + let daily = stats.get("daily").and_then(Value::as_array).expect("daily buckets"); + assert_eq!(daily.len(), 2); + assert_eq!(daily[1].get("avgDurationMs").and_then(Value::as_i64), Some(240_000)); + } + #[test] fn manual_retry_canceled_task_becomes_scheduler_runnable() { let state = test_state();