| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from datetime import datetime |
| from pathlib import Path |
| import sqlite3 |
| from typing import Iterable |
|
|
| from .paths import LOGS_DB, STATE_DB, TUI_LOG |
|
|
|
|
| @dataclass(frozen=True) |
| class ThreadSummary: |
| id: str |
| title: str |
| cwd: str |
| model: str |
| reasoning_effort: str |
| tokens_used: int |
| archived: bool |
| source: str |
| git_branch: str |
| created_ms: int |
| updated_ms: int |
|
|
|
|
| @dataclass(frozen=True) |
| class LogEntry: |
| id: int |
| ts_ms: int |
| level: str |
| target: str |
| thread_id: str |
| message: str |
|
|
|
|
| @dataclass(frozen=True) |
| class DataAccessError: |
| source: str |
| path: Path |
| message: str |
|
|
|
|
| @dataclass(frozen=True) |
| class DataResult: |
| data: object |
| error: DataAccessError | None = None |
|
|
|
|
| @dataclass(frozen=True) |
| class DashboardStats: |
| thread_count: int |
| archived_count: int |
| total_tokens: int |
| recent_hour_count: int |
| latest_thread: ThreadSummary | None |
| error: DataAccessError | None = None |
|
|
|
|
| def _data_error(source: str, path: Path, exc: BaseException) -> DataAccessError: |
| return DataAccessError(source=source, path=path, message=str(exc)) |
|
|
|
|
| def _connect_readonly(path: Path) -> sqlite3.Connection: |
| if not path.exists(): |
| raise FileNotFoundError(path) |
| uri = f"file:{path}?mode=ro" |
| conn = sqlite3.connect(uri, uri=True, timeout=0.05) |
| conn.row_factory = sqlite3.Row |
| return conn |
|
|
|
|
| def _shorten(text: str | None, limit: int = 120) -> str: |
| value = (text or "").strip().replace("\n", " ") |
| if len(value) <= limit: |
| return value |
| return value[: limit - 3].rstrip() + "..." |
|
|
|
|
| def _thread_from_row(row: sqlite3.Row) -> ThreadSummary: |
| title = row["title"] or row["first_user_message"] or "(untitled)" |
| return ThreadSummary( |
| id=row["id"], |
| title=_shorten(title), |
| cwd=row["cwd"] or "", |
| model=row["model"] or row["model_provider"] or "", |
| reasoning_effort=row["reasoning_effort"] or "", |
| tokens_used=int(row["tokens_used"] or 0), |
| archived=bool(row["archived"]), |
| source=row["source"] or "", |
| git_branch=row["git_branch"] or "", |
| created_ms=int(row["created_ms"] or 0), |
| updated_ms=int(row["updated_ms"] or 0), |
| ) |
|
|
|
|
| def list_threads( |
| *, |
| db_path: Path = STATE_DB, |
| query: str = "", |
| include_archived: bool = False, |
| limit: int = 80, |
| ) -> list[ThreadSummary]: |
| return list_threads_result( |
| db_path=db_path, |
| query=query, |
| include_archived=include_archived, |
| limit=limit, |
| ).data |
|
|
|
|
| def list_threads_result( |
| *, |
| db_path: Path = STATE_DB, |
| query: str = "", |
| include_archived: bool = False, |
| limit: int = 80, |
| ) -> DataResult: |
| clauses = [] |
| params: list[object] = [] |
| if not include_archived: |
| clauses.append("archived = 0") |
| if query.strip(): |
| needle = f"%{query.strip()}%" |
| clauses.append( |
| "(title LIKE ? OR first_user_message LIKE ? OR cwd LIKE ? OR git_branch LIKE ?)" |
| ) |
| params.extend([needle, needle, needle, needle]) |
| where = f"WHERE {' AND '.join(clauses)}" if clauses else "" |
| params.append(limit) |
|
|
| sql = f""" |
| SELECT |
| id, |
| title, |
| first_user_message, |
| cwd, |
| model, |
| model_provider, |
| reasoning_effort, |
| tokens_used, |
| archived, |
| source, |
| git_branch, |
| COALESCE(created_at_ms, created_at * 1000) AS created_ms, |
| COALESCE(updated_at_ms, updated_at * 1000) AS updated_ms |
| FROM threads |
| {where} |
| ORDER BY updated_ms DESC, id DESC |
| LIMIT ? |
| """ |
| try: |
| with _connect_readonly(db_path) as conn: |
| return DataResult([_thread_from_row(row) for row in conn.execute(sql, params)]) |
| except FileNotFoundError: |
| return DataResult([]) |
| except (sqlite3.Error, TypeError, ValueError) as exc: |
| return DataResult([], _data_error("state.sqlite", db_path, exc)) |
|
|
|
|
| def get_stats(*, db_path: Path = STATE_DB) -> DashboardStats: |
| threads_result = list_threads_result(db_path=db_path, include_archived=False, limit=1) |
| threads = threads_result.data |
| try: |
| with _connect_readonly(db_path) as conn: |
| row = conn.execute( |
| """ |
| SELECT |
| COUNT(*) AS thread_count, |
| SUM(CASE WHEN archived = 1 THEN 1 ELSE 0 END) AS archived_count, |
| COALESCE(SUM(tokens_used), 0) AS total_tokens, |
| SUM( |
| CASE |
| WHEN COALESCE(updated_at_ms, updated_at * 1000) |
| >= (strftime('%s','now') * 1000 - 3600000) |
| THEN 1 ELSE 0 |
| END |
| ) AS recent_hour_count |
| FROM threads |
| """ |
| ).fetchone() |
| except FileNotFoundError: |
| return DashboardStats(0, 0, 0, 0, None) |
| except sqlite3.Error as exc: |
| return DashboardStats(0, 0, 0, 0, None, _data_error("state.sqlite", db_path, exc)) |
|
|
| return DashboardStats( |
| thread_count=int(row["thread_count"] or 0), |
| archived_count=int(row["archived_count"] or 0), |
| total_tokens=int(row["total_tokens"] or 0), |
| recent_hour_count=int(row["recent_hour_count"] or 0), |
| latest_thread=threads[0] if threads else None, |
| error=threads_result.error, |
| ) |
|
|
|
|
| def list_logs( |
| *, |
| db_path: Path = LOGS_DB, |
| query: str = "", |
| limit: int = 150, |
| ) -> list[LogEntry]: |
| return list_logs_result(db_path=db_path, query=query, limit=limit).data |
|
|
|
|
| def list_logs_result( |
| *, |
| db_path: Path = LOGS_DB, |
| query: str = "", |
| limit: int = 150, |
| ) -> DataResult: |
| clauses = [] |
| params: list[object] = [] |
| if query.strip(): |
| needle = f"%{query.strip()}%" |
| clauses.append( |
| "(target LIKE ? OR module_path LIKE ? OR feedback_log_body LIKE ? OR thread_id LIKE ?)" |
| ) |
| params.extend([needle, needle, needle, needle]) |
| where = f"WHERE {' AND '.join(clauses)}" if clauses else "" |
| params.append(limit) |
| sql = f""" |
| SELECT |
| id, |
| ts, |
| ts_nanos, |
| level, |
| target, |
| thread_id, |
| feedback_log_body |
| FROM logs |
| {where} |
| ORDER BY ts DESC, ts_nanos DESC, id DESC |
| LIMIT ? |
| """ |
| try: |
| with _connect_readonly(db_path) as conn: |
| rows = conn.execute(sql, params).fetchall() |
| except FileNotFoundError: |
| return DataResult([]) |
| except sqlite3.Error as exc: |
| return DataResult([], _data_error("logs_2.sqlite", db_path, exc)) |
| try: |
| return DataResult([ |
| LogEntry( |
| id=int(row["id"]), |
| ts_ms=int(row["ts"] or 0) * 1000 + int((row["ts_nanos"] or 0) / 1_000_000), |
| level=row["level"] or "", |
| target=row["target"] or "", |
| thread_id=row["thread_id"] or "", |
| message=_shorten(row["feedback_log_body"], 240), |
| ) |
| for row in rows |
| ]) |
| except (TypeError, ValueError, sqlite3.Error) as exc: |
| return DataResult([], _data_error("logs_2.sqlite", db_path, exc)) |
|
|
|
|
| def tail_tui_log(path: Path = TUI_LOG, lines: int = 80) -> list[str]: |
| if not path.exists(): |
| return [] |
| try: |
| data = path.read_text(encoding="utf-8", errors="replace").splitlines() |
| except OSError: |
| return [] |
| return data[-lines:] |
|
|
|
|
| def format_time(ms: int) -> str: |
| if not ms: |
| return "-" |
| return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d %H:%M") |
|
|
|
|
| def format_tokens(value: int) -> str: |
| if value >= 1_000_000: |
| return f"{value / 1_000_000:.1f}M" |
| if value >= 1_000: |
| return f"{value / 1_000:.1f}k" |
| return str(value) |
|
|
|
|
| def first_existing(paths: Iterable[Path]) -> Path | None: |
| for path in paths: |
| if path.exists(): |
| return path |
| return None |
|
|
|
|
