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

orch/store/
metrics.rs
