feat: scaffold agent-todo-tracker P2 project
This commit is contained in:
commit
1c20079608
5
.gitignore
vendored
Normal file
5
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,5 @@
|
||||||
|
__pycache__/
|
||||||
|
*.pyc
|
||||||
|
.agent_todos.db
|
||||||
|
dist/
|
||||||
|
*.egg-info/
|
||||||
35
README.md
Normal file
35
README.md
Normal file
|
|
@ -0,0 +1,35 @@
|
||||||
|
# Agent TODO Tracker
|
||||||
|
|
||||||
|
Lightweight TODO CLI for agent engineers across repos.
|
||||||
|
|
||||||
|
## Install
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pip install -e .
|
||||||
|
```
|
||||||
|
|
||||||
|
## Usage
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Initialize DB
|
||||||
|
agent-todo init
|
||||||
|
|
||||||
|
# Create a TODO
|
||||||
|
agent-todo create -t "Implement auth" -p high -r stackchain/stackchain-hackathon
|
||||||
|
|
||||||
|
# List open TODOs
|
||||||
|
agent-todo list
|
||||||
|
|
||||||
|
# Update status
|
||||||
|
agent-todo update 1 -s in_progress
|
||||||
|
|
||||||
|
# Resolve
|
||||||
|
agent-todo close 1
|
||||||
|
|
||||||
|
# JSON output
|
||||||
|
agent-todo list --json
|
||||||
|
```
|
||||||
|
|
||||||
|
## Storage
|
||||||
|
|
||||||
|
Default: `.agent_todos.db` in current directory.
|
||||||
1
agent_todos/__init__.py
Normal file
1
agent_todos/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
||||||
|
__version__ = "0.1.0"
|
||||||
105
agent_todos/cli.py
Normal file
105
agent_todos/cli.py
Normal file
|
|
@ -0,0 +1,105 @@
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import click
|
||||||
|
|
||||||
|
from .models import Priority, Status
|
||||||
|
from .store import Store
|
||||||
|
|
||||||
|
|
||||||
|
@click.group()
|
||||||
|
def main() -> None:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
@main.command()
|
||||||
|
@click.option("--path", default=".agent_todos.db", help="SQLite DB path.")
|
||||||
|
def init(path: str) -> None:
|
||||||
|
Store(db_path=path)
|
||||||
|
click.echo(f"Initialized TODO DB at {path}")
|
||||||
|
|
||||||
|
|
||||||
|
@main.command()
|
||||||
|
@click.argument("title")
|
||||||
|
@click.option("-r", "--repo", default="", help="Repo slug, e.g. org/repo")
|
||||||
|
@click.option("-p", "--priority", default="medium", type=click.Choice(["high", "medium", "low"]))
|
||||||
|
@click.option("-a", "--assignee", default="")
|
||||||
|
@click.option("-t", "--tags", multiple=True)
|
||||||
|
@click.option("--path", default=".agent_todos.db")
|
||||||
|
def create(title: str, repo: str, priority: str, assignee: str, tags: tuple[str, ...], path: str) -> None:
|
||||||
|
todo = Store(db_path=path).create(
|
||||||
|
_make_todo(title=title, repo=repo, priority=priority, assignee=assignee, tags=list(tags))
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@main.command()
|
||||||
|
@click.option("--repo")
|
||||||
|
@click.option("--status", type=click.Choice(["open", "in_progress", "blocked", "resolved"]))
|
||||||
|
@click.option("--json", "as_json", is_flag=True)
|
||||||
|
@click.option("--path", default=".agent_todos.db")
|
||||||
|
def list(repo, status, as_json, path): # type: ignore[no-untyped-def]
|
||||||
|
s = Status(status) if status else None
|
||||||
|
todos = Store(db_path=path).list(status=s, repo=repo)
|
||||||
|
if as_json:
|
||||||
|
click.echo(json.dumps([_serialize(t) for t in todos], indent=2))
|
||||||
|
else:
|
||||||
|
for t in todos:
|
||||||
|
click.echo(f"#{t.id} [{t.status.value}] {t.title} ({t.repo or 'no-repo'}) — p:{t.priority.value}")
|
||||||
|
|
||||||
|
|
||||||
|
@main.command()
|
||||||
|
@click.argument("todo_id", type=int)
|
||||||
|
@click.option("-s", "--status", type=click.Choice(["open", "in_progress", "blocked", "resolved"]))
|
||||||
|
@click.option("-t", "--title")
|
||||||
|
@click.option("-p", "--priority", type=click.Choice(["high", "medium", "low"]))
|
||||||
|
@click.option("-a", "--assignee")
|
||||||
|
@click.option("--path", default=".agent_todos.db")
|
||||||
|
def update(todo_id: int, status, title, priority, assignee, path: str) -> None:
|
||||||
|
fields: dict = {}
|
||||||
|
if status:
|
||||||
|
fields["status"] = Status(status)
|
||||||
|
if title:
|
||||||
|
fields["title"] = title
|
||||||
|
if priority:
|
||||||
|
fields["priority"] = Priority(priority)
|
||||||
|
if assignee is not None:
|
||||||
|
fields["assignee"] = assignee
|
||||||
|
todo = Store(db_path=path).update(todo_id, **fields)
|
||||||
|
if todo:
|
||||||
|
click.echo(f"Updated #{todo.id}: {todo.title} -> {todo.status.value}")
|
||||||
|
else:
|
||||||
|
raise click.ClickException("TODO not found")
|
||||||
|
|
||||||
|
|
||||||
|
@main.command()
|
||||||
|
@click.argument("todo_id", type=int)
|
||||||
|
@click.option("--path", default=".agent_todos.db")
|
||||||
|
def close(todo_id: int, path: str) -> None:
|
||||||
|
todo = Store(db_path=path).update(todo_id, status=Status.resolved)
|
||||||
|
if todo:
|
||||||
|
click.echo(f"Resolved #{todo.id}: {todo.title}")
|
||||||
|
else:
|
||||||
|
raise click.ClickException("TODO not found")
|
||||||
|
|
||||||
|
|
||||||
|
def _make_todo(*, title: str, repo: str, priority: str, assignee: str, tags: list[str]) -> Todo:
|
||||||
|
return Todo(id=None, title=title, repo=repo, priority=Priority(priority), assignee=assignee, tags=tags)
|
||||||
|
|
||||||
|
|
||||||
|
def _serialize(todo: Todo) -> dict:
|
||||||
|
return {
|
||||||
|
"id": todo.id,
|
||||||
|
"title": todo.title,
|
||||||
|
"repo": todo.repo,
|
||||||
|
"status": todo.status.value,
|
||||||
|
"priority": todo.priority.value,
|
||||||
|
"assignee": todo.assignee,
|
||||||
|
"tags": todo.tags,
|
||||||
|
"created_at": todo.created_at,
|
||||||
|
"updated_at": todo.updated_at,
|
||||||
|
"resolved_at": todo.resolved_at,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
26
agent_todos/gitea.py
Normal file
26
agent_todos/gitea.py
Normal file
|
|
@ -0,0 +1,26 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Iterable
|
||||||
|
|
||||||
|
from .models import Priority, Status, Todo
|
||||||
|
|
||||||
|
|
||||||
|
def repo_labels(todos: Iterable[Todo]) -> dict[str, list[str]]:
|
||||||
|
grouped: dict[str, list[str]] = {}
|
||||||
|
for todo in todos:
|
||||||
|
grouped.setdefault(todo.repo, []).append(todo.title)
|
||||||
|
return grouped
|
||||||
|
|
||||||
|
|
||||||
|
def gitea_issue_payload(todo: Todo, org: str = "stackchain") -> dict:
|
||||||
|
title = f"TODO: {todo.title}"
|
||||||
|
labels = [todo.priority.value, todo.status.value]
|
||||||
|
labels.extend(todo.tags)
|
||||||
|
body = (
|
||||||
|
f"**Status:** {todo.status.value}\n"
|
||||||
|
f"**Priority:** {todo.priority.value}\n"
|
||||||
|
f"**Repo:** {todo.repo}\n"
|
||||||
|
f"**Assignee:** {todo.assignee or 'unassigned'}\n\n"
|
||||||
|
"Auto-generated by agent-todo-tracker."
|
||||||
|
)
|
||||||
|
return {"title": title, "body": body, "labels": labels}
|
||||||
31
agent_todos/models.py
Normal file
31
agent_todos/models.py
Normal file
|
|
@ -0,0 +1,31 @@
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from datetime import datetime
|
||||||
|
from enum import Enum
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
|
||||||
|
class Priority(str, Enum):
|
||||||
|
high = "high"
|
||||||
|
medium = "medium"
|
||||||
|
low = "low"
|
||||||
|
|
||||||
|
|
||||||
|
class Status(str, Enum):
|
||||||
|
open = "open"
|
||||||
|
in_progress = "in_progress"
|
||||||
|
blocked = "blocked"
|
||||||
|
resolved = "resolved"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Todo:
|
||||||
|
id: Optional[int]
|
||||||
|
title: str
|
||||||
|
repo: str = ""
|
||||||
|
status: Status = Status.open
|
||||||
|
priority: Priority = Priority.medium
|
||||||
|
assignee: str = ""
|
||||||
|
tags: list[str] = field(default_factory=list)
|
||||||
|
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
|
||||||
|
updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
|
||||||
|
resolved_at: Optional[str] = None
|
||||||
147
agent_todos/store.py
Normal file
147
agent_todos/store.py
Normal file
|
|
@ -0,0 +1,147 @@
|
||||||
|
import sqlite3
|
||||||
|
from datetime import datetime
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from .models import Priority, Status, Todo
|
||||||
|
|
||||||
|
SCHEMA = """
|
||||||
|
CREATE TABLE IF NOT EXISTS todos (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
title TEXT NOT NULL,
|
||||||
|
repo TEXT DEFAULT '',
|
||||||
|
status TEXT NOT NULL DEFAULT 'open',
|
||||||
|
priority TEXT NOT NULL DEFAULT 'medium',
|
||||||
|
assignee TEXT DEFAULT '',
|
||||||
|
tags TEXT DEFAULT '',
|
||||||
|
created_at TEXT NOT NULL,
|
||||||
|
updated_at TEXT NOT NULL,
|
||||||
|
resolved_at TEXT DEFAULT NULL
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_todos_repo ON todos(repo);
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
class Store:
|
||||||
|
def __init__(self, db_path: str = ".agent_todos.db") -> None:
|
||||||
|
self.db_path = db_path
|
||||||
|
self._ensure()
|
||||||
|
|
||||||
|
def _ensure(self) -> None:
|
||||||
|
first = not Path(self.db_path).exists()
|
||||||
|
with sqlite3.connect(self.db_path) as con:
|
||||||
|
con.executescript(SCHEMA)
|
||||||
|
con.commit()
|
||||||
|
if first:
|
||||||
|
self.seed()
|
||||||
|
|
||||||
|
def seed(self) -> None:
|
||||||
|
now = datetime.utcnow().isoformat() + "Z"
|
||||||
|
self.create(Todo(
|
||||||
|
id=None,
|
||||||
|
title="Scaffold agent-todo-tracker CLI",
|
||||||
|
repo="stackchain/agent-todo-tracker",
|
||||||
|
status=Status.resolved,
|
||||||
|
priority=Priority.high,
|
||||||
|
tags=["hackathon", "p2"],
|
||||||
|
created_at=now,
|
||||||
|
updated_at=now,
|
||||||
|
resolved_at=now,
|
||||||
|
))
|
||||||
|
self.create(Todo(
|
||||||
|
id=None,
|
||||||
|
title="Add Gitea issue sync for TODOs",
|
||||||
|
repo="stackchain/stackchain-hackathon",
|
||||||
|
status=Status.open,
|
||||||
|
priority=Priority.medium,
|
||||||
|
tags=["hackathon", "p1"],
|
||||||
|
created_at=now,
|
||||||
|
updated_at=now,
|
||||||
|
))
|
||||||
|
|
||||||
|
def create(self, todo: Todo) -> Todo:
|
||||||
|
now = datetime.utcnow().isoformat() + "Z"
|
||||||
|
todo.created_at = now
|
||||||
|
todo.updated_at = now
|
||||||
|
with sqlite3.connect(self.db_path) as con:
|
||||||
|
cur = con.execute(
|
||||||
|
"INSERT INTO todos (title, repo, status, priority, assignee, tags, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
|
(
|
||||||
|
todo.title,
|
||||||
|
todo.repo,
|
||||||
|
todo.status.value,
|
||||||
|
todo.priority.value,
|
||||||
|
todo.assignee,
|
||||||
|
",".join(todo.tags),
|
||||||
|
todo.created_at,
|
||||||
|
todo.updated_at,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
todo.id = cur.lastrowid
|
||||||
|
con.commit()
|
||||||
|
return todo
|
||||||
|
|
||||||
|
def list(self, status: Optional[Status] = None, repo: Optional[str] = None) -> list[Todo]:
|
||||||
|
query = "SELECT * FROM todos WHERE 1=1"
|
||||||
|
params: list = []
|
||||||
|
if status:
|
||||||
|
query += " AND status = ?"
|
||||||
|
params.append(status.value)
|
||||||
|
if repo:
|
||||||
|
query += " AND repo = ?"
|
||||||
|
params.append(repo)
|
||||||
|
query += " ORDER BY created_at DESC"
|
||||||
|
with sqlite3.connect(self.db_path) as con:
|
||||||
|
con.row_factory = sqlite3.Row
|
||||||
|
rows = con.execute(query, params).fetchall()
|
||||||
|
return [self._row_to_todo(r) for r in rows]
|
||||||
|
|
||||||
|
def update(self, todo_id: int, **kwargs) -> Optional[Todo]:
|
||||||
|
todo = self.get(todo_id)
|
||||||
|
if not todo:
|
||||||
|
return None
|
||||||
|
allowed = {"title", "repo", "status", "priority", "assignee", "tags", "resolved_at"}
|
||||||
|
fields = {k: v for k, v in kwargs.items() if k in allowed}
|
||||||
|
if not fields:
|
||||||
|
return todo
|
||||||
|
now = datetime.utcnow().isoformat() + "Z"
|
||||||
|
sets = ["updated_at = ?"]
|
||||||
|
params = [now]
|
||||||
|
for k, v in fields.items():
|
||||||
|
if k == "tags" and isinstance(v, list):
|
||||||
|
v = ",".join(v)
|
||||||
|
if k == "status":
|
||||||
|
v = v.value if isinstance(v, Status) else v
|
||||||
|
if k == "priority":
|
||||||
|
v = v.value if isinstance(v, Priority) else v
|
||||||
|
sets.append(f"{k} = ?")
|
||||||
|
params.append(v)
|
||||||
|
if fields.get("status") == Status.resolved.value:
|
||||||
|
sets.append("resolved_at = ?")
|
||||||
|
params.append(now)
|
||||||
|
params.append(todo_id)
|
||||||
|
with sqlite3.connect(self.db_path) as con:
|
||||||
|
con.execute(f"UPDATE todos SET {', '.join(sets)} WHERE id = ?", params)
|
||||||
|
con.commit()
|
||||||
|
return self.get(todo_id)
|
||||||
|
|
||||||
|
def get(self, todo_id: int) -> Optional[Todo]:
|
||||||
|
with sqlite3.connect(self.db_path) as con:
|
||||||
|
con.row_factory = sqlite3.Row
|
||||||
|
row = con.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
|
||||||
|
return self._row_to_todo(row) if row else None
|
||||||
|
|
||||||
|
def _row_to_todo(self, row: sqlite3.Row) -> Todo:
|
||||||
|
return Todo(
|
||||||
|
id=row["id"],
|
||||||
|
title=row["title"],
|
||||||
|
repo=row["repo"],
|
||||||
|
status=Status(row["status"]),
|
||||||
|
priority=Priority(row["priority"]),
|
||||||
|
assignee=row["assignee"],
|
||||||
|
tags=[t for t in row["tags"].split(",") if t],
|
||||||
|
created_at=row["created_at"],
|
||||||
|
updated_at=row["updated_at"],
|
||||||
|
resolved_at=row["resolved_at"],
|
||||||
|
)
|
||||||
13
pyproject.toml
Normal file
13
pyproject.toml
Normal file
|
|
@ -0,0 +1,13 @@
|
||||||
|
[build-system]
|
||||||
|
requires = ["setuptools>=61.0"]
|
||||||
|
build-backend = "setuptools.build_meta"
|
||||||
|
|
||||||
|
[project]
|
||||||
|
name = "agent-todo-tracker"
|
||||||
|
version = "0.1.0"
|
||||||
|
description = "Lightweight TODO tracker for agent engineers"
|
||||||
|
requires-python = ">=3.9"
|
||||||
|
dependencies = []
|
||||||
|
|
||||||
|
[project.scripts]
|
||||||
|
agent-todo = "agent_todos.cli:main"
|
||||||
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
29
tests/test_store.py
Normal file
29
tests/test_store.py
Normal file
|
|
@ -0,0 +1,29 @@
|
||||||
|
from agent_todos.models import Priority, Status, Todo
|
||||||
|
from agent_todos.store import Store
|
||||||
|
|
||||||
|
|
||||||
|
def _store(tmp_path, name="test.db") -> tuple[Store, str]:
|
||||||
|
p = str(tmp_path / name)
|
||||||
|
return Store(db_path=p), p
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_and_list(tmp_path):
|
||||||
|
store, _ = _store(tmp_path)
|
||||||
|
todo = store.create(Todo(id=None, title="Test task", repo="org/repo", priority=Priority.high))
|
||||||
|
assert todo.id is not None
|
||||||
|
all_todos = store.list()
|
||||||
|
assert any(t.id == todo.id for t in all_todos)
|
||||||
|
|
||||||
|
|
||||||
|
def test_update_status(tmp_path):
|
||||||
|
store, _ = _store(tmp_path)
|
||||||
|
todo = store.create(Todo(id=None, title="Test task"))
|
||||||
|
updated = store.update(todo.id, status=Status.in_progress)
|
||||||
|
assert updated.status == Status.in_progress
|
||||||
|
|
||||||
|
|
||||||
|
def test_close_resolves(tmp_path):
|
||||||
|
store, _ = _store(tmp_path)
|
||||||
|
todo = store.create(Todo(id=None, title="Test task"))
|
||||||
|
store.update(todo.id, status=Status.resolved)
|
||||||
|
assert store.get(todo.id).resolved_at is not None
|
||||||
Loading…
Reference in New Issue
Block a user