orch/store/
metrics.rs

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

orch/store/
metrics.rs
