1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
#!/usr/bin/env python3
"""Live task tracking with owner-based handling.
Completed/cancelled tasks are removed from this table; their history lives in
the other memory layers.
"""
import sys
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from db import get_conn
def _now() -> str:
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
def _row(r):
return dict(r) if r else None
_VALID_OWNERS = ("agent", "user")
def _validate_owner(owner: str | None) -> dict | None:
"""Return an error dict if owner is set to an invalid value, else None."""
if owner is not None and owner not in _VALID_OWNERS:
return {"error": f"invalid owner '{owner}'. Must be one of: {', '.join(_VALID_OWNERS)}"}
return None
def add_task(owner: str, content: str, due_at: str | None = None,
notes: str | None = None) -> dict:
if err := _validate_owner(owner):
return err
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO tasks (owner, content, due_at, notes) VALUES (?, ?, ?, ?)",
(owner, content, due_at, notes),
)
return _row(conn.execute("SELECT * FROM tasks WHERE id=?", (cur.lastrowid,)).fetchone())
def list_tasks(owner: str | None = None, due_before: str | None = None) -> list[dict]:
sql = "SELECT * FROM tasks WHERE 1=1"
params = []
if owner:
sql += " AND owner=?"; params.append(owner)
if due_before:
sql += " AND due_at IS NOT NULL AND due_at < ?"; params.append(due_before)
sql += " ORDER BY due_at IS NULL, due_at"
with get_conn() as conn:
return [dict(r) for r in conn.execute(sql, params).fetchall()]
def update_task(task_id: int, content: str | None = None, due_at: str | None = None,
owner: str | None = None, notes: str | None = None) -> dict:
if err := _validate_owner(owner):
return err
sets, params = [], []
for col, val in (("content", content), ("due_at", due_at),
("owner", owner), ("notes", notes)):
if val is not None:
sets.append(f"{col}=?"); params.append(val)
if not sets:
return {"error": "no fields to update"}
sets.append("updated_at=?"); params.append(_now())
params.append(task_id)
with get_conn() as conn:
if not conn.execute("SELECT id FROM tasks WHERE id=?", (task_id,)).fetchone():
return {"error": f"no task with id {task_id}"}
conn.execute(f"UPDATE tasks SET {', '.join(sets)} WHERE id=?", params)
return _row(conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone())
def remove_task(task_id: int) -> dict:
with get_conn() as conn:
row = conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
if not row:
return {"error": f"no task with id {task_id}"}
conn.execute("DELETE FROM tasks WHERE id=?", (task_id,))
return {"deleted": _row(row)}
def record_nudge(task_id: int) -> dict:
with get_conn() as conn:
if not conn.execute("SELECT id FROM tasks WHERE id=?", (task_id,)).fetchone():
return {"error": f"no task with id {task_id}"}
conn.execute("UPDATE tasks SET last_nudged_at=? WHERE id=?", (_now(), task_id))
return _row(conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone())
|
