kanban_store/store.py
Rama main · réplica de lectura del repositorio original.
"""Kanban — SQLite store.
Single source of truth: ``tasks.db`` next to the repo root (gitignored), or
the path from env ``KANBAN_DB``.
Public API:
Store.list_tasks(status=..., assignee=..., project_id=..., include_archived=False)
Store.get_task(task_id, history_limit=None)
Store.history(task_id, limit=100, before_id=None)
Store.create_task(title, status="backlog", ...)
Store.move_task(task_id, to_status, actor, comment=None)
Store.assign_task(task_id, assignee, actor)
Store.archive_task(task_id, archived=True, actor)
Store.add_comment(task_id, text, actor)
Store.add_link(task_id, type, value)
Store.set_blockers(task_id, blocker_ids)
Store.snapshot() -> dict (for JSON snapshots)
Every mutation that changes something writes one row to ``task_history``.
Calls that would not change anything (same priority, same assignee, move
into the column the task is already in) write nothing — the history is the
event log that automation and webhooks consume, so no-op rows are noise.
"""
from __future__ import annotations
import json
import os
import sqlite3
import threading
DEFAULT_PROJECT_ID = os.environ.get("KANBAN_DEFAULT_PROJECT_ID", "default")
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
# ============================================================================
# Status model
# ============================================================================
# 9 columns in left-to-right UI order.
STATUSES: list[str] = [
"backlog",
"approved",
"analyst",
"in_progress",
"testing",
"uat",
"done",
"blocked",
"cancelled",
]
# Link types accepted by add_link / create_task(links=...).
# ``plan`` — a planning document (e.g. a Claude Code plan-mode file).
LINK_TYPES: tuple[str, ...] = ("memory", "file", "pr", "url", "plan")
# Statuses in which a blocker counts as resolved.
RESOLVED_STATUSES: tuple[str, ...] = ("done", "cancelled")
PRIORITIES: tuple[str, ...] = ("critical", "high", "normal", "low")
SIZES: tuple[str, ...] = ("S", "M", "L", "XL")
# Spellings agents and people actually use for columns → canonical id.
_STATUS_ALIASES: dict[str, str] = {
"todo": "backlog", "бэклог": "backlog",
"согласовано": "approved",
"analysis": "analyst", "analytics": "analyst", "аналитика": "analyst",
"inprogress": "in_progress", "wip": "in_progress", "doing": "in_progress",
"в_работе": "in_progress",
"review": "testing", "qa": "testing", "тестирование": "testing",
"acceptance": "uat", "приёмка": "uat", "приемка": "uat",
"closed": "done", "закрыто": "done",
"заблокировано": "blocked",
"canceled": "cancelled", "отменено": "cancelled",
}
def normalize_status(value: str) -> str:
"""``"In progress"``, ``"in-progress"``, ``"Testing"``, ``"В работе"`` → canonical id.
Raises ``ValueError`` listing the valid ids for anything unrecognised.
"""
raw = str(value if value is not None else "").strip()
key = raw.lower().replace("-", "_").replace(" ", "_")
if key in STATUSES:
return key
if key in _STATUS_ALIASES:
return _STATUS_ALIASES[key]
if key.replace("_", "") in _STATUS_ALIASES:
return _STATUS_ALIASES[key.replace("_", "")]
raise ValueError(f"unknown status: {value!r}; valid: {', '.join(STATUSES)}")
def normalize_priority(value: str) -> str:
key = str(value if value is not None else "").strip().lower()
if key == "medium":
return "normal"
if key in PRIORITIES:
return key
raise ValueError(f"unknown priority: {value!r}; valid: {', '.join(PRIORITIES)}")
def normalize_size(value: str) -> str:
key = str(value if value is not None else "").strip().upper()
if key in SIZES:
return key
raise ValueError(f"unknown size: {value!r}; valid: {', '.join(SIZES)}")
def status_meta() -> list[dict[str, str]]:
"""Column metadata for the UI (label + cssClass)."""
return [
{"id": "backlog", "title": "Backlog", "owner": "user"},
{"id": "approved", "title": "Approved", "owner": "agent"},
{"id": "analyst", "title": "Analyst", "owner": "agent"},
{"id": "in_progress", "title": "In progress", "owner": "agent"},
{"id": "testing", "title": "Testing", "owner": "agent"},
{"id": "uat", "title": "UAT", "owner": "user"},
{"id": "done", "title": "Done", "owner": "user"},
{"id": "blocked", "title": "Blocked", "owner": "any"},
{"id": "cancelled", "title": "Cancelled", "owner": "user"},
]
# ============================================================================
# Models
# ============================================================================
@dataclass
class TaskHistory:
id: int
task_id: str
ts: str
actor: str
action: str
from_status: str | None
to_status: str | None
comment: str | None
@dataclass
class Task:
id: str
title: str
status: str
priority: str
size: str
assignee: str | None
description: str
acceptance: str
external_blocker: str | None
created_at: str
moved_at: str
column_order: int
project_id: str = DEFAULT_PROJECT_ID
archived_at: str | None = None
links: list[dict[str, str]] = field(default_factory=list)
history: list[TaskHistory] = field(default_factory=list)
# Total number of history rows; ``history`` may hold only the newest ones.
history_total: int = 0
blockers: list[str] = field(default_factory=list)
def to_public(self) -> dict[str, Any]:
d = asdict(self)
d["history"] = [asdict(h) for h in self.history]
return d
@dataclass
class Project:
id: str
name: str
color: str
icon: str
sort_order: int
archived: bool
created_at: str
path: str | None = None
task_counts: dict[str, int] = field(default_factory=dict)
total_tasks: int = 0
def to_public(self) -> dict[str, Any]:
return asdict(self)
# ============================================================================
# Store
# ============================================================================
def _now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
class StatusConflict(RuntimeError):
"""The task is not in the status the caller expected (see move_task)."""
# Actors that are software, not a person. Moving a card out of Blocked resets
# run_command's max_runs budget only when a person does it — otherwise a
# "retry blocked cards" rule would defeat the limit it is meant to respect.
MACHINE_ACTORS = {"automation", "claude", "codex", "plan-import", "maintenance", "inbox"}
def is_machine_actor(actor: str | None) -> bool:
a = (actor or "").strip().lower()
return a in MACHINE_ACTORS or a.startswith(("agent:", "claude", "codex"))
def _default_db_path() -> Path:
env = os.environ.get("KANBAN_DB", "").strip()
if env:
return Path(os.path.expanduser(env))
return Path(__file__).resolve().parent.parent / "tasks.db"
def normalize_links(links: Iterable[Any] | None) -> list[tuple[str, str]]:
"""Validate ``[{"type": ..., "value": ...}]`` and return (type, value) pairs.
``link_type`` is accepted as an alias of ``type`` (the MCP tool uses that
name). Raises ``ValueError`` with a readable message instead of letting a
malformed payload surface as ``KeyError``.
"""
out: list[tuple[str, str]] = []
for i, ln in enumerate(links or []):
if not isinstance(ln, dict):
raise ValueError(f"links[{i}] must be an object with 'type' and 'value'")
type_ = ln.get("type", ln.get("link_type"))
value = ln.get("value")
if not isinstance(type_, str) or type_ not in LINK_TYPES:
raise ValueError(
f"links[{i}].type must be one of {list(LINK_TYPES)}, got {type_!r}"
)
if not isinstance(value, str) or not value.strip():
raise ValueError(f"links[{i}].value must be a non-empty string")
out.append((type_, value.strip()))
return out
class Store:
"""Thread-safe wrapper around SQLite. One instance per process."""
_lock = threading.RLock()
def __init__(self, db_path: str | Path | None = None, *, create: bool = True):
"""``create=False`` refuses to start a new, empty database: used by
the MCP server, where a wrong ``KANBAN_DB`` would otherwise give the
agent an empty board of its own instead of an error."""
self.db_path = Path(db_path) if db_path is not None else _default_db_path()
if not create and not self.db_path.exists():
raise FileNotFoundError(
f"kanban database not found: {self.db_path} — check KANBAN_DB "
"(the web server creates the database on its first start)"
)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._conn = sqlite3.connect(
str(self.db_path),
check_same_thread=False,
isolation_level=None, # autocommit; explicit transactions use BEGIN
# The UI server, every MCP process and helper scripts share one
# file — wait for a concurrent writer instead of failing fast.
timeout=15.0,
)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA foreign_keys = ON")
self._conn.execute("PRAGMA journal_mode = WAL")
self._migrate()
# ------------------------------------------------------------------
# Migration
# ------------------------------------------------------------------
def _migrate(self) -> None:
schema = (Path(__file__).parent / "schema.sql").read_text(encoding="utf-8")
with self._lock:
self._conn.executescript(schema)
self._migrate_v2()
self._migrate_v3()
self._migrate_v4()
self._migrate_v5()
def _schema_version(self) -> int:
row = self._conn.execute(
"SELECT value FROM meta WHERE key='schema_version'"
).fetchone()
return int(row["value"]) if row else 1
def _migrate_v5(self) -> None:
"""v4 → v5: tasks.archived_at (rule_firings comes from schema.sql)."""
cols = {r[1] for r in self._conn.execute("PRAGMA table_info(tasks)").fetchall()}
if "archived_at" not in cols:
self._conn.execute("ALTER TABLE tasks ADD COLUMN archived_at TEXT")
if self._schema_version() < 5:
self._conn.execute("UPDATE meta SET value='5' WHERE key='schema_version'")
def _migrate_v4(self) -> None:
"""v3 → v4: project_sources table (created via schema.sql,
this method only bumps the version)."""
if self._schema_version() < 4:
self._conn.execute("UPDATE meta SET value='4' WHERE key='schema_version'")
def _migrate_v3(self) -> None:
"""v2 → v3: projects.path TEXT (Claude Code project directory)."""
cols = {r[1] for r in self._conn.execute("PRAGMA table_info(projects)").fetchall()}
if "path" not in cols:
self._conn.execute("ALTER TABLE projects ADD COLUMN path TEXT")
if self._schema_version() < 3:
self._conn.execute("UPDATE meta SET value='3' WHERE key='schema_version'")
def _migrate_v2(self) -> None:
"""v1 → v2: adds tasks.project_id for existing databases and
creates a default project (id/name are configurable via env).
Idempotent: checks PRAGMA table_info before running ALTER.
"""
version = self._schema_version()
cols = {r[1] for r in self._conn.execute("PRAGMA table_info(tasks)").fetchall()}
if "project_id" not in cols:
# old pre-v2 database — add the column with a default
default_id = os.environ.get("KANBAN_DEFAULT_PROJECT_ID", "default")
self._conn.execute(
f"ALTER TABLE tasks ADD COLUMN project_id TEXT NOT NULL DEFAULT '{default_id}'"
)
# The project_id index is always created (idempotent). For a fresh
# database the column appeared from CREATE TABLE in schema.sql; for
# older databases it is created after the ALTER above.
self._conn.execute(
"CREATE INDEX IF NOT EXISTS idx_tasks_project_status "
"ON tasks(project_id, status, column_order)"
)
# The default project is created ONLY when the database has no
# projects at all (fresh install). Existing databases keep their
# own projects without an extra "default" being added on top.
row = self._conn.execute("SELECT COUNT(*) AS n FROM projects").fetchone()
if row["n"] == 0:
default_id = os.environ.get("KANBAN_DEFAULT_PROJECT_ID", "default")
default_name = os.environ.get("KANBAN_DEFAULT_PROJECT_NAME", "Default")
default_color = os.environ.get("KANBAN_DEFAULT_PROJECT_COLOR", "#F10D30")
default_icon = os.environ.get(
"KANBAN_DEFAULT_PROJECT_ICON", default_name[:1].upper()
)
self._conn.execute(
"INSERT INTO projects (id, name, color, icon, sort_order, archived, created_at) "
"VALUES (?, ?, ?, ?, 0, 0, ?)",
(default_id, default_name, default_color, default_icon, _now()),
)
if version < 2:
self._conn.execute(
"UPDATE meta SET value='2' WHERE key='schema_version'"
)
# ------------------------------------------------------------------
# ID generation
# ------------------------------------------------------------------
def _next_id(self) -> str:
with self._lock:
row = self._conn.execute(
"SELECT value FROM meta WHERE key='next_id'"
).fetchone()
n = int(row["value"]) if row else 1
self._conn.execute(
"UPDATE meta SET value=? WHERE key='next_id'", (str(n + 1),)
)
return f"T-{n:03d}"
# ------------------------------------------------------------------
# Reads
# ------------------------------------------------------------------
def list_tasks(
self,
status: str | Iterable[str] | None = None,
assignee: str | None = None,
project_id: str | None = None,
*,
include_archived: bool = False,
) -> list[Task]:
"""List of tasks with filters, sorted by (status, column_order).
``project_id=None`` means "all projects". The board UI always
passes a concrete project_id. Archived tasks are skipped unless
``include_archived=True``.
"""
sql = "SELECT * FROM tasks WHERE 1=1"
params: list[Any] = []
if status is not None:
if isinstance(status, str):
statuses = [status]
else:
statuses = list(status)
placeholders = ",".join("?" * len(statuses))
sql += f" AND status IN ({placeholders})"
params.extend(statuses)
if assignee is not None:
sql += " AND assignee = ?"
params.append(assignee)
if project_id is not None:
sql += " AND project_id = ?"
params.append(project_id)
if not include_archived:
sql += " AND archived_at IS NULL"
sql += " ORDER BY status, column_order, id"
with self._lock:
rows = self._conn.execute(sql, params).fetchall()
tasks = [self._row_to_task(r) for r in rows]
self._attach_links_and_blockers(tasks)
return tasks
def get_task(self, task_id: str, *, history_limit: int | None = None) -> Task | None:
"""Full card. ``history_limit`` keeps only the newest N history rows
(``history_total`` still reports the full count)."""
with self._lock:
row = self._conn.execute(
"SELECT * FROM tasks WHERE id = ?", (task_id,)
).fetchone()
if not row:
return None
t = self._row_to_task(row)
self._attach_links_and_blockers([t])
t.history_total = self._conn.execute(
"SELECT COUNT(*) AS n FROM task_history WHERE task_id=?", (task_id,)
).fetchone()["n"]
if history_limit is None:
h_rows = self._conn.execute(
"SELECT * FROM task_history WHERE task_id=? ORDER BY id ASC",
(task_id,),
).fetchall()
else:
h_rows = self._conn.execute(
"SELECT * FROM task_history WHERE task_id=? ORDER BY id DESC LIMIT ?",
(task_id, max(0, int(history_limit))),
).fetchall()[::-1]
t.history = [self._row_to_history(r) for r in h_rows]
return t
def history(
self, task_id: str, *, limit: int = 100, before_id: int | None = None
) -> tuple[list[TaskHistory], int]:
"""One page of a task's history, oldest→newest within the page.
Pages go backwards in time: pass the smallest ``id`` of the current
page as ``before_id`` to get the previous page. Returns
``(items, total)``. Raises ``KeyError`` for an unknown task.
"""
with self._lock:
if not self._task_exists(task_id):
raise KeyError(task_id)
total = self._conn.execute(
"SELECT COUNT(*) AS n FROM task_history WHERE task_id=?", (task_id,)
).fetchone()["n"]
sql = "SELECT * FROM task_history WHERE task_id=?"
params: list[Any] = [task_id]
if before_id is not None:
sql += " AND id < ?"
params.append(int(before_id))
sql += " ORDER BY id DESC LIMIT ?"
params.append(max(1, min(int(limit), 1000)))
rows = self._conn.execute(sql, params).fetchall()[::-1]
return [self._row_to_history(r) for r in rows], total
def history_since(self, after_id: int, *, limit: int = 500) -> list[TaskHistory]:
"""History rows with ``id > after_id`` in insertion order (event feed)."""
with self._lock:
rows = self._conn.execute(
"SELECT * FROM task_history WHERE id > ? ORDER BY id ASC LIMIT ?",
(int(after_id), int(limit)),
).fetchall()
return [self._row_to_history(r) for r in rows]
def max_history_id(self) -> int:
with self._lock:
row = self._conn.execute(
"SELECT COALESCE(MAX(id), 0) AS m FROM task_history"
).fetchone()
return int(row["m"])
def board(self) -> dict[str, list[Task]]:
"""Group tasks by status, in column order."""
result: dict[str, list[Task]] = {s: [] for s in STATUSES}
for t in self.list_tasks():
result.setdefault(t.status, []).append(t)
return result
# ------------------------------------------------------------------
# Writes
# ------------------------------------------------------------------
def create_task(
self,
title: str,
*,
status: str = "backlog",
priority: str = "normal",
size: str = "M",
description: str = "",
acceptance: str = "",
assignee: str | None = None,
external_blocker: str | None = None,
actor: str = "user",
links: list[dict[str, str]] | None = None,
task_id: str | None = None,
project_id: str = DEFAULT_PROJECT_ID,
) -> Task:
status = normalize_status(status)
priority = normalize_priority(priority)
size = normalize_size(size)
if not title or not title.strip():
raise ValueError("title must not be empty")
link_pairs = normalize_links(links)
ts = _now()
with self._lock:
if self.get_project(project_id) is None:
known = ", ".join(p.id for p in self.list_projects(include_archived=True))
raise ValueError(f"unknown project: {project_id} (known: {known})")
self._conn.execute("BEGIN IMMEDIATE")
try:
tid = task_id or self._next_id()
# column_order — last in the column + 1 (per project)
row = self._conn.execute(
"SELECT COALESCE(MAX(column_order), -1) AS m FROM tasks "
"WHERE status=? AND project_id=?",
(status, project_id),
).fetchone()
col_order = (row["m"] + 1) if row else 0
self._conn.execute(
"""
INSERT INTO tasks (id, title, status, priority, size, assignee,
description, acceptance, external_blocker,
created_at, moved_at, column_order, project_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
tid,
title.strip(),
status,
priority,
size,
assignee or None,
description,
acceptance,
external_blocker or None,
ts,
ts,
col_order,
project_id,
),
)
for type_, value in link_pairs:
self._conn.execute(
"INSERT OR IGNORE INTO task_links (task_id, type, value) VALUES (?, ?, ?)",
(tid, type_, value),
)
self._insert_history(tid, ts, actor, "create", None, status, None)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
task = self.get_task(tid)
assert task is not None
return task
def move_task(
self,
task_id: str,
to_status: str,
*,
actor: str = "user",
comment: str | None = None,
column_order: int | None = None,
expected_from: str | None = None,
) -> Task:
"""Move a task to ``to_status``.
Moving into the column the task is already in is not a status
change: only the order within the column changes (when
``column_order`` is given), ``moved_at`` is left alone and the
optional comment is recorded as a plain comment.
``expected_from`` makes the move conditional: if the task is no
longer in that status (a human or another agent moved it), a
``StatusConflict`` is raised and nothing changes.
"""
to_status = normalize_status(to_status)
if expected_from is not None:
expected_from = normalize_status(expected_from)
ts = _now()
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
row = self._conn.execute(
"SELECT status, project_id FROM tasks WHERE id=?", (task_id,)
).fetchone()
if not row:
raise KeyError(task_id)
from_status = row["status"]
project_id = row["project_id"]
if expected_from is not None and from_status != expected_from:
raise StatusConflict(
f"task {task_id} is in '{from_status}', not '{expected_from}' — "
"someone else moved it; re-read the card before acting"
)
if from_status == to_status:
if column_order is not None:
self._place_in_column(task_id, project_id, to_status, column_order)
if comment:
self._insert_history(task_id, ts, actor, "comment", None, None, comment)
else:
self._conn.execute(
"UPDATE tasks SET status=?, moved_at=? WHERE id=?",
(to_status, ts, task_id),
)
if column_order is None:
# append to the end of the project's column
r2 = self._conn.execute(
"SELECT COALESCE(MAX(column_order), -1) AS m FROM tasks "
"WHERE status=? AND project_id=? AND id != ?",
(to_status, project_id, task_id),
).fetchone()
self._conn.execute(
"UPDATE tasks SET column_order=? WHERE id=?",
((r2["m"] + 1) if r2 else 0, task_id),
)
else:
self._place_in_column(task_id, project_id, to_status, column_order)
self._insert_history(
task_id, ts, actor, "move", from_status, to_status, comment
)
if from_status == "blocked" and not is_machine_actor(actor):
# A person unblocking the card is the "try again"
# signal: fresh max_runs budget for run_command rules.
self._conn.execute(
"DELETE FROM command_runs WHERE task_id=?", (task_id,)
)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
task = self.get_task(task_id, history_limit=0)
assert task is not None
return task
def _place_in_column(
self, task_id: str, project_id: str, status: str, index: int
) -> None:
"""Put ``task_id`` at ``index`` of its column and renumber the column
0..n-1 (drag-drop sends the target index, not a free order value)."""
ids = [
r["id"]
for r in self._conn.execute(
"SELECT id FROM tasks WHERE project_id=? AND status=? AND id != ? "
"AND archived_at IS NULL ORDER BY column_order, id",
(project_id, status, task_id),
).fetchall()
]
index = max(0, min(int(index), len(ids)))
ids.insert(index, task_id)
for i, tid in enumerate(ids):
self._conn.execute("UPDATE tasks SET column_order=? WHERE id=?", (i, tid))
def assign_task(self, task_id: str, assignee: str | None, *, actor: str) -> Task:
"""Set (or clear with ``None``/``""``) the assignee. No-op if unchanged."""
assignee = (assignee or "").strip() or None
ts = _now()
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
row = self._conn.execute(
"SELECT assignee FROM tasks WHERE id=?", (task_id,)
).fetchone()
if not row:
raise KeyError(task_id)
if row["assignee"] != assignee:
self._conn.execute(
"UPDATE tasks SET assignee=? WHERE id=?", (assignee, task_id)
)
self._insert_history(
task_id, ts, actor, "assign", None, None,
f"assignee → {assignee if assignee else '—'}",
)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
t = self.get_task(task_id, history_limit=0)
assert t is not None
return t
def pull_task(self, task_id: str, assignee: str = "claude") -> Task:
"""Atomic: assignee IS NULL → assignee, status approved → analyst.
Used by Claude/an agent for a safe "claim the task" operation.
"""
ts = _now()
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
row = self._conn.execute(
"SELECT assignee, status, project_id FROM tasks WHERE id=?", (task_id,)
).fetchone()
if not row:
raise KeyError(task_id)
if row["assignee"] is not None and row["assignee"] != assignee:
raise RuntimeError(
f"task {task_id} already assigned to {row['assignee']}"
)
if row["status"] != "approved":
raise RuntimeError(
f"task {task_id} is in '{row['status']}', not 'approved'"
)
# move to analyst and claim (per project)
r2 = self._conn.execute(
"SELECT COALESCE(MAX(column_order), -1) AS m FROM tasks "
"WHERE status='analyst' AND project_id=?",
(row["project_id"],),
).fetchone()
col_order = (r2["m"] + 1) if r2 else 0
self._conn.execute(
"""UPDATE tasks SET status='analyst', assignee=?, moved_at=?, column_order=?
WHERE id=?""",
(assignee, ts, col_order, task_id),
)
self._insert_history(task_id, ts, assignee, "move", "approved", "analyst", "pulled")
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
t = self.get_task(task_id, history_limit=0)
assert t is not None
return t
def add_comment(self, task_id: str, text: str, *, actor: str) -> None:
if not text or not text.strip():
raise ValueError("comment text must not be empty")
ts = _now()
with self._lock:
if not self._task_exists(task_id):
raise KeyError(task_id)
self._insert_history(task_id, ts, actor, "comment", None, None, text)
def add_link(self, task_id: str, type_: str, value: str) -> bool:
"""Attach a link. Returns False if the same link already exists."""
((type_, value),) = normalize_links([{"type": type_, "value": value}])
with self._lock:
if not self._task_exists(task_id):
raise KeyError(task_id)
cur = self._conn.execute(
"INSERT OR IGNORE INTO task_links (task_id, type, value) VALUES (?, ?, ?)",
(task_id, type_, value),
)
return cur.rowcount > 0
def set_blockers(
self, task_id: str, blocker_ids: list[str], *, actor: str = "user"
) -> list[str]:
"""Replace the task's internal blockers. Returns the stored list.
Rejects unknown ids, self-references and cycles with ``ValueError``
(``KeyError`` for an unknown ``task_id``).
"""
wanted: list[str] = []
for b in blocker_ids:
b = (b or "").strip()
if b and b not in wanted:
wanted.append(b)
ts = _now()
with self._lock:
if not self._task_exists(task_id):
raise KeyError(task_id)
if task_id in wanted:
raise ValueError(f"{task_id} cannot block itself")
unknown = [b for b in wanted if not self._task_exists(b)]
if unknown:
raise ValueError(f"unknown blocker task(s): {', '.join(unknown)}")
cycle = self._find_cycle(task_id, wanted)
if cycle:
raise ValueError("blockers would create a cycle: " + " → ".join(cycle))
current = sorted(
r["blocker_id"]
for r in self._conn.execute(
"SELECT blocker_id FROM task_blockers WHERE task_id=?", (task_id,)
).fetchall()
)
if current == sorted(wanted):
return wanted
self._conn.execute("BEGIN IMMEDIATE")
try:
self._conn.execute(
"DELETE FROM task_blockers WHERE task_id=?", (task_id,)
)
for b in wanted:
self._conn.execute(
"INSERT INTO task_blockers (task_id, blocker_id) VALUES (?, ?)",
(task_id, b),
)
self._insert_history(
task_id, ts, actor, "update", None, None,
"blockers: " + (", ".join(wanted) if wanted else "—"),
)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
return wanted
def _find_cycle(self, task_id: str, new_blockers: list[str]) -> list[str] | None:
"""Path back to ``task_id`` through existing blocker edges, if any."""
stack: list[tuple[str, list[str]]] = [(b, [task_id, b]) for b in new_blockers]
seen: set[str] = set()
while stack:
node, path = stack.pop()
if node == task_id:
return path
if node in seen:
continue
seen.add(node)
for r in self._conn.execute(
"SELECT blocker_id FROM task_blockers WHERE task_id=?", (node,)
).fetchall():
stack.append((r["blocker_id"], path + [r["blocker_id"]]))
return None
# Fields whose old → new values are short enough to show in history.
_SHOW_VALUES = ("priority", "size")
def update_fields(
self,
task_id: str,
*,
actor: str,
title: str | None = None,
priority: str | None = None,
size: str | None = None,
description: str | None = None,
acceptance: str | None = None,
external_blocker: str | None = None,
) -> Task:
"""Update the given fields (``None`` = leave alone).
Only fields whose value actually changes are written, and the
history row lists just those (``priority: normal → high; title``).
``external_blocker=""`` clears the external blocker. A call that
changes nothing writes nothing.
"""
requested = {
"title": title.strip() if isinstance(title, str) else None,
"priority": priority,
"size": size,
"description": description,
"acceptance": acceptance,
"external_blocker": external_blocker,
}
if requested["title"] == "":
raise ValueError("title must not be empty")
if requested["priority"] is not None:
requested["priority"] = normalize_priority(requested["priority"])
if requested["size"] is not None:
requested["size"] = normalize_size(requested["size"])
ts = _now()
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
row = self._conn.execute(
"SELECT * FROM tasks WHERE id=?", (task_id,)
).fetchone()
if not row:
raise KeyError(task_id)
sets: list[str] = []
params: list[Any] = []
notes: list[str] = []
for col, val in requested.items():
if val is None:
continue
if col == "external_blocker":
val = val.strip() or None
if row[col] == val:
continue
sets.append(f"{col} = ?")
params.append(val)
if col in self._SHOW_VALUES:
notes.append(f"{col}: {row[col]} → {val}")
else:
notes.append(col)
if sets:
params.append(task_id)
self._conn.execute(
f"UPDATE tasks SET {', '.join(sets)} WHERE id = ?", params
)
self._insert_history(
task_id, ts, actor, "update", None, None, "; ".join(notes)
)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
t = self.get_task(task_id, history_limit=0)
assert t is not None
return t
def archive_task(
self, task_id: str, *, archived: bool = True, actor: str = "user",
comment: str | None = None,
) -> Task:
"""Hide a task from the board without changing its status.
Archiving keeps ``done`` tasks ``done`` (instead of relabelling them
``cancelled``); ``archived=False`` brings the task back. No-op if the
task is already in the requested state.
"""
ts = _now()
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
row = self._conn.execute(
"SELECT archived_at FROM tasks WHERE id=?", (task_id,)
).fetchone()
if not row:
raise KeyError(task_id)
is_archived = row["archived_at"] is not None
if is_archived != archived:
self._conn.execute(
"UPDATE tasks SET archived_at=? WHERE id=?",
(ts if archived else None, task_id),
)
self._insert_history(
task_id, ts, actor, "archive" if archived else "unarchive",
None, None, comment,
)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
t = self.get_task(task_id, history_limit=0)
assert t is not None
return t
def reorder(self, task_id: str, new_order: int) -> None:
"""Change the order within the current column."""
with self._lock:
self._conn.execute(
"UPDATE tasks SET column_order=? WHERE id=?", (new_order, task_id)
)
# ------------------------------------------------------------------
# Rule firings (idempotency for polling automation rules)
# ------------------------------------------------------------------
def rule_last_fired(self, rule_key: str, task_id: str, anchor: str) -> str | None:
"""When ``rule_key`` last acted on ``task_id`` in episode ``anchor``."""
with self._lock:
row = self._conn.execute(
"SELECT fired_at FROM rule_firings WHERE rule_key=? AND task_id=? AND anchor=?",
(rule_key, task_id, anchor),
).fetchone()
return row["fired_at"] if row else None
def record_rule_firing(self, rule_key: str, task_id: str, anchor: str) -> None:
with self._lock:
self._conn.execute(
"""INSERT INTO rule_firings (rule_key, task_id, anchor, fired_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(rule_key, task_id, anchor) DO UPDATE SET fired_at=excluded.fired_at""",
(rule_key, task_id, anchor, _now()),
)
def command_runs(self, rule_key: str, task_id: str) -> int:
"""How many times run_command ``rule_key`` was started for ``task_id``."""
with self._lock:
row = self._conn.execute(
"SELECT runs FROM command_runs WHERE rule_key=? AND task_id=?",
(rule_key, task_id),
).fetchone()
return int(row["runs"]) if row else 0
def bump_command_runs(self, rule_key: str, task_id: str, delta: int = 1) -> None:
with self._lock:
self._conn.execute(
"""INSERT INTO command_runs (rule_key, task_id, runs, last_started_at)
VALUES (?, ?, MAX(0, ?), ?)
ON CONFLICT(rule_key, task_id) DO UPDATE
SET runs = MAX(0, runs + ?),
last_started_at = CASE WHEN ? > 0 THEN excluded.last_started_at
ELSE last_started_at END""",
(rule_key, task_id, delta, _now(), delta, delta),
)
def set_command_rc(self, rule_key: str, task_id: str, rc: int | None) -> None:
with self._lock:
self._conn.execute(
"UPDATE command_runs SET last_rc=? WHERE rule_key=? AND task_id=?",
(rc, rule_key, task_id),
)
def reset_command_runs(self, task_id: str) -> None:
with self._lock:
self._conn.execute("DELETE FROM command_runs WHERE task_id=?", (task_id,))
def save_command_job(
self, rule_key: str, task_id: str, *, state: str, rule_name: str,
ctx: dict[str, Any], pid: int | None = None, proc_start: str | None = None,
) -> None:
now = _now()
with self._lock:
self._conn.execute(
"""INSERT INTO command_jobs
(rule_key, task_id, state, rule_name, ctx, pid, proc_start,
queued_at, started_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(rule_key, task_id) DO UPDATE SET
state=excluded.state, rule_name=excluded.rule_name, ctx=excluded.ctx,
pid=excluded.pid, proc_start=excluded.proc_start,
started_at=CASE WHEN excluded.state='running' THEN excluded.started_at
ELSE command_jobs.started_at END""",
(rule_key, task_id, state, rule_name,
json.dumps(ctx, ensure_ascii=False), pid, proc_start, now,
now if state == "running" else None),
)
def delete_command_job(self, rule_key: str, task_id: str) -> None:
with self._lock:
self._conn.execute(
"DELETE FROM command_jobs WHERE rule_key=? AND task_id=?", (rule_key, task_id)
)
def list_command_jobs(self) -> list[dict[str, Any]]:
with self._lock:
# running first, then the queue in insertion order (rowid: the
# upsert keeps it when a queued job starts)
rows = self._conn.execute(
"SELECT * FROM command_jobs "
"ORDER BY CASE state WHEN 'running' THEN 0 ELSE 1 END, rowid"
).fetchall()
return [{**dict(r), "ctx": json.loads(r["ctx"])} for r in rows]
# ------------------------------------------------------------------
# Projects
# ------------------------------------------------------------------
def list_projects(self, *, include_archived: bool = False) -> list[Project]:
"""All projects with task_counts aggregated by status (archived
tasks are not counted)."""
sql = "SELECT * FROM projects"
if not include_archived:
sql += " WHERE archived = 0"
sql += " ORDER BY sort_order, name"
with self._lock:
rows = self._conn.execute(sql).fetchall()
counts_rows = self._conn.execute(
"SELECT project_id, status, COUNT(*) AS n "
"FROM tasks WHERE archived_at IS NULL GROUP BY project_id, status"
).fetchall()
counts: dict[str, dict[str, int]] = {}
totals: dict[str, int] = {}
for r in counts_rows:
counts.setdefault(r["project_id"], {})[r["status"]] = r["n"]
totals[r["project_id"]] = totals.get(r["project_id"], 0) + r["n"]
result = []
for r in rows:
p = self._row_to_project(r)
p.task_counts = counts.get(p.id, {})
p.total_tasks = totals.get(p.id, 0)
result.append(p)
return result
def get_project(self, project_id: str) -> Project | None:
with self._lock:
row = self._conn.execute(
"SELECT * FROM projects WHERE id=?", (project_id,)
).fetchone()
if not row:
return None
return self._row_to_project(row)
def create_project(
self,
project_id: str,
name: str,
*,
color: str = "#F10D30",
icon: str = "",
sort_order: int | None = None,
path: str | None = None,
) -> Project:
ts = _now()
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
if sort_order is None:
r = self._conn.execute(
"SELECT COALESCE(MAX(sort_order), -1) AS m FROM projects"
).fetchone()
sort_order = (r["m"] + 1) if r else 0
self._conn.execute(
"""INSERT INTO projects
(id, name, color, icon, sort_order, archived, path, created_at)
VALUES (?, ?, ?, ?, ?, 0, ?, ?)""",
(project_id, name, color, icon or name[:1].upper(), sort_order, path, ts),
)
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
p = self.get_project(project_id)
assert p is not None
return p
def update_project(
self,
project_id: str,
*,
name: str | None = None,
color: str | None = None,
icon: str | None = None,
sort_order: int | None = None,
path: str | None = None,
) -> Project:
sets: list[str] = []
params: list[Any] = []
# path is forwarded as-is (None means "leave alone", "" means "clear").
for col, val in (
("name", name), ("color", color), ("icon", icon),
("sort_order", sort_order), ("path", path),
):
if val is not None:
sets.append(f"{col} = ?")
# an empty string for path becomes NULL in the database
params.append(None if (col == "path" and val == "") else val)
if not sets:
p = self.get_project(project_id)
if p is None:
raise KeyError(project_id)
return p
params.append(project_id)
with self._lock:
row = self._conn.execute(
"SELECT 1 FROM projects WHERE id=?", (project_id,)
).fetchone()
if not row:
raise KeyError(project_id)
self._conn.execute(
f"UPDATE projects SET {', '.join(sets)} WHERE id=?", params
)
p = self.get_project(project_id)
assert p is not None
return p
# ------------------------------------------------------------------
# Project sources (one source per project)
# ------------------------------------------------------------------
def set_project_source(
self, project_id: str, type_: str, config: dict[str, Any]
) -> None:
ts = _now()
with self._lock:
self._conn.execute(
"""INSERT INTO project_sources (project_id, type, config, created_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(project_id) DO UPDATE
SET type=excluded.type, config=excluded.config""",
(project_id, type_, json.dumps(config, ensure_ascii=False), ts),
)
def get_project_source(self, project_id: str) -> dict[str, Any] | None:
with self._lock:
row = self._conn.execute(
"SELECT * FROM project_sources WHERE project_id=?", (project_id,)
).fetchone()
if not row:
return None
return {
"project_id": row["project_id"],
"type": row["type"],
"config": json.loads(row["config"]),
"last_sync_at": row["last_sync_at"],
"created_at": row["created_at"],
}
def update_source_sync_time(self, project_id: str) -> None:
with self._lock:
self._conn.execute(
"UPDATE project_sources SET last_sync_at=? WHERE project_id=?",
(_now(), project_id),
)
def delete_project_source(self, project_id: str) -> None:
with self._lock:
self._conn.execute(
"DELETE FROM project_sources WHERE project_id=?", (project_id,)
)
def archive_project(self, project_id: str, archived: bool = True) -> Project:
with self._lock:
row = self._conn.execute(
"SELECT 1 FROM projects WHERE id=?", (project_id,)
).fetchone()
if not row:
raise KeyError(project_id)
self._conn.execute(
"UPDATE projects SET archived=? WHERE id=?",
(1 if archived else 0, project_id),
)
p = self.get_project(project_id)
assert p is not None
return p
def _row_to_project(self, row: sqlite3.Row) -> Project:
return Project(
id=row["id"],
name=row["name"],
color=row["color"],
icon=row["icon"],
sort_order=row["sort_order"],
archived=bool(row["archived"]),
created_at=row["created_at"],
path=row["path"] if "path" in row.keys() else None,
)
# ------------------------------------------------------------------
# Snapshot
# ------------------------------------------------------------------
def snapshot(self) -> dict[str, Any]:
"""Dump the whole board as a plain dict for JSON persistence."""
tasks = self.list_tasks(include_archived=True)
projects = self.list_projects(include_archived=True)
return {
"exported_at": _now(),
"schema_version": 5,
"projects": [p.to_public() for p in projects],
"tasks": [t.to_public() for t in tasks],
}
def save_snapshot(self, dest_dir: str | Path | None = None) -> Path:
if dest_dir is None:
dest_dir = Path(__file__).resolve().parent.parent / "snapshots"
dest_dir = Path(dest_dir)
dest_dir.mkdir(parents=True, exist_ok=True)
date_part = datetime.now(timezone.utc).strftime("%Y-%m-%d")
fp = dest_dir / f"{date_part}.json"
fp.write_text(
json.dumps(self.snapshot(), ensure_ascii=False, indent=2),
encoding="utf-8",
)
return fp
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _task_exists(self, task_id: str) -> bool:
return self._conn.execute(
"SELECT 1 FROM tasks WHERE id=?", (task_id,)
).fetchone() is not None
def _insert_history(
self,
task_id: str,
ts: str,
actor: str,
action: str,
from_status: str | None,
to_status: str | None,
comment: str | None,
) -> None:
self._conn.execute(
"""INSERT INTO task_history
(task_id, ts, actor, action, from_status, to_status, comment)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(task_id, ts, actor or "user", action, from_status, to_status, comment),
)
def _attach_links_and_blockers(self, tasks: list[Task]) -> None:
"""Load links and blockers for many tasks with two queries."""
if not tasks:
return
by_id = {t.id: t for t in tasks}
ids = list(by_id)
for chunk_start in range(0, len(ids), 500):
chunk = ids[chunk_start:chunk_start + 500]
ph = ",".join("?" * len(chunk))
for r in self._conn.execute(
f"SELECT task_id, type, value FROM task_links WHERE task_id IN ({ph}) "
"ORDER BY type, value",
chunk,
).fetchall():
by_id[r["task_id"]].links.append({"type": r["type"], "value": r["value"]})
for r in self._conn.execute(
f"SELECT task_id, blocker_id FROM task_blockers WHERE task_id IN ({ph}) "
"ORDER BY blocker_id",
chunk,
).fetchall():
by_id[r["task_id"]].blockers.append(r["blocker_id"])
@staticmethod
def _row_to_history(r: sqlite3.Row) -> TaskHistory:
return TaskHistory(
id=r["id"],
task_id=r["task_id"],
ts=r["ts"],
actor=r["actor"],
action=r["action"],
from_status=r["from_status"],
to_status=r["to_status"],
comment=r["comment"],
)
@staticmethod
def _row_to_task(row: sqlite3.Row) -> Task:
keys = row.keys()
return Task(
id=row["id"],
title=row["title"],
status=row["status"],
priority=row["priority"],
size=row["size"],
assignee=row["assignee"],
description=row["description"],
acceptance=row["acceptance"],
external_blocker=row["external_blocker"],
created_at=row["created_at"],
moved_at=row["moved_at"],
column_order=row["column_order"],
project_id=row["project_id"] if "project_id" in keys else DEFAULT_PROJECT_ID,
archived_at=row["archived_at"] if "archived_at" in keys else None,
)
def close(self) -> None:
with self._lock:
self._conn.close()