orch/store/
metrics.rs

1use std::collections::HashMap;
2
3use super::TaskStore;
4#[cfg(test)]
5use chrono::Datelike;
6use chrono::Utc;
7use serde::{Deserialize, Serialize};
8use sqlx::Row;
9
10/// Parameters for inserting a new task metric record.
11#[derive(Debug, Clone)]
12pub struct InsertTaskMetric<'a> {
13    pub repo: &'a str,
14    pub task_id: &'a str,
15    pub agent: &'a str,
16    pub model: Option<&'a str>,
17    pub complexity: Option<&'a str>,
18    pub outcome: &'a str,
19    pub duration_seconds: f64,
20    pub started_at: &'a chrono::DateTime<Utc>,
21    pub completed_at: &'a chrono::DateTime<Utc>,
22    pub attempts: i32,
23    pub files_changed: i32,
24    pub error_type: Option<&'a str>,
25    pub input_tokens: Option<i64>,
26    pub output_tokens: Option<i64>,
27    pub input_cost_usd: Option<f64>,
28    pub output_cost_usd: Option<f64>,
29    pub total_cost_usd: Option<f64>,
30}
31
32/// Metrics summary for the CLI output.
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct MetricsSummary {
35    pub tasks_completed_24h: i64,
36    pub tasks_failed_24h: i64,
37    pub avg_duration_simple: Option<f64>,
38    pub avg_duration_medium: Option<f64>,
39    pub avg_duration_complex: Option<f64>,
40    pub agent_stats: Vec<AgentStat>,
41    pub rate_limits_24h: i64,
42}
43
44/// Agent statistics from metrics.
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct AgentStat {
47    pub agent: String,
48    pub total_runs: i64,
49    pub success_count: i64,
50    pub success_rate: f64,
51}
52
53/// Slow task info for pattern detection.
54#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct SlowTaskInfo {
56    pub task_id: String,
57    pub agent: String,
58    pub complexity: Option<String>,
59    pub duration_seconds: f64,
60}
61
62/// Error type distribution for pattern detection.
63#[derive(Debug, Clone, Serialize, Deserialize)]
64pub struct ErrorStat {
65    pub error_type: Option<String>,
66    pub count: i64,
67}
68
69/// Task with high review cycle count (persistent review loop).
70#[derive(Debug, Clone, Serialize, Deserialize)]
71pub struct HighReviewCycleTask {
72    pub external_id: Option<String>,
73    pub agent: Option<String>,
74    pub review_cycles: i64,
75    pub title: String,
76}
77
78/// Cost summary across multiple time periods.
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct CostSummary {
81    pub periods: Vec<CostPeriod>,
82}
83
84/// Cost data for a single time period.
85#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct CostPeriod {
87    pub label: String,
88    pub input_tokens: i64,
89    pub output_tokens: i64,
90    pub input_cost_usd: f64,
91    pub output_cost_usd: f64,
92    pub total_cost_usd: f64,
93    pub task_count: i64,
94}
95
96/// Cost breakdown by a grouping dimension (agent, model, etc.).
97#[derive(Debug, Clone, Serialize, Deserialize)]
98pub struct CostByGroup {
99    pub name: String,
100    pub input_tokens: i64,
101    pub output_tokens: i64,
102    pub total_cost_usd: f64,
103    pub task_count: i64,
104}
105
106/// Build the week-scoped KV key for the self-improvement counter.
107/// Test-only helper.
108#[cfg(test)]
109fn self_improvement_key() -> String {
110    let now = Utc::now();
111    format!(
112        "self_improvement_issues_{}_w{}",
113        now.iso_week().year(),
114        now.iso_week().week()
115    )
116}
117
118impl TaskStore {
119    pub async fn cost_summary(&self, repo: &str) -> anyhow::Result<(i64, i64, f64)> {
120        let row = sqlx::query(
121            "SELECT
122            COALESCE(SUM(input_tokens), 0) as total_input,
123            COALESCE(SUM(output_tokens), 0) as total_output,
124            COALESCE(SUM(total_cost_usd), 0.0) as total_cost
125         FROM tasks WHERE repo = ?",
126        )
127        .bind(repo)
128        .fetch_one(&self.pool)
129        .await?;
130
131        Ok((
132            row.try_get::<i64, _>("total_input")?,
133            row.try_get::<i64, _>("total_output")?,
134            row.try_get::<f64, _>("total_cost")?,
135        ))
136    }
137
138    /// Count tasks by status for a repo.
139    /// Returns a map of status string → count.
140    /// Only needed by tests.
141    #[cfg(test)]
142    pub async fn status_counts(&self, repo: &str) -> anyhow::Result<HashMap<String, i64>> {
143        let rows =
144            sqlx::query("SELECT status, COUNT(*) as cnt FROM tasks WHERE repo = ? GROUP BY status")
145                .bind(repo)
146                .fetch_all(&self.pool)
147                .await?;
148
149        let mut map = HashMap::new();
150        for row in &rows {
151            let status: String = row.try_get("status")?;
152            let count: i64 = row.try_get("cnt")?;
153            map.insert(status, count);
154        }
155        Ok(map)
156    }
157
158    pub async fn insert_task_metric(&self, metric: &InsertTaskMetric<'_>) -> anyhow::Result<i64> {
159        let mut tx = self.begin_write().await?;
160
161        sqlx::query(
162            "DELETE FROM task_metrics WHERE datetime(completed_at) < datetime('now', '-90 days')",
163        )
164        .execute(&mut *tx)
165        .await?;
166
167        let row = sqlx::query(
168        "INSERT INTO task_metrics (repo, task_id, agent, model, complexity, outcome, duration_seconds,
169         started_at, completed_at, attempts, files_changed, error_type,
170         input_tokens, output_tokens, input_cost_usd, output_cost_usd, total_cost_usd)
171         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
172         RETURNING id",
173    )
174    .bind(metric.repo)
175    .bind(metric.task_id)
176    .bind(metric.agent)
177    .bind(metric.model)
178    .bind(metric.complexity)
179    .bind(metric.outcome)
180    .bind(metric.duration_seconds)
181    .bind(metric.started_at.to_rfc3339())
182    .bind(metric.completed_at.to_rfc3339())
183    .bind(metric.attempts)
184    .bind(metric.files_changed)
185    .bind(metric.error_type)
186    .bind(metric.input_tokens)
187    .bind(metric.output_tokens)
188    .bind(metric.input_cost_usd)
189    .bind(metric.output_cost_usd)
190    .bind(metric.total_cost_usd)
191    .fetch_one(&mut *tx)
192    .await?;
193
194        tx.commit().await?;
195        Ok(row.try_get("id")?)
196    }
197
198    /// Get aggregated metrics for a configurable time window.
199    ///
200    /// `hours` controls how far back to look (e.g. 24 for 24 h, 168 for 7 days).
201    /// The interval string is built from a `u32` config value, not user input.
202    pub async fn get_metrics_summary(&self, hours: u32) -> anyhow::Result<MetricsSummary> {
203        let interval = format!("-{hours} hours");
204
205        let row = sqlx::query(
206            "SELECT
207                COALESCE(SUM(CASE WHEN outcome = 'success' THEN 1 ELSE 0 END), 0) AS completed,
208                COALESCE(SUM(CASE WHEN outcome != 'success' THEN 1 ELSE 0 END), 0) AS failed,
209                AVG(CASE WHEN complexity = 'simple' THEN duration_seconds END) AS avg_simple,
210                AVG(CASE WHEN complexity = 'medium' THEN duration_seconds END) AS avg_medium,
211                AVG(CASE WHEN complexity = 'complex' THEN duration_seconds END) AS avg_complex
212             FROM task_metrics
213             WHERE datetime(completed_at) >= datetime('now', ?)",
214        )
215        .bind(&interval)
216        .fetch_one(&self.pool)
217        .await?;
218
219        let completed: i64 = row.try_get("completed")?;
220        let failed: i64 = row.try_get("failed")?;
221        // AVG over a filtered subset is legitimately NULL when no rows match the complexity.
222        let avg_simple: Option<f64> = row.try_get("avg_simple")?;
223        let avg_medium: Option<f64> = row.try_get("avg_medium")?;
224        let avg_complex: Option<f64> = row.try_get("avg_complex")?;
225
226        let agent_rows = sqlx::query(
227            "SELECT agent, COUNT(*) as total,
228                COALESCE(SUM(CASE WHEN outcome = 'success' THEN 1 ELSE 0 END), 0) as success_count
229             FROM task_metrics
230             WHERE datetime(completed_at) >= datetime('now', ?)
231             GROUP BY agent",
232        )
233        .bind(&interval)
234        .fetch_all(&self.pool)
235        .await?;
236
237        #[expect(clippy::cast_precision_loss, reason = "a display percentage")]
238        let agent_stats: Vec<AgentStat> = agent_rows
239            .iter()
240            .map(|row| {
241                let total: i64 = row.try_get("total")?;
242                let success: i64 = row.try_get("success_count")?;
243                Ok::<AgentStat, sqlx::Error>(AgentStat {
244                    agent: row.try_get("agent")?,
245                    total_runs: total,
246                    success_count: success,
247                    success_rate: if total > 0 {
248                        (success as f64 / total as f64) * 100.0
249                    } else {
250                        0.0
251                    },
252                })
253            })
254            .collect::<Result<Vec<_>, _>>()?;
255
256        let rate_limit_count: (i64,) = sqlx::query_as(
257            "SELECT COUNT(*) FROM rate_limits WHERE occurred_at >= datetime('now', ?)",
258        )
259        .bind(&interval)
260        .fetch_one(&self.pool)
261        .await?;
262
263        Ok(MetricsSummary {
264            tasks_completed_24h: completed,
265            tasks_failed_24h: failed,
266            avg_duration_simple: avg_simple,
267            avg_duration_medium: avg_medium,
268            avg_duration_complex: avg_complex,
269            agent_stats,
270            rate_limits_24h: rate_limit_count.0,
271        })
272    }
273
274    /// Get aggregated metrics for the last 24 hours.
275    /// Test-only convenience wrapper — delegates to `get_metrics_summary(24)` so the
276    /// test exercises the same code path as production.
277    #[cfg(test)]
278    pub async fn get_metrics_summary_24h(&self) -> anyhow::Result<MetricsSummary> {
279        self.get_metrics_summary(24).await
280    }
281
282    /// Get metrics summary for a configurable time window, filtered by repo.
283    ///
284    /// `hours` controls how far back to look. Uses `COALESCE(NULLIF(m.repo, ''), t.repo)`
285    /// join pattern so older metric rows without a repo column still match.
286    ///
287    /// Optimized to use a single CTE query with conditional aggregation instead of 7 sequential queries.
288    pub async fn get_metrics_summary_by_repo(
289        &self,
290        repo: &str,
291        hours: u32,
292    ) -> anyhow::Result<MetricsSummary> {
293        let interval = format!("-{hours} hours");
294
295        let rows = sqlx::query(
296            "WITH base AS (
297                SELECT m.outcome, m.complexity, m.duration_seconds, m.agent,
298                    COALESCE(NULLIF(m.repo, ''), t.repo) AS resolved_repo
299                FROM task_metrics m
300                LEFT JOIN tasks t ON m.task_id = CAST(t.id AS TEXT)
301                    OR (m.task_id = t.external_id AND NOT EXISTS (
302                        SELECT 1 FROM tasks t2 WHERE t2.id = CAST(m.task_id AS INTEGER) AND t2.repo = t.repo
303                    ))
304                WHERE datetime(m.completed_at) >= datetime('now', ?)
305            )
306            SELECT
307                COALESCE(SUM(CASE WHEN outcome = 'success' THEN 1 ELSE 0 END), 0) AS completed,
308                COALESCE(SUM(CASE WHEN outcome != 'success' THEN 1 ELSE 0 END), 0) AS failed,
309                AVG(CASE WHEN complexity = 'simple' THEN duration_seconds END) AS avg_simple,
310                AVG(CASE WHEN complexity = 'medium' THEN duration_seconds END) AS avg_medium,
311                AVG(CASE WHEN complexity = 'complex' THEN duration_seconds END) AS avg_complex
312            FROM base
313            WHERE resolved_repo = ?",
314        )
315        .bind(&interval)
316        .bind(repo)
317        .fetch_one(&self.pool)
318        .await?;
319
320        let completed: i64 = rows.try_get("completed")?;
321        let failed: i64 = rows.try_get("failed")?;
322        // AVG over a filtered subset is legitimately NULL when no rows match the complexity.
323        let avg_simple: Option<f64> = rows.try_get("avg_simple")?;
324        let avg_medium: Option<f64> = rows.try_get("avg_medium")?;
325        let avg_complex: Option<f64> = rows.try_get("avg_complex")?;
326
327        let agent_rows = sqlx::query(
328            "SELECT m.agent, COUNT(*) as total,
329                COALESCE(SUM(CASE WHEN m.outcome = 'success' THEN 1 ELSE 0 END), 0) as success_count
330             FROM task_metrics m
331             LEFT JOIN tasks t ON m.task_id = CAST(t.id AS TEXT)
332                 OR (m.task_id = t.external_id AND NOT EXISTS (
333                     SELECT 1 FROM tasks t2 WHERE t2.id = CAST(m.task_id AS INTEGER) AND t2.repo = t.repo
334                 ))
335             WHERE datetime(m.completed_at) >= datetime('now', ?)
336             AND COALESCE(NULLIF(m.repo, ''), t.repo) = ?
337             GROUP BY m.agent",
338        )
339        .bind(&interval)
340        .bind(repo)
341        .fetch_all(&self.pool)
342        .await?;
343
344        #[expect(clippy::cast_precision_loss, reason = "a display percentage")]
345        let agent_stats: Vec<AgentStat> = agent_rows
346            .iter()
347            .map(|row| {
348                let total: i64 = row.try_get("total")?;
349                let success: i64 = row.try_get("success_count")?;
350                Ok::<AgentStat, sqlx::Error>(AgentStat {
351                    agent: row.try_get("agent")?,
352                    total_runs: total,
353                    success_count: success,
354                    success_rate: if total > 0 {
355                        (success as f64 / total as f64) * 100.0
356                    } else {
357                        0.0
358                    },
359                })
360            })
361            .collect::<Result<Vec<_>, _>>()?;
362
363        let rate_limits: (i64,) = sqlx::query_as(
364            "SELECT COUNT(*)
365             FROM rate_limits r
366             LEFT JOIN tasks t ON (
367                 -- If any task exists with this value as an external_id, prefer
368                 -- external_id matching (global preference). This avoids mapping a
369                 -- rate-limit that references an external task to an unrelated
370                 -- repo that merely happens to have an internal numeric id equal
371                 -- to the same string.
372                 (EXISTS (SELECT 1 FROM tasks te WHERE te.external_id = r.task_id) AND r.task_id = t.external_id)
373                 OR
374                 (NOT EXISTS (SELECT 1 FROM tasks te WHERE te.external_id = r.task_id) AND r.task_id = CAST(t.id AS TEXT))
375             )
376             WHERE r.occurred_at >= datetime('now', ?)
377             AND t.repo = ?",
378        )
379        .bind(&interval)
380        .bind(repo)
381        .fetch_one(&self.pool)
382        .await?;
383
384        Ok(MetricsSummary {
385            tasks_completed_24h: completed,
386            tasks_failed_24h: failed,
387            avg_duration_simple: avg_simple,
388            avg_duration_medium: avg_medium,
389            avg_duration_complex: avg_complex,
390            agent_stats,
391            rate_limits_24h: rate_limits.0,
392        })
393    }
394
395    /// Get cost summary filtered to a specific repository for a configurable window.
396    ///
397    /// Uses the same `COALESCE(NULLIF(m.repo, ''), t.repo)` join pattern as
398    /// `get_metrics_summary_by_repo` so older metrics rows that lack a `repo`
399    /// column value are still matched via the tasks table.
400    pub async fn get_cost_summary_by_repo(
401        &self,
402        repo: &str,
403        hours: u32,
404    ) -> anyhow::Result<CostSummary> {
405        let interval = format!("-{hours} hours");
406        let row = sqlx::query(
407            "SELECT COALESCE(SUM(m.input_tokens), 0) as input_tokens,
408                COALESCE(SUM(m.output_tokens), 0) as output_tokens,
409                COALESCE(SUM(m.input_cost_usd), 0.0) as input_cost_usd,
410                COALESCE(SUM(m.output_cost_usd), 0.0) as output_cost_usd,
411                COALESCE(SUM(m.total_cost_usd), 0.0) as total_cost_usd,
412                COUNT(*) as task_count
413         FROM task_metrics m
414         LEFT JOIN tasks t ON m.task_id = CAST(t.id AS TEXT)
415             OR (m.task_id = t.external_id AND NOT EXISTS (
416                 SELECT 1 FROM tasks t2 WHERE t2.id = CAST(m.task_id AS INTEGER) AND t2.repo = t.repo
417             ))
418         WHERE datetime(m.completed_at) >= datetime('now', ?)
419         AND COALESCE(NULLIF(m.repo, ''), t.repo) = ?",
420        )
421        .bind(&interval)
422        .bind(repo)
423        .fetch_one(&self.pool)
424        .await?;
425
426        Ok(CostSummary {
427            periods: vec![CostPeriod {
428                label: if hours == 24 {
429                    "24h".to_string()
430                } else if hours.is_multiple_of(24) {
431                    format!("{}d", hours / 24)
432                } else {
433                    format!("{hours}h")
434                },
435                input_tokens: row.try_get("input_tokens")?,
436                output_tokens: row.try_get("output_tokens")?,
437                input_cost_usd: row.try_get("input_cost_usd")?,
438                output_cost_usd: row.try_get("output_cost_usd")?,
439                total_cost_usd: row.try_get("total_cost_usd")?,
440                task_count: row.try_get("task_count")?,
441            }],
442        })
443    }
444
445    /// Record a rate limit event. Prunes records older than 30 days.
446    pub async fn record_rate_limit(
447        &self,
448        agent: &str,
449        limit_type: &str,
450        task_id: Option<&str>,
451    ) -> anyhow::Result<i64> {
452        let mut tx = self.begin_write().await?;
453
454        sqlx::query("DELETE FROM rate_limits WHERE occurred_at < datetime('now', '-30 days')")
455            .execute(&mut *tx)
456            .await?;
457
458        let row = sqlx::query(
459            "INSERT INTO rate_limits (agent, limit_type, occurred_at, task_id)
460         VALUES (?, ?, datetime('now'), ?)
461         RETURNING id",
462        )
463        .bind(agent)
464        .bind(limit_type)
465        .bind(task_id)
466        .fetch_one(&mut *tx)
467        .await?;
468
469        tx.commit().await?;
470        Ok(row.try_get("id")?)
471    }
472
473    /// Record a rate limit event at an explicit SQLite datetime (`YYYY-MM-DD HH:MM:SS`)
474    pub async fn record_rate_limit_at(
475        &self,
476        agent: &str,
477        limit_type: &str,
478        task_id: Option<&str>,
479        occurred_at: &str,
480    ) -> anyhow::Result<i64> {
481        let mut tx = self.begin_write().await?;
482
483        sqlx::query("DELETE FROM rate_limits WHERE occurred_at < datetime('now', '-30 days')")
484            .execute(&mut *tx)
485            .await?;
486
487        let row = sqlx::query(
488            "INSERT INTO rate_limits (agent, limit_type, occurred_at, task_id)
489         VALUES (?, ?, ?, ?)
490         RETURNING id",
491        )
492        .bind(agent)
493        .bind(limit_type)
494        .bind(occurred_at)
495        .bind(task_id)
496        .fetch_one(&mut *tx)
497        .await?;
498
499        tx.commit().await?;
500        Ok(row.try_get("id")?)
501    }
502
503    /// Distinct agents that have at least one `rate_limits` row
504    pub async fn rate_limit_agents(&self) -> anyhow::Result<Vec<String>> {
505        let rows = sqlx::query("SELECT DISTINCT agent FROM rate_limits")
506            .fetch_all(&self.pool)
507            .await?;
508        let mut agents = Vec::with_capacity(rows.len());
509        for row in &rows {
510            agents.push(row.try_get("agent")?);
511        }
512        Ok(agents)
513    }
514
515    /// Count recent rate-limit events per agent within the given window (in hours).
516    ///
517    /// Only `limit_type = 'rate'` rows count. Failures, auth errors and
518    /// timeouts keep their own cooldowns and must not mark an agent degraded.
519    /// Rows at or before `rate_limits_cleared_at:{agent}` are ignored so
520    /// `orch cooldown clear` can un-degrade without deleting metrics
521    pub async fn recent_rate_limit_counts(
522        &self,
523        window_hours: u32,
524    ) -> anyhow::Result<HashMap<String, i64>> {
525        // SQLite datetime modifiers don't support parameterised intervals, so
526        // we build the interval string and embed it directly.  The value comes
527        // from a config integer, not user input, so this is safe.
528        let interval = format!("-{window_hours} hours");
529        let rows = sqlx::query(
530            "SELECT r.agent, COUNT(*) as cnt
531             FROM rate_limits r
532             LEFT JOIN kv ON kv.key = 'rate_limits_cleared_at:' || r.agent
533             WHERE r.occurred_at >= datetime('now', ?)
534               AND r.limit_type = 'rate'
535               AND (
536                 kv.value IS NULL
537                 OR r.occurred_at > datetime(CAST(kv.value AS INTEGER), 'unixepoch')
538               )
539             GROUP BY r.agent",
540        )
541        .bind(&interval)
542        .fetch_all(&self.pool)
543        .await?;
544
545        let mut map = HashMap::new();
546        for row in &rows {
547            let agent: String = row.try_get("agent")?;
548            let cnt: i64 = row.try_get("cnt")?;
549            map.insert(agent, cnt);
550        }
551        Ok(map)
552    }
553
554    /// Get slow tasks (top 10 longest running) for a configurable time window.
555    pub async fn get_slow_tasks(&self, hours: u32) -> anyhow::Result<Vec<SlowTaskInfo>> {
556        let interval = format!("-{hours} hours");
557        let rows = sqlx::query(
558            "SELECT task_id, agent, complexity, duration_seconds
559         FROM task_metrics
560         WHERE datetime(completed_at) >= datetime('now', ?) AND outcome = 'success'
561         ORDER BY duration_seconds DESC
562         LIMIT 10",
563        )
564        .bind(&interval)
565        .fetch_all(&self.pool)
566        .await?;
567
568        rows.iter()
569            .map(|row| {
570                Ok(SlowTaskInfo {
571                    task_id: row.try_get("task_id")?,
572                    agent: row.try_get("agent")?,
573                    // complexity is a nullable column — NULL is a valid value
574                    complexity: row.try_get("complexity")?,
575                    duration_seconds: row.try_get("duration_seconds")?,
576                })
577            })
578            .collect()
579    }
580
581    /// Get slow tasks (top 10 longest running) from the last 7 days.
582    /// Test-only wrapper.
583    #[cfg(test)]
584    pub async fn get_slow_tasks_7d(&self) -> anyhow::Result<Vec<SlowTaskInfo>> {
585        let rows = sqlx::query(
586            "SELECT task_id, agent, complexity, duration_seconds
587         FROM task_metrics
588         WHERE datetime(completed_at) >= datetime('now', '-7 days') AND outcome = 'success'
589         ORDER BY duration_seconds DESC
590         LIMIT 10",
591        )
592        .fetch_all(&self.pool)
593        .await?;
594
595        rows.iter()
596            .map(|row| {
597                Ok(SlowTaskInfo {
598                    task_id: row.try_get("task_id")?,
599                    agent: row.try_get("agent")?,
600                    // complexity is a nullable column — NULL is a valid value
601                    complexity: row.try_get("complexity")?,
602                    duration_seconds: row.try_get("duration_seconds")?,
603                })
604            })
605            .collect()
606    }
607
608    /// Get error type distribution for a configurable time window.
609    pub async fn get_error_distribution(&self, hours: u32) -> anyhow::Result<Vec<ErrorStat>> {
610        let interval = format!("-{hours} hours");
611        let rows = sqlx::query(
612            "SELECT error_type, COUNT(*) as count
613         FROM task_metrics
614         WHERE datetime(completed_at) >= datetime('now', ?)
615           AND outcome != 'success'
616           AND error_type IS NOT NULL
617         GROUP BY error_type
618         ORDER BY count DESC",
619        )
620        .bind(&interval)
621        .fetch_all(&self.pool)
622        .await?;
623
624        rows.iter()
625            .map(|row| {
626                Ok(ErrorStat {
627                    // error_type is a nullable column — NULL is a valid value
628                    error_type: row.try_get("error_type")?,
629                    count: row.try_get("count")?,
630                })
631            })
632            .collect()
633    }
634
635    /// Get cost summary over multiple time windows (24h, 7d, 30d).
636    ///
637    /// Uses separate static queries per window because SQLite's `datetime()`
638    /// modifier argument cannot be parameterised via `?` bind.
639    pub async fn get_cost_summary(&self) -> anyhow::Result<CostSummary> {
640        // Each entry: (static SQL with the interval baked in, label)
641        let windows: &[(&'static str, &str)] = &[
642            (
643                "SELECT COALESCE(SUM(input_tokens), 0) as input_tokens,
644                    COALESCE(SUM(output_tokens), 0) as output_tokens,
645                    COALESCE(SUM(input_cost_usd), 0.0) as input_cost_usd,
646                    COALESCE(SUM(output_cost_usd), 0.0) as output_cost_usd,
647                    COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
648                    COUNT(*) as task_count
649             FROM task_metrics
650             WHERE datetime(completed_at) >= datetime('now', '-24 hours')",
651                "24h",
652            ),
653            (
654                "SELECT COALESCE(SUM(input_tokens), 0) as input_tokens,
655                    COALESCE(SUM(output_tokens), 0) as output_tokens,
656                    COALESCE(SUM(input_cost_usd), 0.0) as input_cost_usd,
657                    COALESCE(SUM(output_cost_usd), 0.0) as output_cost_usd,
658                    COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
659                    COUNT(*) as task_count
660             FROM task_metrics
661             WHERE datetime(completed_at) >= datetime('now', '-7 days')",
662                "7d",
663            ),
664            (
665                "SELECT COALESCE(SUM(input_tokens), 0) as input_tokens,
666                    COALESCE(SUM(output_tokens), 0) as output_tokens,
667                    COALESCE(SUM(input_cost_usd), 0.0) as input_cost_usd,
668                    COALESCE(SUM(output_cost_usd), 0.0) as output_cost_usd,
669                    COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
670                    COUNT(*) as task_count
671             FROM task_metrics
672             WHERE datetime(completed_at) >= datetime('now', '-30 days')",
673                "30d",
674            ),
675        ];
676
677        let mut periods = Vec::new();
678        for (query, label) in windows {
679            let row = sqlx::query(*query).fetch_one(&self.pool).await?;
680            periods.push(CostPeriod {
681                label: label.to_string(),
682                input_tokens: row.try_get("input_tokens")?,
683                output_tokens: row.try_get("output_tokens")?,
684                input_cost_usd: row.try_get("input_cost_usd")?,
685                output_cost_usd: row.try_get("output_cost_usd")?,
686                total_cost_usd: row.try_get("total_cost_usd")?,
687                task_count: row.try_get("task_count")?,
688            });
689        }
690
691        Ok(CostSummary { periods })
692    }
693
694    /// Get cost breakdown by agent.
695    pub async fn get_cost_by_agent(&self) -> anyhow::Result<Vec<CostByGroup>> {
696        let rows = sqlx::query(
697            "SELECT agent as name,
698                COALESCE(SUM(input_tokens), 0) as input_tokens,
699                COALESCE(SUM(output_tokens), 0) as output_tokens,
700                COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
701                COUNT(*) as task_count
702         FROM task_metrics
703         GROUP BY agent
704         ORDER BY SUM(total_cost_usd) DESC",
705        )
706        .fetch_all(&self.pool)
707        .await?;
708
709        rows.iter()
710            .map(|row| {
711                Ok(CostByGroup {
712                    name: row.try_get("name")?,
713                    input_tokens: row.try_get("input_tokens")?,
714                    output_tokens: row.try_get("output_tokens")?,
715                    total_cost_usd: row.try_get("total_cost_usd")?,
716                    task_count: row.try_get("task_count")?,
717                })
718            })
719            .collect()
720    }
721
722    /// Get cost breakdown by model.
723    pub async fn get_cost_by_model(&self) -> anyhow::Result<Vec<CostByGroup>> {
724        let rows = sqlx::query(
725            "SELECT COALESCE(model, 'unknown') as name,
726                COALESCE(SUM(input_tokens), 0) as input_tokens,
727                COALESCE(SUM(output_tokens), 0) as output_tokens,
728                COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
729                COUNT(*) as task_count
730         FROM task_metrics
731         GROUP BY model
732         ORDER BY SUM(total_cost_usd) DESC",
733        )
734        .fetch_all(&self.pool)
735        .await?;
736
737        rows.iter()
738            .map(|row| {
739                Ok(CostByGroup {
740                    name: row.try_get("name")?,
741                    input_tokens: row.try_get("input_tokens")?,
742                    output_tokens: row.try_get("output_tokens")?,
743                    total_cost_usd: row.try_get("total_cost_usd")?,
744                    task_count: row.try_get("task_count")?,
745                })
746            })
747            .collect()
748    }
749
750    /// Get the duration_seconds from the most recent metric row for a task.
751    /// Returns `None` when no metric has been recorded yet.
752    pub async fn latest_task_metric_duration(&self, task_id: &str) -> Option<f64> {
753        sqlx::query_scalar(
754            "SELECT duration_seconds FROM task_metrics WHERE task_id = ? ORDER BY completed_at DESC LIMIT 1",
755        )
756        .bind(task_id)
757        .fetch_optional(&self.pool)
758        .await
759        .ok()
760        .flatten()
761    }
762
763    /// Get count of self-improvement issues created this week.
764    /// Test-only helper that reads the KV counter.
765    #[cfg(test)]
766    pub async fn count_self_improvement_issues_7d(&self) -> anyhow::Result<i64> {
767        let key = self_improvement_key();
768        let count = self.kv_get(&key).await?;
769        Ok(count.and_then(|c| c.parse().ok()).unwrap_or(0))
770    }
771
772    /// Get tasks with high review cycle counts (persistent review loops) for a configurable window.
773    pub async fn get_high_review_cycle_tasks(
774        &self,
775        hours: u32,
776    ) -> anyhow::Result<Vec<HighReviewCycleTask>> {
777        let interval = format!("-{hours} hours");
778        let rows = sqlx::query(
779            "SELECT external_id, agent, review_cycles, title
780         FROM tasks
781         WHERE review_cycles >= 2
782           AND datetime(updated_at) >= datetime('now', ?)
783         ORDER BY review_cycles DESC
784         LIMIT 10",
785        )
786        .bind(&interval)
787        .fetch_all(&self.pool)
788        .await?;
789
790        rows.iter()
791            .map(|row| {
792                Ok(HighReviewCycleTask {
793                    // external_id and agent are nullable columns — NULL is a valid value
794                    external_id: row.try_get("external_id")?,
795                    agent: row.try_get("agent")?,
796                    review_cycles: row.try_get("review_cycles")?,
797                    title: row.try_get("title")?,
798                })
799            })
800            .collect()
801    }
802
803    /// Get tasks with high review cycle counts (persistent review loops) from the last 7 days.
804    /// Test-only wrapper.
805    #[cfg(test)]
806    pub async fn get_high_review_cycle_tasks_7d(&self) -> anyhow::Result<Vec<HighReviewCycleTask>> {
807        let rows = sqlx::query(
808            "SELECT external_id, agent, review_cycles, title
809         FROM tasks
810         WHERE review_cycles >= 2
811           AND datetime(updated_at) >= datetime('now', '-7 days')
812         ORDER BY review_cycles DESC
813         LIMIT 10",
814        )
815        .fetch_all(&self.pool)
816        .await?;
817
818        rows.iter()
819            .map(|row| {
820                Ok(HighReviewCycleTask {
821                    // external_id and agent are nullable columns — NULL is a valid value
822                    external_id: row.try_get("external_id")?,
823                    agent: row.try_get("agent")?,
824                    review_cycles: row.try_get("review_cycles")?,
825                    title: row.try_get("title")?,
826                })
827            })
828            .collect()
829    }
830
831    /// Increment the self-improvement issue counter for the current week.
832    /// Uses a single atomic SQL statement to avoid TOCTOU race conditions.
833    /// Test-only helper.
834    #[cfg(test)]
835    pub async fn increment_self_improvement_counter(&self) -> anyhow::Result<()> {
836        let key = self_improvement_key();
837        sqlx::query(
838        "INSERT INTO kv (key, value, updated_at) VALUES (?, '1', strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
839         ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')",
840    )
841    .bind(&key)
842    .execute(&self.pool)
843    .await?;
844        Ok(())
845    }
846}
orch/store/metrics.rs

orch/store/
metrics.rs
