stackchain/agent-todo-tracker
1
0
Fork 0
Code Issues 10 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(box): ADHD-friendly time-boxing mode (issue #16)

Browse Source 
- TimeBox model with explicit recoverable BoxState state machine
- Store layer: time_boxes schema, CRUD, active-box recovery
- BoxController: start/pause/resume/interrupt/complete/expire with elapsed math
- CLI  group (start/status/pause/resume/interrupt/complete/abandon/recover/list/focus)
- Live ADHD-friendly focus REPL with single-key commands, calm UX, no-shame quit
- Interruptions logged with timestamp + reason for pattern recognition
- Non-interactive helpers + auto-resume on entry
- 65 tests green, strict vertical RED-GREEN TDD
This commit is contained in:
Hermes Agent 2026-08-22 21:49:35 +00:00
parent 565d660bfe
commit 7a03de305a
12 changed files with 1576 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
README.md
View File

@ -30,6 +30,46 @@ agent-todo close 1
agent-todo list --json
```
## Time-boxing (issue #16)
Calm, ADHD-friendly focus sessions with explicit state, interruptions you can
resume from, and a live focus loop.
```bash
# Start a 25-minute focus box (default)
agent-todo box start
# Custom duration (0.5–240 min)
agent-todo box start 10
# Link a box to a TODO
agent-todo box start 15 --todo 3
# Enter the live focus REPL (auto-starts or auto-resumes)
agent-todo box focus
# Non-interactive helpers
agent-todo box status
agent-todo box pause
agent-todo box resume
agent-todo box interrupt "meeting popped up"
agent-todo box complete --reason "made progress"
agent-todo box abandon --reason "tired"
agent-todo box recover # after a restart / crash
agent-todo box list
agent-todo box list --all
```
Inside `box focus`, single keys drive the loop: `p` pause · `r` resume ·
`i <why>` interrupt · `c` complete · `q` quit · `h` help · Enter refresh.
There's no judgment for quitting — your time is saved.
### Interruptions are recoverable
Pause, interrupt, or even close the terminal — `box recover` or `box focus`
resumes from where you left off. Interruptions are logged with a timestamp and
a reason so you can notice patterns later, without being shamed.
## Storage
Default: `.agent_todos.db` in current directory.

124
agent_todos/box.py Normal file
View File

@ -0,0 +1,124 @@
"""BoxController: drives state transitions and elapsed-time bookkeeping.
Pure logic, no I/O. Takes any object that quacks like a Store (get/update on
time_boxes) so it stays testable and decoupled from SQLite.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any, Optional, Protocol
from .timebox import BoxState, TimeBox
class TimeBoxStore(Protocol):
def get_timebox(self, box_id: int) -> Optional[ TimeBox]: ...
def update_timebox(self, box_id: int, **kwargs: Any) -> Optional[TimeBox]: ...
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _now_epoch() -> float:
return datetime.now(timezone.utc).timestamp()
class BoxController:
"""State-machine driver for a single TimeBox.
Every public method validates the transition via TimeBox.can(), applies
elapsed-time accounting, persists via the store, and returns the updated box.
"""
def __init__(self, store: TimeBoxStore) -> None:
self.store = store
def _accrue_and_freeze(self, box: TimeBox) -> None:
"""Fold live running time into accumulated_seconds and clear the clock."""
if box.state == BoxState.running and box.last_started_at is not None:
try:
started = datetime.fromisoformat(box.last_started_at).timestamp()
box.accumulated_seconds += max(0.0, _now_epoch() - started)
except (ValueError, TypeError):
pass
box.last_started_at = None
def _persist(self, box: TimeBox, **fields: Any) -> TimeBox:
if box.id is None:
raise ValueError("box has no id — persist via Store.create_timebox first")
updated = self.store.update_timebox(box.id, **fields)
if updated is None:
raise ValueError(f"box {box.id} disappeared during update")
return updated
def start(self, box: TimeBox) -> TimeBox:
if box.is_terminal():
raise ValueError(f"Cannot start terminal box in state {box.state.value}")
if box.state != BoxState.open:
raise ValueError(f"Cannot start box in state {box.state.value}; use resume()")
box.last_started_at = _now_iso()
return self._persist(box, state=BoxState.running, last_started_at=box.last_started_at)
def pause(self, box: TimeBox) -> TimeBox:
if box.is_terminal():
raise ValueError(f"Cannot pause terminal box in state {box.state.value}")
if box.state != BoxState.running:
raise ValueError(f"Cannot pause box in state {box.state.value}")
self._accrue_and_freeze(box)
return self._persist(box, state=BoxState.paused, accumulated_seconds=box.accumulated_seconds)
def resume(self, box: TimeBox) -> TimeBox:
if box.is_terminal():
raise ValueError(f"Cannot resume terminal box in state {box.state.value}")
if box.state not in (BoxState.paused, BoxState.interrupted):
raise ValueError(f"Cannot resume box in state {box.state.value}")
box.last_started_at = _now_iso()
return self._persist(box, state=BoxState.running, last_started_at=box.last_started_at)
def interrupt(self, box: TimeBox, reason: str) -> TimeBox:
if box.is_terminal():
raise ValueError(f"Cannot interrupt terminal box in state {box.state.value}")
self._accrue_and_freeze(box)
interruptions = list(box.interruptions)
interruptions.append({"at": _now_iso(), "reason": reason})
return self._persist(
box,
state=BoxState.interrupted,
accumulated_seconds=box.accumulated_seconds,
interruptions=interruptions,
reason=reason,
)
def complete(self, box: TimeBox) -> TimeBox:
if box.is_terminal():
raise ValueError(f"Cannot complete terminal box in state {box.state.value}")
self._accrue_and_freeze(box)
return self._persist(
box,
state=BoxState.completed,
accumulated_seconds=box.accumulated_seconds,
completed_at=_now_iso(),
)
def abandon(self, box: TimeBox, reason: str = "") -> TimeBox:
if box.is_terminal():
raise ValueError(f"Cannot abandon terminal box in state {box.state.value}")
self._accrue_and_freeze(box)
return self._persist(
box,
state=BoxState.abandoned,
accumulated_seconds=box.accumulated_seconds,
reason=reason,
)
def expire(self, box: TimeBox) -> TimeBox:
if box.state != BoxState.running:
raise ValueError(f"Cannot expire box in state {box.state.value}")
self._accrue_and_freeze(box)
return self._persist(
box,
state=BoxState.expired,
accumulated_seconds=box.accumulated_seconds,
completed_at=_now_iso(),
)

3
agent_todos/cli.py
View File

@ -5,12 +5,15 @@ import click
from .models import Priority, Status, Todo
from .store import Store
from . import cli_box
@click.group()
def main() -> None:
pass
main.add_command(cli_box.box_group)
@main.command()
@click.option("--path", default=".agent_todos.db", help="SQLite DB path.")

223
agent_todos/cli_box.py Normal file
View File

@ -0,0 +1,223 @@
"""CLI for the ADHD-friendly time-boxing workflow (issue #16).
Calm, explicit, accessible: every command reports the resulting state in plain
language, interruptions are recoverable, and no one is shamed for switching.
"""
from __future__ import annotations
import json
import math
from typing import Optional
import click
from .box import BoxController
from . import cli_repl
from .store import Store
from .timebox import BoxState
def _mmss(seconds) -> str:
seconds = max(0, int(float(seconds)))
return f"{seconds // 60:02d}:{seconds % 60:02d}"
def _calm_state_phrase(box) -> str:
"""Plain-language description of box state for the user."""
if box.state == BoxState.open:
return "fresh — press start when you're ready"
if box.state == BoxState.running:
return f"focusing — {_mmss(box.remaining_seconds())} left of {_mmss(box.duration_seconds)}"
if box.state == BoxState.paused:
return f"paused with {_mmss(box.remaining_seconds())} left — you can resume anytime"
if box.state == BoxState.interrupted:
return f"interrupted — {_mmss(box.remaining_seconds())} left; resume when the distraction passes"
if box.state == BoxState.completed:
return "completed — nice work"
if box.state == BoxState.abandoned:
return "abandoned — no judgment, come back when you want"
if box.state == BoxState.expired:
return "timer ran out"
return box.state.value
def _format_box(box, title: Optional[str] = None) -> str:
head = f"Box #{box.id}"
if title:
head += f" on \"{title}\""
head += f" — {_calm_state_phrase(box)}"
parts = [head]
if box.interruptions:
parts.append(f" interruptions: {len(box.interruptions)}")
for i in box.interruptions:
parts.append(f" - [{i.get('at', '?')[:19]}] {i.get('reason', '(no reason)')}")
if box.reason and box.state in (BoxState.abandoned, BoxState.completed):
parts.append(f" note: {box.reason}")
return "\n".join(parts)
@click.group(name="box")
def box_group() -> None:
"""ADHD-friendly time-boxing (issue #16). Calm focus sessions, recoverable interruptions."""
pass
def _active_box(store: Store):
"""Return the most-recent active box, or raise a friendly ClickException."""
box = store.recover_timebox()
if box is None:
raise click.ClickException(
"no active time-box. run `box start [minutes]` to begin one."
)
return box
def _run(store: Store, fn):
"""Run a controller action and surface ValueError as a friendly ClickException."""
try:
return fn()
except ValueError as e:
raise click.ClickException(str(e))
@box_group.command()
@click.argument("minutes", type=float, required=False, default=25)
@click.option("--todo", "todo_id", type=int, default=None, help="Link this box to an existing TODO id.")
@click.option("--path", default=".agent_todos.db")
def start(minutes: float, todo_id: Optional[int], path: str) -> None:
"""Start a fresh focus box (default 25m)."""
if not (0.5 < minutes <= 240):
raise click.ClickException("minutes must be between 0.5 and 240")
store = Store(db_path=path)
duration = int(math.ceil(minutes * 60))
box = store.create_timebox(
__import__("agent_todos.timebox", fromlist=["TimeBox"]).TimeBox(
id=None, todo_id=todo_id, duration_seconds=duration
)
)
box = _run(store, lambda: BoxController(store).start(box))
click.echo(f"started box #{box.id} — {_mmss(box.duration_seconds)} on the clock")
click.echo(_calm_state_phrase(box))
@box_group.command()
@click.option("--path", default=".agent_todos.db")
def status(path: str) -> None:
"""Show the active box and its countdown."""
store = Store(db_path=path)
box = _active_box(store)
click.echo(_format_box(box))
@box_group.command()
@click.option("--path", default=".agent_todos.db")
def pause(path: str) -> None:
"""Pause the active box. Time stops; nothing is lost."""
store = Store(db_path=path)
box = _active_box(store)
box = _run(store, lambda: BoxController(store).pause(box))
click.echo(f"paused box #{box.id} — {_mmss(box.remaining_seconds())} saved")
@box_group.command()
@click.option("--path", default=".agent_todos.db")
def resume(path: str) -> None:
"""Resume a paused or interrupted box."""
store = Store(db_path=path)
box = _active_box(store)
box = _run(store, lambda: BoxController(store).resume(box))
click.echo(f"resumed box #{box.id} — {_mmss(box.remaining_seconds())} to go")
@box_group.command()
@click.argument("reason")
@click.option("--path", default=".agent_todos.db")
def interrupt(reason: str, path: str) -> None:
"""Interrupt the active box with a reason (recoverable)."""
store = Store(db_path=path)
box = _active_box(store)
box = _run(store, lambda: BoxController(store).interrupt(box, reason=reason))
click.echo(f"interrupted box #{box.id} — saved {_mmss(box.remaining_seconds())} for later")
click.echo(f" reason: {reason}")
@box_group.command()
@click.option("--reason", default="", help="Optional note about how it went.")
@click.option("--path", default=".agent_todos.db")
def complete(reason: str, path: str) -> None:
"""Mark the active box complete."""
store = Store(db_path=path)
box = _active_box(store)
box = _run(store, lambda: BoxController(store).complete(box))
elapsed = box.elapsed_seconds()
click.echo(f"completed box #{box.id} after {_mmss(elapsed)} — well done")
if reason:
click.echo(f" note: {reason}")
@box_group.command()
@click.option("--reason", default="", help="Optional note about why.")
@click.option("--path", default=".agent_todos.db")
def abandon(reason: str, path: str) -> None:
"""Abandon the active box. No judgment."""
store = Store(db_path=path)
box = _active_box(store)
box = BoxController(store).abandon(box, reason=reason or "")
click.echo(f"abandoned box #{box.id}. it's okay — come back when you want.")
@box_group.command()
@click.option("--path", default=".agent_todos.db")
def recover(path: str) -> None:
"""Recover a box from a previous session (after restart/crash)."""
store = Store(db_path=path)
box = store.recover_timebox()
if box is None:
click.echo("nothing to recover — start a new box with `box start`")
return
click.echo(f"recovered box #{box.id}")
click.echo(_format_box(box))
@box_group.command(name="list")
@click.option("--all", "show_all", is_flag=True, help="Include completed/abandoned boxes too.")
@click.option("--path", default=".agent_todos.db")
def list_boxes(show_all: bool, path: str) -> None:
"""List time-boxes."""
store = Store(db_path=path)
boxes = store.list_timeboxes(active_only=not show_all)
if not boxes:
click.echo("no boxes yet — run `box start` to create one")
return
for b in boxes:
click.echo(f"#{b.id} [{b.state.value}] {_mmss(b.duration_seconds)} box "
f"— {_mmss(b.elapsed_seconds())} used, {_mmss(b.remaining_seconds())} left")
@box_group.command()
@click.option("--path", default=".agent_todos.db")
def focus(path: str) -> None:
"""Enter the live focus loop for the active box (or start a fresh one).
A calm, single-key REPL: p pause · r resume · i <why> interrupt ·
c complete · q quit · h help. Auto-resumes a paused box; auto-starts a
fresh one if none exists.
"""
import sys
store = Store(db_path=path)
box = store.recover_timebox()
if box is None:
box = store.create_timebox(
__import__("agent_todos.timebox", fromlist=["TimeBox"]).TimeBox(
id=None, todo_id=None, duration_seconds=1500
)
)
def _input(timeout):
import select
ready, _, _ = select.select([sys.stdin], [], [], timeout)
if ready:
return sys.stdin.readline().rstrip("\n")
return None
cli_repl.run_repl(store, box, wait_input=_input, display=click.echo, tick_interval=1.0)

182
agent_todos/cli_repl.py Normal file
View File

@ -0,0 +1,182 @@
"""ADHD-friendly live focus REPL (issue #16).
Single-key, calm, interruptible. Designed to be testable without a real TTY:
every interaction is driven by injected `wait_input` and `display` callables.
"""
from __future__ import annotations
import time
from typing import Callable, Optional
from .box import BoxController
from .timebox import BoxState
DisplayFn = Callable[[str], None]
InputFn = Callable[[float], Optional[str]]
def _mmss(seconds) -> str:
seconds = max(0, int(float(seconds)))
return f"{seconds // 60:02d}:{seconds % 60:02d}"
HELP = (
"commands (single key, Enter to repeat):\n"
" p - pause the timer (nothing lost)\n"
" r - resume where you left off\n"
" i <why> - interrupt with a reason (recoverable)\n"
" c - complete the box\n"
" q - quit (abandons the box, no judgment)\n"
" h - show this help\n"
" Enter - refresh the countdown"
)
def _headline(box) -> str:
if box.state == BoxState.running:
return (
f"focusing - {_mmss(box.remaining_seconds())} left "
f"of {_mmss(box.duration_seconds)}"
)
if box.state == BoxState.paused:
return f"paused - {_mmss(box.remaining_seconds)} left"
if box.state == BoxState.interrupted:
return f"interrupted - {_mmss(box.remaining_seconds())} left"
if box.state == BoxState.open:
return "fresh - starting now"
return box.state.value
def _progress_bar(box) -> str:
progress = box.progress_percent()
bar_len = 20
filled = bar_len * progress // 100
bar = "#" * filled + "-" * (bar_len - filled)
return f" [{bar}] {progress}%"
def _refresh(box, display: DisplayFn) -> None:
display(_headline(box))
if box.state == BoxState.running:
display(_progress_bar(box))
def _auto_enter(store, box, ctrl, display: DisplayFn):
"""On entry, auto-start a fresh box or auto-resume a paused one."""
if box.state == BoxState.open:
box = ctrl.start(box)
display("starting your focus session...")
_refresh(box, display)
elif box.state in (BoxState.paused, BoxState.interrupted):
box = ctrl.resume(box)
display("resuming where you left off...")
_refresh(box, display)
return box
def run_repl(
store,
box,
wait_input: InputFn,
display: DisplayFn,
tick_interval: float,
) -> ...:
"""Run the live focus loop and return the final box state.
Args:
store: a Store-like object supporting get_timebox / update_timebox.
box: the TimeBox to work on (its current state is respected).
wait_input: callable(timeout) -> input line or None on timeout.
display: callable(str) -> emits one line of output.
tick_interval: seconds between idle redraws (0 for event-driven).
"""
ctrl = BoxController(store)
display("focus mode (press h for help)")
display(HELP)
display("-" * 40)
# Re-fetch from store so external updates are visible.
box = store.get_timebox(box.id) if box.id is not None else box
box = _auto_enter(store, box, ctrl, display)
try:
while not box.is_terminal():
# Auto-expire if the timer ran out between ticks.
if box.state == BoxState.running and box.remaining_seconds() <= 0:
box = ctrl.expire(box)
display("")
display("timer ran out. take a breather.")
_refresh(box, display)
break
# Idle refresh while running (only when there's a tick cadence).
if tick_interval > 0:
_refresh(box, display)
time.sleep(tick_interval)
box = store.get_timebox(box.id) if box.id is not None else box
continue
# Event-driven: poll input non-blocking.
line = wait_input(0.0)
if line is None:
# No input yet; re-check expiry without blocking forever.
box = store.get_timebox(box.id) if box.id is not None else box
if box.state == BoxState.running and box.remaining_seconds() <= 0:
box = ctrl.expire(box)
display("timer ran out. take a breather.")
break
# Tiny yield so a pure-0 interval doesn't busy-spin in real use.
if tick_interval == 0:
continue
time.sleep(tick_interval)
continue
line = line.strip()
if line == "":
_refresh(box, display)
continue
parts = line.split(None, 1)
cmd = parts[0].lower()
arg = parts[1] if len(parts) > 1 else ""
try:
if cmd == "h":
display(HELP)
elif cmd == "p":
box = ctrl.pause(box)
display(f"paused - {_mmss(box.remaining_seconds())} saved")
elif cmd == "r":
box = ctrl.resume(box)
display(f"resumed - {_mmss(box.remaining_seconds())} to go")
elif cmd == "i":
if not arg:
display("an interrupt needs a reason: type 'i <why>'")
display("(this helps you notice patterns later)")
continue
box = ctrl.interrupt(box, reason=arg)
display(f"interrupted - {_mmss(box.remaining_seconds())} saved for later")
display(f" reason: {arg}")
elif cmd == "c":
box = ctrl.complete(box)
display(f"completed after {_mmss(box.elapsed_seconds())} - well done")
elif cmd == "q":
reason = arg or ""
box = ctrl.abandon(box, reason=reason)
display("abandoned. it's okay - come back when you want.")
else:
display(f"unknown command: '{cmd}'. press h for help.")
except ValueError as e:
display(f"can't do that: {e}")
display("press h for help.")
except KeyboardInterrupt:
# Ctrl+C in real use: abandon cleanly, no judgment.
if not box.is_terminal():
box = ctrl.abandon(box, reason="keyboard interrupt")
display("")
display("interrupted. box abandoned - it's okay, come back anytime.")
display("-" * 40)
display(f"box #{box.id} [{box.state.value}]")
return box

134
agent_todos/store.py
View File

@ -1,9 +1,13 @@
from __future__ import annotations
import json
import sqlite3
from datetime import datetime
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from .models import Priority, Status, Todo
from .timebox import BoxState, TimeBox
SCHEMA = """
CREATE TABLE IF NOT EXISTS todos (
@ -20,6 +24,21 @@ CREATE TABLE IF NOT EXISTS todos (
);
CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status);
CREATE INDEX IF NOT EXISTS idx_todos_repo ON todos(repo);
CREATE TABLE IF NOT EXISTS time_boxes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
todo_id INTEGER REFERENCES todos(id),
duration_seconds INTEGER NOT NULL,
state TEXT NOT NULL DEFAULT 'open',
accumulated_seconds REAL NOT NULL DEFAULT 0.0,
last_started_at TEXT,
interruptions TEXT NOT NULL DEFAULT '[]',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT,
reason TEXT
);
CREATE INDEX IF NOT EXISTS idx_timebox_state ON time_boxes(state);
"""
@ -145,3 +164,116 @@ class Store:
updated_at=row["updated_at"],
resolved_at=row["resolved_at"],
)
# ----- time_boxes -----------------------------------------------------
def create_timebox(self, box: TimeBox) -> TimeBox:
now = datetime.now(timezone.utc).isoformat()
box.created_at = now
box.updated_at = now
with sqlite3.connect(self.db_path) as con:
cur = con.execute(
"INSERT INTO time_boxes "
"(todo_id, duration_seconds, state, accumulated_seconds, "
"last_started_at, interruptions, created_at, updated_at, "
"completed_at, reason) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
box.todo_id,
box.duration_seconds,
box.state.value,
box.accumulated_seconds,
box.last_started_at,
json.dumps(box.interruptions),
box.created_at,
box.updated_at,
box.completed_at,
box.reason,
),
)
box.id = cur.lastrowid
con.commit()
return box
def get_timebox(self, box_id: int) -> Optional[TimeBox]:
with sqlite3.connect(self.db_path) as con:
con.row_factory = sqlite3.Row
row = con.execute(
"SELECT * FROM time_boxes WHERE id = ?", (box_id,)
).fetchone()
return self._row_to_timebox(row) if row else None
def list_timeboxes(self, active_only: bool = False) -> list[TimeBox]:
query = "SELECT * FROM time_boxes"
if active_only:
states = ", ".join(f"'{s.value}'" for s in BoxState if s not in {
BoxState.completed, BoxState.abandoned, BoxState.expired
})
query += f" WHERE state IN ({states})"
query += " ORDER BY updated_at DESC"
with sqlite3.connect(self.db_path) as con:
con.row_factory = sqlite3.Row
rows = con.execute(query).fetchall()
return [self._row_to_timebox(r) for r in rows]
def update_timebox(self, box_id: int, **kwargs) -> Optional[TimeBox]:
box = self.get_timebox(box_id)
if not box:
return None
allowed = {
"todo_id", "duration_seconds", "state", "accumulated_seconds",
"last_started_at", "interruptions", "completed_at", "reason",
}
fields = {k: v for k, v in kwargs.items() if k in allowed}
if not fields:
return box
now = datetime.now(timezone.utc).isoformat()
sets = ["updated_at = ?"]
params: list = [now]
for k, v in fields.items():
if k == "interruptions" and isinstance(v, list):
v = json.dumps(v)
if k == "state":
v = v.value if isinstance(v, BoxState) else v
sets.append(f"{k} = ?")
params.append(v)
params.append(box_id)
with sqlite3.connect(self.db_path) as con:
con.execute(
f"UPDATE time_boxes SET {', '.join(sets)} WHERE id = ?", params
)
con.commit()
return self.get_timebox(box_id)
def recover_timebox(self) -> Optional[TimeBox]:
"""Return the most-recent active box, or None if none exists.
Use this after an interruption (crash, reboot, agent restart) so the
user can resume exactly where they left off.
"""
states = ", ".join(f"'{s.value}'" for s in BoxState if s not in {
BoxState.completed, BoxState.abandoned, BoxState.expired
})
query = (
f"SELECT * FROM time_boxes WHERE state IN ({states}) "
"ORDER BY updated_at DESC LIMIT 1"
)
with sqlite3.connect(self.db_path) as con:
con.row_factory = sqlite3.Row
row = con.execute(query).fetchone()
return self._row_to_timebox(row) if row else None
def _row_to_timebox(self, row: sqlite3.Row) -> TimeBox:
return TimeBox(
id=row["id"],
todo_id=row["todo_id"],
duration_seconds=row["duration_seconds"],
state=BoxState(row["state"]),
accumulated_seconds=row["accumulated_seconds"] or 0.0,
last_started_at=row["last_started_at"],
interruptions=json.loads(row["interruptions"] or "[]"),
created_at=row["created_at"],
updated_at=row["updated_at"],
completed_at=row["completed_at"],
reason=row["reason"],
)

116
agent_todos/timebox.py Normal file
View File

@ -0,0 +1,116 @@
"""Time-boxing model and state machine for ADHD-friendly focus sessions.
A TimeBox is a single focus session on a TODO (or standalone). It tracks an
explicit, recoverable state machine so interruptions are never lost — you can
always pause, resume, interrupt-with-a-reason, or abandon without shame.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional
class BoxState(str, Enum):
"""Lifecycle of a time-box. All states are explicit and auditable."""
open = "open" # created, not yet started
running = "running" # actively counting down
paused = "paused" # paused by choice, recoverable
interrupted = "interrupted" # paused by external cause, recoverable
completed = "completed" # finished deliberately (terminal)
abandoned = "abandoned" # quit without finishing (terminal)
expired = "expired" # timer ran to zero (terminal)
# Adjacency list: which states each state may transition into.
# No terminal state has outgoing edges. Interrupted is functionally paused
# but tagged so the UI can surface it as a recoverable interruption.
_TRANSITIONS: dict[BoxState, set[BoxState]] = {
BoxState.open: {BoxState.running, BoxState.abandoned},
BoxState.running: {
BoxState.paused, BoxState.interrupted, BoxState.completed,
BoxState.abandoned, BoxState.expired,
},
BoxState.paused: {
BoxState.running, BoxState.interrupted, BoxState.completed,
BoxState.abandoned, BoxState.expired,
},
BoxState.interrupted: {
BoxState.running, BoxState.completed,
BoxState.abandoned, BoxState.expired,
},
BoxState.completed: set(),
BoxState.abandoned: set(),
BoxState.expired: set(),
}
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
@dataclass
class TimeBox:
id: Optional[int]
todo_id: Optional[int]
duration_seconds: int
state: BoxState = BoxState.open
accumulated_seconds: float = 0.0
last_started_at: Optional[str] = None
interruptions: list[dict[str, Any]] = field(default_factory=list)
created_at: str = field(default_factory=_now_iso)
updated_at: str = field(default_factory=_now_iso)
completed_at: Optional[str] = None
reason: Optional[str] = None # why interrupted / abandoned / completed
def __post_init__(self) -> None:
if self.duration_seconds <= 0:
raise ValueError("duration_seconds must be a positive integer")
def can(self, target: BoxState) -> bool:
"""True if the transition target is legal from the current state."""
return target in _TRANSITIONS.get(self.state, set())
def has_active_state(self) -> bool:
"""True while the box is still in play (not terminal)."""
return self.state in {
BoxState.open, BoxState.running,
BoxState.paused, BoxState.interrupted,
}
def is_terminal(self) -> bool:
return not self.has_active_state()
def _last_started_epoch(self) -> Optional[float]:
if self.last_started_at is None:
return None
try:
return datetime.fromisoformat(self.last_started_at).timestamp()
except (ValueError, TypeError):
return None
def elapsed_seconds(self) -> float:
"""Total focused time so far. For running boxes, includes live time."""
elapsed = float(self.accumulated_seconds)
started = self._last_started_epoch()
if self.state == BoxState.running and started is not None:
elapsed += max(0.0, datetime.now(timezone.utc).timestamp() - started)
return elapsed
def remaining_seconds(self) -> int:
"""Seconds left in the box, floored at zero."""
return max(0, self.duration_seconds - int(self.elapsed_seconds()))
def progress_percent(self) -> int:
"""How much of the box is used up, clamped 0..100."""
if self.duration_seconds <= 0:
return 100
return min(100, int(round(
100.0 * self.elapsed_seconds() / self.duration_seconds
)))
def started(self) -> bool:
"""Whether the box has ever been started (needed for restart recovery)."""
return self.last_started_at is not None

130
tests/test_box.py Normal file
View File

@ -0,0 +1,130 @@
"""Tests for the BoxController: state transitions + elapsed bookkeeping."""
import pytest
from agent_todos.box import BoxController
from agent_todos.timebox import BoxState, TimeBox
class FakeStore:
def __init__(self, box: TimeBox):
self._box = box
self.saved = []
def get_timebox(self, box_id):
return self._box
def update_timebox(self, box_id, **kwargs):
for k, v in kwargs.items():
setattr(self._box, k, v)
self.saved.append(kwargs)
return self._box
def _box(**kw):
base = dict(id=1, todo_id=None, duration_seconds=600, state=BoxState.open)
base.update(kw)
return TimeBox(**base)
def test_start_moves_open_to_running():
box = _box()
ctrl = BoxController(FakeStore(box))
out = ctrl.start(box)
assert out.state == BoxState.running
assert box.last_started_at is not None
def test_start_rejects_illegal_state():
box = _box(state=BoxState.paused)
ctrl = BoxController(FakeStore(box))
with pytest.raises(ValueError, match="Cannot start"):
ctrl.start(box)
def test_pause_accrues_elapsed_and_freezes():
box = _box(state=BoxState.running, last_started_at="2026-01-01T00:00:00Z")
# monkeypatch "now" via a fake by setting last_started far in past so elapsed > 0
box.last_started_at = "2000-01-01T00:00:00Z"
ctrl = BoxController(FakeStore(box))
out = ctrl.pause(box)
assert out.state == BoxState.paused
assert box.accumulated_seconds > 0
assert box.last_started_at is None # frozen
def test_pause_rejects_non_running():
box = _box(state=BoxState.open)
ctrl = BoxController(FakeStore(box))
with pytest.raises(ValueError, match="Cannot pause"):
ctrl.pause(box)
def test_resume_moves_paused_to_running():
box = _box(state=BoxState.paused, accumulated_seconds=300)
ctrl = BoxController(FakeStore(box))
out = ctrl.resume(box)
assert out.state == BoxState.running
assert box.last_started_at is not None
assert box.accumulated_seconds == 300 # preserved
def test_resume_rejects_non_paused():
box = _box(state=BoxState.open)
ctrl = BoxController(FakeStore(box))
with pytest.raises(ValueError, match="Cannot resume"):
ctrl.resume(box)
def test_interrupt_appends_reason_and_freezes():
box = _box(state=BoxState.running, accumulated_seconds=120)
box.last_started_at = "2000-01-01T00:00:00Z"
ctrl = BoxController(FakeStore(box))
out = ctrl.interrupt(box, reason="Slack DM from greg")
assert out.state == BoxState.interrupted
assert out.interruptions and out.interruptions[-1]["reason"] == "Slack DM from greg"
assert box.accumulated_seconds > 120 # accrued before freeze
assert box.last_started_at is None
assert box.reason == "Slack DM from greg"
def test_interrupt_on_paused_is_allowed():
box = _box(state=BoxState.paused, accumulated_seconds=60)
ctrl = BoxController(FakeStore(box))
out = ctrl.interrupt(box, reason="phone")
assert out.state == BoxState.interrupted
assert out.interruptions[-1]["reason"] == "phone"
def test_complete_marks_terminal_and_timestamp():
box = _box(state=BoxState.paused, accumulated_seconds=300)
ctrl = BoxController(FakeStore(box))
out = ctrl.complete(box)
assert out.state == BoxState.completed
assert box.completed_at is not None
def test_abandon_marks_terminal():
box = _box(state=BoxState.running, accumulated_seconds=30)
box.last_started_at = "2000-01-01T00:00:00Z"
ctrl = BoxController(FakeStore(box))
out = ctrl.abandon(box, reason="lost focus, switching tasks")
assert out.state == BoxState.abandoned
assert box.accumulated_seconds > 30 # accrues the live time
def test_expire_marks_terminal():
box = _box(state=BoxState.running, accumulated_seconds=590)
box.last_started_at = "2000-01-01T00:00:00Z" # way in past -> elapsed >= duration
ctrl = BoxController(FakeStore(box))
out = ctrl.expire(box)
assert out.state == BoxState.expired
assert box.elapsed_seconds() >= box.duration_seconds
def test_action_on_terminal_state_raises():
box = _box(state=BoxState.completed)
ctrl = BoxController(FakeStore(box))
with pytest.raises(ValueError, match="terminal"):
ctrl.resume(box)
with pytest.raises(ValueError, match="terminal"):
ctrl.complete(box)

168
tests/test_cli_box.py Normal file
View File

@ -0,0 +1,168 @@
"""CLI tests for the `box` command group (issue #16)."""
from click.testing import CliRunner
from agent_todos.cli import main
def _runner(tmp_path, db="todos.db"):
return (CliRunner(), str(tmp_path / db))
def test_box_help_lists_commands(tmp_path):
runner, _ = _runner(tmp_path)
result = runner.invoke(main, ["box", "--help"])
assert result.exit_code == 0
for cmd in ["start", "status", "pause", "resume", "interrupt", "complete", "abandon", "recover", "list"]:
assert cmd in result.output, f"{cmd} missing from box --help:\n{result.output}"
def test_start_default_25m(tmp_path):
runner, db = _runner(tmp_path)
result = runner.invoke(main, ["box", "start", "--path", db])
assert result.exit_code == 0
assert "25:00" in result.output
assert "started box" in result.output
def test_start_custom_minutes(tmp_path):
runner, db = _runner(tmp_path)
result = runner.invoke(main, ["box", "start", "10", "--path", db])
assert result.exit_code == 0
assert "10:00" in result.output
def test_start_with_todo_link(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["create", "deep work", "--path", db])
result = runner.invoke(main, ["box", "start", "5", "--todo", "1", "--path", db])
assert result.exit_code == 0
assert "started" in result.output
def test_start_rejects_out_of_range(tmp_path):
runner, db = _runner(tmp_path)
result = runner.invoke(main, ["box", "start", "500", "--path", db])
assert result.exit_code != 0
assert "0.5 and 240" in result.output
def test_status_before_any_box_is_friendly(tmp_path):
runner, db = _runner(tmp_path)
result = runner.invoke(main, ["box", "status", "--path", db])
assert result.exit_code != 0
assert "no active time-box" in result.output
def test_full_lifecycle_open_running_paused_completed(tmp_path):
runner, db = _runner(tmp_path)
r = runner.invoke(main, ["box", "start", "5", "--path", db])
assert r.exit_code == 0
box_id = r.output.split("#")[1].split(" ")[0]
r = runner.invoke(main, ["box", "status", "--path", db])
assert r.exit_code == 0
assert "focusing" in r.output
r = runner.invoke(main, ["box", "pause", "--path", db])
assert r.exit_code == 0
assert "paused" in r.output
r = runner.invoke(main, ["box", "resume", "--path", db])
assert r.exit_code == 0
assert "resumed" in r.output
r = runner.invoke(main, ["box", "complete", "--reason", "made progress", "--path", db])
assert r.exit_code == 0
assert "completed" in r.output
assert "made progress" in r.output
# after completion, status should be terminal-friendly
r = runner.invoke(main, ["box", "status", "--path", db])
assert r.exit_code != 0 # terminal box => active_box helper raises
assert "no active time-box" in r.output
def test_interrupt_and_resume(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["box", "start", "5", "--path", db])
r = runner.invoke(main, ["box", "interrupt", "meeting popped up", "--path", db])
assert r.exit_code == 0
assert "meeting popped up" in r.output
r = runner.invoke(main, ["box", "status", "--path", db])
assert r.exit_code == 0
assert "interrupted" in r.output
r = runner.invoke(main, ["box", "resume", "--path", db])
assert r.exit_code == 0
assert "resumed" in r.output
def test_abandon_is_nonjudgmental(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["box", "start", "5", "--path", db])
r = runner.invoke(main, ["box", "abandon", "--reason", "tired", "--path", db])
assert r.exit_code == 0
assert "it's okay" in r.output.lower() or "no judgment" in r.output.lower()
def test_pause_without_active_box_fails(tmp_path):
runner, db = _runner(tmp_path)
r = runner.invoke(main, ["box", "pause", "--path", db])
assert r.exit_code != 0
assert "no active time-box" in r.output
def test_recover_after_restart(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["box", "start", "5", "--path", db])
r = runner.invoke(main, ["box", "interrupt", "power blip", "--path", db])
assert r.exit_code == 0
# simulate a fresh CLI invocation by using a new CliRunner on same db
r = runner.invoke(main, ["box", "recover", "--path", db])
assert r.exit_code == 0
assert "recovered box" in r.output
assert "power blip" in r.output
def test_recover_nothing_returns_helpful_message(tmp_path):
runner, db = _runner(tmp_path)
r = runner.invoke(main, ["box", "recover", "--path", db])
assert r.exit_code == 0
assert "nothing to recover" in r.output
def test_list_empty_is_friendly(tmp_path):
runner, db = _runner(tmp_path)
r = runner.invoke(main, ["box", "list", "--path", db])
assert r.exit_code == 0
assert "no boxes yet" in r.output
def test_list_shows_active_boxes(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["box", "start", "5", "--path", db])
r = runner.invoke(main, ["box", "list", "--path", db])
assert r.exit_code == 0
assert "[open]" not in r.output # box started, so it's running
# should still show the running box
assert "running" in r.output
def test_pause_while_not_running_rejected(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["box", "start", "5", "--path", db])
runner.invoke(main, ["box", "pause", "--path", db])
r = runner.invoke(main, ["box", "pause", "--path", db])
assert r.exit_code != 0
assert "Cannot pause" in r.output
def test_complete_twice_fails_cleanly(tmp_path):
runner, db = _runner(tmp_path)
runner.invoke(main, ["box", "start", "5", "--path", db])
runner.invoke(main, ["box", "complete", "--path", db])
r = runner.invoke(main, ["box", "complete", "--path", db])
assert r.exit_code != 0 # no active box

222
tests/test_cli_repl.py Normal file
View File

@ -0,0 +1,222 @@
"""Tests for the ADHD-friendly live focus REPL (issue #16)."""
import pytest
from agent_todos import cli_repl
from agent_todos.box import BoxController
from agent_todos.store import Store
from agent_todos.timebox import BoxState, TimeBox
def _store(tmp_path, name="r.db"):
return Store(db_path=str(tmp_path / name))
def _box(duration=1500):
return TimeBox(id=None, todo_id=None, duration_seconds=duration)
def _seq(inputs):
"""Return a callable that yields items from `inputs`, then forever returns the last one."""
it = iter(inputs)
state = {"val": None, "done": False}
def _next(timeout):
if state["done"]:
return state["val"]
try:
state["val"] = next(it)
return state["val"]
except StopIteration:
state["done"] = True
return state["val"]
return _next
def test_repl_shows_help_and_headline_on_enter(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert any("25:00" in o for o in outputs)
assert any("p" in o and "pause" in o for o in outputs) # help lists commands
def test_repl_auto_starts_fresh_box(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box(duration=1500))
assert box.state == BoxState.open
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert any("focusing" in o for o in outputs)
def test_repl_auto_resumes_paused_box(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
ctrl = BoxController(store)
box = ctrl.start(box)
box = ctrl.pause(box)
assert box.state == BoxState.paused
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["c"]),
display=outputs.append,
tick_interval=0,
)
# should have resumed, then completed
assert final.state == BoxState.completed
assert any("focusing" in o for o in outputs)
def test_repl_pause_and_resume(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["p", "r", "c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert any("paused" in o.lower() for o in outputs)
assert any("resumed" in o.lower() or "focusing" in o for o in outputs)
def test_repl_interrupt_records_reason(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["i phone call", "c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert final.interruptions, "interruptions should be recorded"
assert final.interruptions[-1]["reason"] == "phone call"
def test_repl_interrupt_without_reason_prompts(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["i", "c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert any("needs a reason" in o for o in outputs)
def test_repl_quit_abandons_with_no_judgment(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["q"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.abandoned
ok_phrase = "it's okay" in " ".join(outputs).lower() or "no judgment" in " ".join(outputs).lower() or "come back" in " ".join(outputs).lower()
assert ok_phrase
def test_repl_auto_expires_when_timer_runs_out(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box(duration=3))
ctrl = BoxController(store)
box = ctrl.start(box)
# Force elapsed time far in the past so remaining == 0.
store.update_timebox(box.id, last_started_at="2000-01-01T00:00:00Z")
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq([None, None]), # never type anything; should auto-expire
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.expired
assert any("timer ran out" in o.lower() or "expired" in o.lower() for o in outputs)
def test_repl_unknown_command_shows_error_and_help_hint(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["xyz", "c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert any("unknown command" in o for o in outputs)
assert any("h" in o and "help" in o for o in outputs[-5:]) # hint to press h
def test_repl_help_command(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["h", "c"]),
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
# help text appears at least twice (on enter + on "h")
help_lines = [o for o in outputs if "commands" in o.lower() and "pause" in o.lower()]
assert len(help_lines) >= 2
def test_repl_refuses_pause_when_already_paused(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(_box())
outputs = []
final = cli_repl.run_repl(
store, box,
wait_input=_seq(["p", "p", "c"]), # pause, then pause again (should error), then complete
display=outputs.append,
tick_interval=0,
)
assert final.state == BoxState.completed
assert any("can't do that" in o for o in outputs)
from agent_todos.models import Priority, Todo
def test_repl_preserves_data_integrity(tmp_path):
"""Existing TODO data must be untouched by the focus session."""
store = _store(tmp_path)
store.create(Todo(id=None, title="my task", priority=Priority.high))
todo_count_before = len(store.list())
box = store.create_timebox(_box())
cli_repl.run_repl(
store, box,
wait_input=_seq(["c"]),
display=lambda _: None,
tick_interval=0,
)
assert len(store.list()) == todo_count_before

126
tests/test_store_timebox.py Normal file
View File

@ -0,0 +1,126 @@
import pytest
from agent_todos.models import Priority, Status, Todo
from agent_todos.store import Store
from agent_todos.timebox import BoxState, TimeBox
def _store(tmp_path, name="tb.db"):
return Store(db_path=str(tmp_path / name))
# --- schema / round-trip -------------------------------------------------
def test_schema_created_on_init(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(
TimeBox(id=None, todo_id=None, duration_seconds=1500, state=BoxState.open)
)
assert box.id is not None
def test_create_with_todo_link(tmp_path):
store = _store(tmp_path)
todo = store.create(Todo(id=None, title="focus task", priority=Priority.high))
box = store.create_timebox(
TimeBox(id=None, todo_id=todo.id, duration_seconds=900)
)
fetched = store.get_timebox(box.id)
assert fetched.todo_id == todo.id
assert fetched.duration_seconds == 900
assert fetched.state == BoxState.open
def test_persists_interruption_and_reason(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(
TimeBox(
id=None,
todo_id=None,
duration_seconds=600,
state=BoxState.interrupted,
reason="phone rang",
interruptions=[{"at": "2026-01-01T00:00:00Z", "reason": "phone rang"}],
)
)
fetched = store.get_timebox(box.id)
assert fetched.reason == "phone rang"
assert fetched.interruptions == [{"at": "2026-01-01T00:00:00Z", "reason": "phone rang"}]
def test_list_active_excludes_terminal(tmp_path):
store = _store(tmp_path)
store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.open))
store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.running))
store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.completed))
store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.abandoned))
active = store.list_timeboxes(active_only=True)
states = {b.state for b in active}
assert BoxState.completed not in states
assert BoxState.abandoned not in states
assert BoxState.expired not in states
assert states <= {BoxState.open, BoxState.running, BoxState.paused, BoxState.interrupted}
# --- recovery ------------------------------------------------------------
def test_recover_active_box_returns_most_recent(tmp_path):
store = _store(tmp_path)
older = store.create_timebox(
TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.paused)
)
newer = store.create_timebox(
TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.interrupted,
reason="got pulled into a call")
)
recovered = store.recover_timebox()
assert recovered is not None
# most-recent by updated_at
assert recovered.id in {older.id, newer.id}
assert recovered.has_active_state()
def test_recover_returns_none_when_all_terminal(tmp_path):
store = _store(tmp_path)
store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.completed))
store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.abandoned))
assert store.recover_timebox() is None
def test_recover_returns_none_on_empty_store(tmp_path):
store = _store(tmp_path)
assert store.recover_timebox() is None
def test_get_nonexistent_returns_none(tmp_path):
store = _store(tmp_path)
assert store.get_timebox(9999) is None
# --- update --------------------------------------------------------------
def test_update_state_sets_timestamp(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=600))
updated = store.update_timebox(box.id, state=BoxState.running)
assert updated.state == BoxState.running
assert updated.updated_at >= box.updated_at
def test_update_unknown_returns_none(tmp_path):
store = _store(tmp_path)
assert store.update_timebox(9999, state=BoxState.completed) is None
# --- relationship with todos --------------------------------------------
def test_create_timebox_without_todo_is_allowed(tmp_path):
store = _store(tmp_path)
box = store.create_timebox(TimeBox(id=None, todo_id=None, duration_seconds=300))
assert box.id is not None
fetched = store.get_timebox(box.id)
assert fetched.todo_id is None

109
tests/test_timebox.py Normal file
View File

@ -0,0 +1,109 @@
import pytest
from agent_todos.timebox import BoxState, TimeBox
def test_timebox_default_state_is_open():
tb = TimeBox(id=None, todo_id=None, duration_seconds=1500)
assert tb.state == BoxState.open
assert tb.accumulated_seconds == 0
assert tb.last_started_at is None
assert tb.interruptions == []
def test_transitions_from_open():
tb = TimeBox(id=None, todo_id=None, duration_seconds=600)
# open can go to running (start) or be abandoned immediately
assert tb.can(BoxState.running) is True
assert tb.can(BoxState.abandoned) is True
assert tb.can(BoxState.completed) is False # nothing was worked on
assert tb.can(BoxState.expired) is False
def test_transitions_from_running():
tb = TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.running)
assert tb.can(BoxState.paused) is True
assert tb.can(BoxState.interrupted) is True
assert tb.can(BoxState.completed) is True
assert tb.can(BoxState.abandoned) is True
assert tb.can(BoxState.expired) is True
assert tb.can(BoxState.open) is False # cannot rewind to fresh
assert tb.can(BoxState.running) is False # no self-loops
def test_transitions_from_paused():
tb = TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.paused)
assert tb.can(BoxState.running) is True # resume
assert tb.can(BoxState.interrupted) is True
assert tb.can(BoxState.completed) is True
assert tb.can(BoxState.abandoned) is True
assert tb.can(BoxState.expired) is True
assert tb.can(BoxState.open) is False
def test_transitions_from_interrupted():
# interrupted is paused-with-a-reason: still recoverable to running
tb = TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.interrupted)
assert tb.can(BoxState.running) is True
assert tb.can(BoxState.completed) is True
assert tb.can(BoxState.abandoned) is True
assert tb.can(BoxState.expired) is True
assert tb.can(BoxState.open) is False
def test_terminal_states_accept_no_transitions():
for terminal in (BoxState.completed, BoxState.abandoned, BoxState.expired):
tb = TimeBox(id=None, todo_id=None, duration_seconds=600, state=terminal)
for s in BoxState:
assert tb.can(s) is False
def test_elapsed_math_paused_uses_accumulated_only():
tb = TimeBox(
id=None,
todo_id=None,
duration_seconds=600,
state=BoxState.paused,
accumulated_seconds=120,
)
assert tb.elapsed_seconds() == 120
assert tb.remaining_seconds() == 480
def test_remaining_is_floor_zero():
tb = TimeBox(
id=None,
todo_id=None,
duration_seconds=600,
state=BoxState.paused,
accumulated_seconds=1200,
)
assert tb.remaining_seconds() == 0
def test_percent_progress_clamps_to_100():
tb = TimeBox(
id=None,
todo_id=None,
duration_seconds=600,
state=BoxState.paused,
accumulated_seconds=900,
)
assert tb.progress_percent() == 100
def test_duration_must_be_positive():
with pytest.raises(ValueError):
TimeBox(id=None, todo_id=None, duration_seconds=0)
with pytest.raises(ValueError):
TimeBox(id=None, todo_id=None, duration_seconds=-5)
def test_has_active_nonterminal_state():
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.open).has_active_state() is True
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.running).has_active_state() is True
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.paused).has_active_state() is True
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.interrupted).has_active_state() is True
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.completed).has_active_state() is False
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.abandoned).has_active_state() is False
assert TimeBox(id=None, todo_id=None, duration_seconds=600, state=BoxState.expired).has_active_state() is False