self-tasking engine: persistent queue, idle-impulse + follow-up generators, approval mode

fjerkroa_bot/persistence.py

@@ -13,7 +13,7 @@ from contextlib import closing
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-SCHEMA_VERSION = 4
+SCHEMA_VERSION = 5
 
 
 class PersistentStore:
@@ -66,6 +66,12 @@ class PersistentStore:
                     " user TEXT NOT NULL, message_id TEXT, ext TEXT NOT NULL, bytes INTEGER NOT NULL,"
                     " created_at TEXT NOT NULL DEFAULT (datetime('now')))"
                 )
+            if version < 5:
+                conn.execute(
+                    "CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, kind TEXT NOT NULL, channel TEXT NOT NULL,"
+                    " due_at TEXT NOT NULL, payload TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'queued',"
+                    " created_at TEXT NOT NULL DEFAULT (datetime('now')), executed_at TEXT)"
+                )
             if version < SCHEMA_VERSION:
                 conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
         os.chmod(self.db_path, 0o600)  # conversation data (PER-04)
@@ -195,6 +201,39 @@ class PersistentStore:
             )
         return cursor.rowcount
 
+    # --- task queue (SPEC-005, FDB-011) ---
+
+    def task_add(self, kind: str, channel: str, due_at: str, payload: str, state: str = "queued") -> int:
+        with closing(self._connect()) as conn, conn:
+            cursor = conn.execute(
+                "INSERT INTO tasks (kind, channel, due_at, payload, state) VALUES (?, ?, ?, ?, ?)",
+                (kind, channel, due_at, payload, state),
+            )
+        return int(cursor.lastrowid or 0)
+
+    def tasks_due(self) -> List[Dict[str, Any]]:
+        with closing(self._connect()) as conn:
+            rows = conn.execute(
+                "SELECT id, kind, channel, payload FROM tasks WHERE state = 'queued' AND due_at <= datetime('now') ORDER BY due_at"
+            ).fetchall()
+        return [{"id": row[0], "kind": row[1], "channel": row[2], "payload": row[3]} for row in rows]
+
+    def tasks_open(self) -> List[Dict[str, Any]]:
+        with closing(self._connect()) as conn:
+            rows = conn.execute(
+                "SELECT id, kind, channel, due_at, state FROM tasks WHERE state IN ('queued', 'approval') ORDER BY id"
+            ).fetchall()
+        return [{"id": row[0], "kind": row[1], "channel": row[2], "due_at": row[3], "state": row[4]} for row in rows]
+
+    def task_set_state(self, task_id: int, state: str) -> int:
+        with closing(self._connect()) as conn, conn:
+            return conn.execute("UPDATE tasks SET state = ?, executed_at = datetime('now') WHERE id = ?", (state, task_id)).rowcount
+
+    def tasks_pending_of_kind(self, kind: str) -> int:
+        with closing(self._connect()) as conn:
+            row = conn.execute("SELECT COUNT(*) FROM tasks WHERE kind = ? AND state IN ('queued', 'approval')", (kind,)).fetchone()
+        return int(row[0])
+
     # --- image cache index (SPEC-004, FDB-010) ---
 
     def image_add(self, sha256: str, channel: str, user: str, message_id: Optional[str], ext: str, nbytes: int) -> None: