Load recent analyses. Args: skill_id: True ``skill_id`` (e.g. ``weather__imp_a1b2c3d4``). ``skill_judgments.skill_id`` now stores the true skill_id, so filtering uses exact match. If None, return pure-execution analyses (no judgmen
(
self,
skill_id: Optional[str] = None,
limit: int = 50,
)
| 772 | |
| 773 | @_db_retry() |
| 774 | def load_analyses( |
| 775 | self, |
| 776 | skill_id: Optional[str] = None, |
| 777 | limit: int = 50, |
| 778 | ) -> List[ExecutionAnalysis]: |
| 779 | """Load recent analyses. |
| 780 | |
| 781 | Args: |
| 782 | skill_id: True ``skill_id`` (e.g. ``weather__imp_a1b2c3d4``). |
| 783 | ``skill_judgments.skill_id`` now stores the true skill_id, |
| 784 | so filtering uses exact match. |
| 785 | If None, return pure-execution analyses (no judgments). |
| 786 | """ |
| 787 | with self._reader() as conn: |
| 788 | if skill_id is not None: |
| 789 | rows = conn.execute( |
| 790 | "SELECT ea.* FROM execution_analyses ea " |
| 791 | "JOIN skill_judgments sj ON ea.id = sj.analysis_id " |
| 792 | "WHERE sj.skill_id = ? " |
| 793 | "ORDER BY ea.timestamp DESC LIMIT ?", |
| 794 | (skill_id, limit), |
| 795 | ).fetchall() |
| 796 | else: |
| 797 | rows = conn.execute( |
| 798 | "SELECT ea.* FROM execution_analyses ea " |
| 799 | "LEFT JOIN skill_judgments sj ON ea.id = sj.analysis_id " |
| 800 | "WHERE sj.id IS NULL " |
| 801 | "ORDER BY ea.timestamp DESC LIMIT ?", |
| 802 | (limit,), |
| 803 | ).fetchall() |
| 804 | return [self._to_analysis(conn, r) for r in reversed(rows)] |
| 805 | |
| 806 | @_db_retry() |
| 807 | def load_analyses_for_task( |
no test coverage detected