orch/store/
metrics.rs

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

orch/store/
metrics.rs
