| from __future__ import annotations |
|
|
| import json |
| import os |
| import sqlite3 |
| from datetime import UTC, datetime |
| from pathlib import Path |
| from typing import Iterable |
|
|
| from .models import Decision, Job, JobSnapshot, Phase, Result |
|
|
|
|
| def now_iso() -> str: |
| return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") |
|
|
|
|
| def default_db_path() -> Path: |
| configured = os.environ.get("CLFGUI_DB_PATH") |
| if configured: |
| return Path(configured).expanduser() |
|
|
| xdg_data_home = os.environ.get("XDG_DATA_HOME") |
| base = Path(xdg_data_home).expanduser() if xdg_data_home else Path.home() / ".local" / "share" |
| return base / "clfgui" / "clfgui.sqlite3" |
|
|
|
|
| class SQLiteStatusStore: |
| def __init__(self, path: str | Path | None = None) -> None: |
| self.path = Path(path) if path is not None else default_db_path() |
| if str(self.path) != ":memory:": |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
|
|
| self._conn = sqlite3.connect(str(self.path)) |
| self._conn.row_factory = sqlite3.Row |
| self._conn.execute("pragma foreign_keys = on") |
| if str(self.path) != ":memory:": |
| self._conn.execute("pragma journal_mode = wal") |
| self._init_schema() |
|
|
| def close(self) -> None: |
| self._conn.close() |
|
|
| def _init_schema(self) -> None: |
| self._conn.executescript( |
| """ |
| create table if not exists jobs ( |
| id integer primary key autoincrement, |
| intent text not null, |
| status text not null, |
| created_at text not null, |
| updated_at text not null |
| ); |
| |
| create table if not exists phases ( |
| id integer primary key autoincrement, |
| job_id integer not null references jobs(id) on delete cascade, |
| name text not null, |
| position integer not null, |
| status text not null, |
| summary text not null default '', |
| started_at text, |
| completed_at text, |
| unique(job_id, name) |
| ); |
| |
| create table if not exists decisions ( |
| id integer primary key autoincrement, |
| job_id integer not null references jobs(id) on delete cascade, |
| phase_name text not null, |
| title text not null, |
| context text not null, |
| options_json text not null, |
| status text not null, |
| selected_option text, |
| note text not null default '', |
| created_at text not null, |
| answered_at text |
| ); |
| |
| create table if not exists results ( |
| id integer primary key autoincrement, |
| job_id integer not null unique references jobs(id) on delete cascade, |
| summary text not null, |
| blocked integer not null, |
| decisions_json text not null, |
| next_steps_json text not null, |
| approved integer, |
| created_at text not null, |
| updated_at text not null |
| ); |
| """ |
| ) |
| self._conn.commit() |
|
|
| def create_job(self, intent: str) -> Job: |
| timestamp = now_iso() |
| cursor = self._conn.execute( |
| "insert into jobs (intent, status, created_at, updated_at) values (?, ?, ?, ?)", |
| (intent, "running", timestamp, timestamp), |
| ) |
| self._conn.commit() |
| return self.get_job(cursor.lastrowid) |
|
|
| def list_jobs(self) -> list[Job]: |
| rows = self._conn.execute("select * from jobs order by updated_at desc, id desc").fetchall() |
| return [self._job_from_row(row) for row in rows] |
|
|
| def get_job(self, job_id: int) -> Job: |
| row = self._conn.execute("select * from jobs where id = ?", (job_id,)).fetchone() |
| if row is None: |
| raise KeyError(f"Unknown job id: {job_id}") |
| return self._job_from_row(row) |
|
|
| def update_job_status(self, job_id: int, status: str) -> Job: |
| self._conn.execute( |
| "update jobs set status = ?, updated_at = ? where id = ?", |
| (status, now_iso(), job_id), |
| ) |
| self._conn.commit() |
| return self.get_job(job_id) |
|
|
| def create_phases(self, job_id: int, names: Iterable[str]) -> list[Phase]: |
| for position, name in enumerate(names, start=1): |
| self._conn.execute( |
| """ |
| insert into phases (job_id, name, position, status) |
| values (?, ?, ?, ?) |
| on conflict(job_id, name) do nothing |
| """, |
| (job_id, name, position, "pending"), |
| ) |
| self._touch_job(job_id) |
| self._conn.commit() |
| return self.list_phases(job_id) |
|
|
| def update_phase(self, job_id: int, name: str, status: str, summary: str) -> Phase: |
| existing = self._conn.execute( |
| "select started_at, completed_at from phases where job_id = ? and name = ?", |
| (job_id, name), |
| ).fetchone() |
| if existing is None: |
| raise KeyError(f"Unknown phase {name!r} for job {job_id}") |
|
|
| timestamp = now_iso() |
| started_at = existing["started_at"] |
| completed_at = existing["completed_at"] |
| if status == "running" and started_at is None: |
| started_at = timestamp |
| if status in {"completed", "blocked", "failed"} and completed_at is None: |
| completed_at = timestamp |
|
|
| self._conn.execute( |
| """ |
| update phases |
| set status = ?, summary = ?, started_at = ?, completed_at = ? |
| where job_id = ? and name = ? |
| """, |
| (status, summary, started_at, completed_at, job_id, name), |
| ) |
| self._touch_job(job_id) |
| self._conn.commit() |
| return self.get_phase(job_id, name) |
|
|
| def get_phase(self, job_id: int, name: str) -> Phase: |
| row = self._conn.execute( |
| "select * from phases where job_id = ? and name = ?", |
| (job_id, name), |
| ).fetchone() |
| if row is None: |
| raise KeyError(f"Unknown phase {name!r} for job {job_id}") |
| return self._phase_from_row(row) |
|
|
| def list_phases(self, job_id: int) -> list[Phase]: |
| rows = self._conn.execute( |
| "select * from phases where job_id = ? order by position asc", |
| (job_id,), |
| ).fetchall() |
| return [self._phase_from_row(row) for row in rows] |
|
|
| def create_decision( |
| self, |
| job_id: int, |
| phase_name: str, |
| title: str, |
| context: str, |
| options: list[str], |
| ) -> Decision: |
| if len(options) < 2: |
| raise ValueError("A decision needs at least two options.") |
|
|
| timestamp = now_iso() |
| cursor = self._conn.execute( |
| """ |
| insert into decisions ( |
| job_id, phase_name, title, context, options_json, status, created_at |
| ) values (?, ?, ?, ?, ?, ?, ?) |
| """, |
| (job_id, phase_name, title, context, json.dumps(options), "open", timestamp), |
| ) |
| self._touch_job(job_id) |
| self._conn.commit() |
| return self.get_decision(cursor.lastrowid) |
|
|
| def get_decision(self, decision_id: int) -> Decision: |
| row = self._conn.execute("select * from decisions where id = ?", (decision_id,)).fetchone() |
| if row is None: |
| raise KeyError(f"Unknown decision id: {decision_id}") |
| return self._decision_from_row(row) |
|
|
| def list_open_decisions(self, job_id: int) -> list[Decision]: |
| rows = self._conn.execute( |
| """ |
| select * from decisions |
| where job_id = ? and status = 'open' |
| order by created_at asc, id asc |
| """, |
| (job_id,), |
| ).fetchall() |
| return [self._decision_from_row(row) for row in rows] |
|
|
| def list_decisions(self, job_id: int) -> list[Decision]: |
| rows = self._conn.execute( |
| "select * from decisions where job_id = ? order by created_at asc, id asc", |
| (job_id,), |
| ).fetchall() |
| return [self._decision_from_row(row) for row in rows] |
|
|
| def answer_decision(self, decision_id: int, selected_option: str, note: str = "") -> Decision: |
| decision = self.get_decision(decision_id) |
| if decision.status != "open": |
| raise ValueError("Decision has already been answered.") |
| if selected_option not in decision.options: |
| raise ValueError("Selected option is not valid for this decision.") |
|
|
| self._conn.execute( |
| """ |
| update decisions |
| set status = 'answered', selected_option = ?, note = ?, answered_at = ? |
| where id = ? |
| """, |
| (selected_option, note, now_iso(), decision_id), |
| ) |
| self._touch_job(decision.job_id) |
| self._conn.commit() |
| return self.get_decision(decision_id) |
|
|
| def save_result( |
| self, |
| job_id: int, |
| summary: str, |
| blocked: bool, |
| decisions: list[str], |
| next_steps: list[str], |
| ) -> Result: |
| timestamp = now_iso() |
| self._conn.execute( |
| """ |
| insert into results ( |
| job_id, summary, blocked, decisions_json, next_steps_json, approved, |
| created_at, updated_at |
| ) values (?, ?, ?, ?, ?, null, ?, ?) |
| on conflict(job_id) do update set |
| summary = excluded.summary, |
| blocked = excluded.blocked, |
| decisions_json = excluded.decisions_json, |
| next_steps_json = excluded.next_steps_json, |
| approved = excluded.approved, |
| updated_at = excluded.updated_at |
| """, |
| ( |
| job_id, |
| summary, |
| int(blocked), |
| json.dumps(decisions), |
| json.dumps(next_steps), |
| timestamp, |
| timestamp, |
| ), |
| ) |
| self._touch_job(job_id) |
| self._conn.commit() |
| return self.get_result(job_id) |
|
|
| def get_result(self, job_id: int) -> Result | None: |
| row = self._conn.execute("select * from results where job_id = ?", (job_id,)).fetchone() |
| return self._result_from_row(row) if row else None |
|
|
| def set_result_approval(self, job_id: int, approved: bool) -> Result: |
| result = self.get_result(job_id) |
| if result is None: |
| raise KeyError(f"Job {job_id} has no result to approve.") |
|
|
| self._conn.execute( |
| "update results set approved = ?, updated_at = ? where job_id = ?", |
| (int(approved), now_iso(), job_id), |
| ) |
| self._touch_job(job_id) |
| self._conn.commit() |
| return self.get_result(job_id) |
|
|
| def snapshot(self, job_id: int) -> JobSnapshot: |
| return JobSnapshot( |
| job=self.get_job(job_id), |
| phases=self.list_phases(job_id), |
| decisions=self.list_decisions(job_id), |
| result=self.get_result(job_id), |
| ) |
|
|
| def _touch_job(self, job_id: int) -> None: |
| self._conn.execute("update jobs set updated_at = ? where id = ?", (now_iso(), job_id)) |
|
|
| @staticmethod |
| def _job_from_row(row: sqlite3.Row) -> Job: |
| return Job( |
| id=row["id"], |
| intent=row["intent"], |
| status=row["status"], |
| created_at=row["created_at"], |
| updated_at=row["updated_at"], |
| ) |
|
|
| @staticmethod |
| def _phase_from_row(row: sqlite3.Row) -> Phase: |
| return Phase( |
| id=row["id"], |
| job_id=row["job_id"], |
| name=row["name"], |
| position=row["position"], |
| status=row["status"], |
| summary=row["summary"], |
| started_at=row["started_at"], |
| completed_at=row["completed_at"], |
| ) |
|
|
| @staticmethod |
| def _decision_from_row(row: sqlite3.Row) -> Decision: |
| return Decision( |
| id=row["id"], |
| job_id=row["job_id"], |
| phase_name=row["phase_name"], |
| title=row["title"], |
| context=row["context"], |
| options=json.loads(row["options_json"]), |
| status=row["status"], |
| selected_option=row["selected_option"], |
| note=row["note"], |
| created_at=row["created_at"], |
| answered_at=row["answered_at"], |
| ) |
|
|
| @staticmethod |
| def _result_from_row(row: sqlite3.Row) -> Result: |
| approved = None if row["approved"] is None else bool(row["approved"]) |
| return Result( |
| id=row["id"], |
| job_id=row["job_id"], |
| summary=row["summary"], |
| blocked=bool(row["blocked"]), |
| decisions=json.loads(row["decisions_json"]), |
| next_steps=json.loads(row["next_steps_json"]), |
| approved=approved, |
| created_at=row["created_at"], |
| updated_at=row["updated_at"], |
| ) |
|
|
