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    /// Count recent rate limit events per agent within the given window (in hours).
475    ///
476    /// Returns a map of `agent -> count` for agents that have rate_limit or
477    /// out_of_credits events in the `rate_limits` table within the window.
478    /// Used by the pre-emptive health check to detect degraded agents.
479    pub async fn recent_rate_limit_counts(
480        &self,
481        window_hours: u32,
482    ) -> anyhow::Result<std::collections::HashMap<String, i64>> {
483        // SQLite datetime modifiers don't support parameterised intervals, so
484        // we build the interval string and embed it directly.  The value comes
485        // from a config integer, not user input, so this is safe.
486        let interval = format!("-{window_hours} hours");
487        let rows = sqlx::query(
488            "SELECT agent, COUNT(*) as cnt
489             FROM rate_limits
490             WHERE occurred_at >= datetime('now', ?)
491             GROUP BY agent",
492        )
493        .bind(&interval)
494        .fetch_all(&self.pool)
495        .await?;
496
497        let mut map = std::collections::HashMap::new();
498        for row in &rows {
499            let agent: String = row.try_get("agent")?;
500            let cnt: i64 = row.try_get("cnt")?;
501            map.insert(agent, cnt);
502        }
503        Ok(map)
504    }
505
506    /// Get slow tasks (top 10 longest running) for a configurable time window.
507    pub async fn get_slow_tasks(&self, hours: u32) -> anyhow::Result<Vec<SlowTaskInfo>> {
508        let interval = format!("-{hours} hours");
509        let rows = sqlx::query(
510            "SELECT task_id, agent, complexity, duration_seconds
511         FROM task_metrics
512         WHERE datetime(completed_at) >= datetime('now', ?) AND outcome = 'success'
513         ORDER BY duration_seconds DESC
514         LIMIT 10",
515        )
516        .bind(&interval)
517        .fetch_all(&self.pool)
518        .await?;
519
520        rows.iter()
521            .map(|row| {
522                Ok(SlowTaskInfo {
523                    task_id: row.try_get("task_id")?,
524                    agent: row.try_get("agent")?,
525                    // complexity is a nullable column — NULL is a valid value
526                    complexity: row.try_get("complexity")?,
527                    duration_seconds: row.try_get("duration_seconds")?,
528                })
529            })
530            .collect()
531    }
532
533    /// Get slow tasks (top 10 longest running) from the last 7 days.
534    /// Test-only wrapper.
535    #[cfg(test)]
536    pub async fn get_slow_tasks_7d(&self) -> anyhow::Result<Vec<SlowTaskInfo>> {
537        let rows = sqlx::query(
538            "SELECT task_id, agent, complexity, duration_seconds
539         FROM task_metrics
540         WHERE datetime(completed_at) >= datetime('now', '-7 days') AND outcome = 'success'
541         ORDER BY duration_seconds DESC
542         LIMIT 10",
543        )
544        .fetch_all(&self.pool)
545        .await?;
546
547        rows.iter()
548            .map(|row| {
549                Ok(SlowTaskInfo {
550                    task_id: row.try_get("task_id")?,
551                    agent: row.try_get("agent")?,
552                    // complexity is a nullable column — NULL is a valid value
553                    complexity: row.try_get("complexity")?,
554                    duration_seconds: row.try_get("duration_seconds")?,
555                })
556            })
557            .collect()
558    }
559
560    /// Get error type distribution for a configurable time window.
561    pub async fn get_error_distribution(&self, hours: u32) -> anyhow::Result<Vec<ErrorStat>> {
562        let interval = format!("-{hours} hours");
563        let rows = sqlx::query(
564            "SELECT error_type, COUNT(*) as count
565         FROM task_metrics
566         WHERE datetime(completed_at) >= datetime('now', ?)
567           AND outcome != 'success'
568           AND error_type IS NOT NULL
569         GROUP BY error_type
570         ORDER BY count DESC",
571        )
572        .bind(&interval)
573        .fetch_all(&self.pool)
574        .await?;
575
576        rows.iter()
577            .map(|row| {
578                Ok(ErrorStat {
579                    // error_type is a nullable column — NULL is a valid value
580                    error_type: row.try_get("error_type")?,
581                    count: row.try_get("count")?,
582                })
583            })
584            .collect()
585    }
586
587    /// Get cost summary over multiple time windows (24h, 7d, 30d).
588    ///
589    /// Uses separate static queries per window because SQLite's `datetime()`
590    /// modifier argument cannot be parameterised via `?` bind.
591    pub async fn get_cost_summary(&self) -> anyhow::Result<CostSummary> {
592        // Each entry: (static SQL with the interval baked in, label)
593        let windows: &[(&str, &str)] = &[
594            (
595                "SELECT COALESCE(SUM(input_tokens), 0) as input_tokens,
596                    COALESCE(SUM(output_tokens), 0) as output_tokens,
597                    COALESCE(SUM(input_cost_usd), 0.0) as input_cost_usd,
598                    COALESCE(SUM(output_cost_usd), 0.0) as output_cost_usd,
599                    COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
600                    COUNT(*) as task_count
601             FROM task_metrics
602             WHERE datetime(completed_at) >= datetime('now', '-24 hours')",
603                "24h",
604            ),
605            (
606                "SELECT COALESCE(SUM(input_tokens), 0) as input_tokens,
607                    COALESCE(SUM(output_tokens), 0) as output_tokens,
608                    COALESCE(SUM(input_cost_usd), 0.0) as input_cost_usd,
609                    COALESCE(SUM(output_cost_usd), 0.0) as output_cost_usd,
610                    COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
611                    COUNT(*) as task_count
612             FROM task_metrics
613             WHERE datetime(completed_at) >= datetime('now', '-7 days')",
614                "7d",
615            ),
616            (
617                "SELECT COALESCE(SUM(input_tokens), 0) as input_tokens,
618                    COALESCE(SUM(output_tokens), 0) as output_tokens,
619                    COALESCE(SUM(input_cost_usd), 0.0) as input_cost_usd,
620                    COALESCE(SUM(output_cost_usd), 0.0) as output_cost_usd,
621                    COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
622                    COUNT(*) as task_count
623             FROM task_metrics
624             WHERE datetime(completed_at) >= datetime('now', '-30 days')",
625                "30d",
626            ),
627        ];
628
629        let mut periods = Vec::new();
630        for (query, label) in windows {
631            let row = sqlx::query(query).fetch_one(&self.pool).await?;
632            periods.push(CostPeriod {
633                label: label.to_string(),
634                input_tokens: row.try_get("input_tokens")?,
635                output_tokens: row.try_get("output_tokens")?,
636                input_cost_usd: row.try_get("input_cost_usd")?,
637                output_cost_usd: row.try_get("output_cost_usd")?,
638                total_cost_usd: row.try_get("total_cost_usd")?,
639                task_count: row.try_get("task_count")?,
640            });
641        }
642
643        Ok(CostSummary { periods })
644    }
645
646    /// Get cost breakdown by agent.
647    pub async fn get_cost_by_agent(&self) -> anyhow::Result<Vec<CostByGroup>> {
648        let rows = sqlx::query(
649            "SELECT agent as name,
650                COALESCE(SUM(input_tokens), 0) as input_tokens,
651                COALESCE(SUM(output_tokens), 0) as output_tokens,
652                COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
653                COUNT(*) as task_count
654         FROM task_metrics
655         GROUP BY agent
656         ORDER BY SUM(total_cost_usd) DESC",
657        )
658        .fetch_all(&self.pool)
659        .await?;
660
661        rows.iter()
662            .map(|row| {
663                Ok(CostByGroup {
664                    name: row.try_get("name")?,
665                    input_tokens: row.try_get("input_tokens")?,
666                    output_tokens: row.try_get("output_tokens")?,
667                    total_cost_usd: row.try_get("total_cost_usd")?,
668                    task_count: row.try_get("task_count")?,
669                })
670            })
671            .collect()
672    }
673
674    /// Get cost breakdown by model.
675    pub async fn get_cost_by_model(&self) -> anyhow::Result<Vec<CostByGroup>> {
676        let rows = sqlx::query(
677            "SELECT COALESCE(model, 'unknown') as name,
678                COALESCE(SUM(input_tokens), 0) as input_tokens,
679                COALESCE(SUM(output_tokens), 0) as output_tokens,
680                COALESCE(SUM(total_cost_usd), 0.0) as total_cost_usd,
681                COUNT(*) as task_count
682         FROM task_metrics
683         GROUP BY model
684         ORDER BY SUM(total_cost_usd) DESC",
685        )
686        .fetch_all(&self.pool)
687        .await?;
688
689        rows.iter()
690            .map(|row| {
691                Ok(CostByGroup {
692                    name: row.try_get("name")?,
693                    input_tokens: row.try_get("input_tokens")?,
694                    output_tokens: row.try_get("output_tokens")?,
695                    total_cost_usd: row.try_get("total_cost_usd")?,
696                    task_count: row.try_get("task_count")?,
697                })
698            })
699            .collect()
700    }
701
702    /// Get the duration_seconds from the most recent metric row for a task.
703    /// Returns `None` when no metric has been recorded yet.
704    pub async fn latest_task_metric_duration(&self, task_id: &str) -> Option<f64> {
705        sqlx::query_scalar(
706            "SELECT duration_seconds FROM task_metrics WHERE task_id = ? ORDER BY completed_at DESC LIMIT 1",
707        )
708        .bind(task_id)
709        .fetch_optional(&self.pool)
710        .await
711        .ok()
712        .flatten()
713    }
714
715    /// Get count of self-improvement issues created this week.
716    /// Test-only helper that reads the KV counter.
717    #[cfg(test)]
718    pub async fn count_self_improvement_issues_7d(&self) -> anyhow::Result<i64> {
719        let key = self_improvement_key();
720        let count = self.kv_get(&key).await?;
721        Ok(count.and_then(|c| c.parse().ok()).unwrap_or(0))
722    }
723
724    /// Get tasks with high review cycle counts (persistent review loops) for a configurable window.
725    pub async fn get_high_review_cycle_tasks(
726        &self,
727        hours: u32,
728    ) -> anyhow::Result<Vec<HighReviewCycleTask>> {
729        let interval = format!("-{hours} hours");
730        let rows = sqlx::query(
731            "SELECT external_id, agent, review_cycles, title
732         FROM tasks
733         WHERE review_cycles >= 2
734           AND datetime(updated_at) >= datetime('now', ?)
735         ORDER BY review_cycles DESC
736         LIMIT 10",
737        )
738        .bind(&interval)
739        .fetch_all(&self.pool)
740        .await?;
741
742        rows.iter()
743            .map(|row| {
744                Ok(HighReviewCycleTask {
745                    // external_id and agent are nullable columns — NULL is a valid value
746                    external_id: row.try_get("external_id")?,
747                    agent: row.try_get("agent")?,
748                    review_cycles: row.try_get("review_cycles")?,
749                    title: row.try_get("title")?,
750                })
751            })
752            .collect()
753    }
754
755    /// Get tasks with high review cycle counts (persistent review loops) from the last 7 days.
756    /// Test-only wrapper.
757    #[cfg(test)]
758    pub async fn get_high_review_cycle_tasks_7d(&self) -> anyhow::Result<Vec<HighReviewCycleTask>> {
759        let rows = sqlx::query(
760            "SELECT external_id, agent, review_cycles, title
761         FROM tasks
762         WHERE review_cycles >= 2
763           AND datetime(updated_at) >= datetime('now', '-7 days')
764         ORDER BY review_cycles DESC
765         LIMIT 10",
766        )
767        .fetch_all(&self.pool)
768        .await?;
769
770        rows.iter()
771            .map(|row| {
772                Ok(HighReviewCycleTask {
773                    // external_id and agent are nullable columns — NULL is a valid value
774                    external_id: row.try_get("external_id")?,
775                    agent: row.try_get("agent")?,
776                    review_cycles: row.try_get("review_cycles")?,
777                    title: row.try_get("title")?,
778                })
779            })
780            .collect()
781    }
782
783    /// Increment the self-improvement issue counter for the current week.
784    /// Uses a single atomic SQL statement to avoid TOCTOU race conditions.
785    /// Test-only helper.
786    #[cfg(test)]
787    pub async fn increment_self_improvement_counter(&self) -> anyhow::Result<()> {
788        let key = self_improvement_key();
789        sqlx::query(
790        "INSERT INTO kv (key, value, updated_at) VALUES (?, '1', strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
791         ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')",
792    )
793    .bind(&key)
794    .execute(&self.pool)
795    .await?;
796        Ok(())
797    }
798}
orch/store/metrics.rs

orch/store/
metrics.rs
