Menu
akurai-tasks
publicLatest change cb00647e2dddee7d80e8f67a6858572f544a2ed6 - plan: track Framework convergence for AkurAI-Tasks by Ólafur Búi Ólafsson
#!/usr/bin/env python3
"""pm — manage the markdown project state in .plan/ .tasks/ .docs/ .memories/.
Commands:
new task <slug> [--epic <epic>] create a task from the template
new epic <slug> create an epic
new doc <slug> create a doc
status <slug> <value> set task/epic status, bump Updated
list tasks grouped by status
sync regenerate TASKS.md, PLAN.md epic table, docs INDEX.md
validate check fields, links, index drift, secret leaks
log "<message>" dated entry in .memories/MEMORIES.md
search "<query>" [-n N] semantic search over all state files (local TEI servers)
"""
import argparse
import datetime
import hashlib
import json
import math
import os
import re
import sys
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parent
EMBED_URL = os.environ.get("PM_EMBED_URL", "http://192.168.1.10:8081")
RERANK_URL = os.environ.get("PM_RERANK_URL", "http://192.168.1.10:8082")
STATE_DIRS = (".plan", ".plan/epics", ".tasks", ".docs", ".memories", ".codebase_index")
STATUSES = ("backlog", "active", "blocked", "done")
EPIC_STATUSES = ("planned", "active", "done")
INDEXES = {"TASKS.md", "INDEX.md", "PLAN.md", "MEMORIES.md", "_template.md"}
SECRET_PATTERNS = (
r"-----BEGIN",
r"AKIA[0-9A-Z]{16}",
r"eyJ[A-Za-z0-9_-]{20,}",
r"sk-[A-Za-z0-9]{20,}",
r"(?i)(password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S{8,}",
)
def today():
return datetime.date.today().isoformat()
def rel(p):
return str(p.relative_to(ROOT))
def field(text, name):
m = re.search(rf"^- \*\*{name}:\*\* (.+)$", text, re.M)
return m.group(1).strip() if m else None
def set_field(text, name, value):
return re.sub(rf"^(- \*\*{name}:\*\*) .+$", rf"\g<1> {value}", text, count=1, flags=re.M)
def entries(d):
return [p for p in sorted(d.glob("*.md")) if p.name not in INDEXES]
def new(kind, slug, epic=None):
if not re.fullmatch(r"[a-z0-9][a-z0-9-]*", slug):
sys.exit(f"bad slug {slug!r}: use lowercase-with-dashes")
dirs = {"task": ROOT / ".tasks", "epic": ROOT / ".plan/epics", "doc": ROOT / ".docs"}
path = dirs[kind] / f"{slug}.md"
if path.exists():
sys.exit(f"{rel(path)} already exists")
title = slug.replace("-", " ").title()
body = (dirs[kind] / "_template.md").read_text()
body = body.replace("<name>", title).replace("<Topic>", title).replace("YYYY-MM-DD", today())
if kind == "task":
if epic and not (ROOT / f".plan/epics/{epic}.md").exists():
sys.exit(f"epic {epic!r} does not exist — pm new epic {epic}")
body = set_field(body, "Status", "backlog")
body = set_field(body, "Epic", f"[{epic}](../.plan/epics/{epic}.md)" if epic else "none")
elif kind == "epic":
body = set_field(body, "Status", "planned")
path.write_text(body)
sync()
print(rel(path))
def set_status(slug, value):
for d, allowed in ((ROOT / ".tasks", STATUSES), (ROOT / ".plan/epics", EPIC_STATUSES)):
path = d / f"{slug}.md"
if path.exists():
if value not in allowed:
sys.exit(f"status for {slug} must be one of: {', '.join(allowed)}")
text = set_field(path.read_text(), "Status", value)
path.write_text(set_field(text, "Updated", today()))
sync()
return
sys.exit(f"no task or epic named {slug!r}")
def replace_table(path, header, sep, rows):
ncols = header.count("|") - 1
table = "\n".join([header, sep] + (rows or ["|" + " — |" * ncols])) + "\n"
text = re.sub(re.escape(header) + r"\n(\|[^\n]*\n?)+", table, path.read_text(), count=1)
path.write_text(text)
def sync():
groups = {"Active": [], "Backlog": [], "Done": []}
for p in entries(ROOT / ".tasks"):
st = field(p.read_text(), "Status") or "?"
line = f"- [{p.stem}]({p.name})" + (" — blocked" if st == "blocked" else "")
groups[{"active": "Active", "blocked": "Active", "done": "Done"}.get(st, "Backlog")].append(line)
out = ["# Tasks", "", "> Generated by `pm sync` — edit the task files, not this list.", ""]
for k, lines in groups.items():
out += [f"## {k}"] + (lines or ["- —"]) + [""]
(ROOT / ".tasks/TASKS.md").write_text("\n".join(out))
rows = [f"| {p.stem} | {field(p.read_text(), 'Status') or '?'} | [{p.name}](epics/{p.name}) |"
for p in entries(ROOT / ".plan/epics")]
replace_table(ROOT / ".plan/PLAN.md", "| Epic | Status | File |", "|------|--------|------|", rows)
rows = []
for p in entries(ROOT / ".docs"):
m = re.search(r"## Summary\n+([^\n]+)", p.read_text())
rows.append(f"| [{p.stem}]({p.name}) | {m.group(1).strip() if m else '—'} |")
replace_table(ROOT / ".docs/INDEX.md", "| Doc | What it covers |", "|-----|----------------|", rows)
def _check_entry(p, allowed, errs):
text = p.read_text()
st = field(text, "Status")
if st not in allowed:
errs.append(f"{rel(p)}: status must be one of {', '.join(allowed)} (got {st!r})")
up = field(text, "Updated")
if not (up and re.fullmatch(r"\d{4}-\d{2}-\d{2}", up)):
errs.append(f"{rel(p)}: Updated must be YYYY-MM-DD (got {up!r})")
def validate():
errs = []
for p in entries(ROOT / ".tasks"):
_check_entry(p, STATUSES, errs)
for p in entries(ROOT / ".plan/epics"):
_check_entry(p, EPIC_STATUSES, errs)
for d in STATE_DIRS:
for p in (ROOT / d).glob("*.md"):
if p.name == "_template.md":
continue
for m in re.finditer(r"\]\(([^)#\s]+\.md)\)", p.read_text()):
if not (p.parent / m.group(1)).exists():
errs.append(f"{rel(p)}: broken link {m.group(1)}")
for p in (ROOT / ".memories").glob("*.md"):
text = p.read_text()
for pat in SECRET_PATTERNS:
if re.search(pat, text):
errs.append(f"{rel(p)}: looks like a secret value ({pat}) — store locations, not values")
for d, idx in ((".tasks", ".tasks/TASKS.md"),
(".plan/epics", ".plan/PLAN.md"),
(".docs", ".docs/INDEX.md")):
listed = (ROOT / idx).read_text()
for p in entries(ROOT / d):
if p.name not in listed:
errs.append(f"{idx}: missing {p.name} — run `pm sync`")
return errs
def list_cmd():
rank = {s: i for i, s in enumerate(("active", "blocked", "backlog", "done"))}
items = [(p, p.read_text()) for p in entries(ROOT / ".tasks")]
for p, text in sorted(items, key=lambda it: rank.get(field(it[1], "Status"), 9)):
print(f"{field(text, 'Status') or '?':8} {p.stem} (updated {field(text, 'Updated')})")
def log(message):
p = ROOT / ".memories/MEMORIES.md"
p.write_text(p.read_text().replace(
"## Project status\n", f"## Project status\n- {today()} — {message}\n", 1))
print("logged")
def chunks():
"""Yield (id, text) per ## section of every state file, file title prepended."""
for d in STATE_DIRS:
for p in sorted((ROOT / d).glob("*.md")):
if p.name == "_template.md":
continue
parts = re.split(r"^(?=## )", p.read_text(), flags=re.M)
for part in parts:
if not part.strip():
continue
m = re.match(r"## (.+)", part)
heading = m.group(1).strip() if m else "top"
yield f"{rel(p)}::{heading}", f"{p.stem} / {heading}\n{part.strip()}"
CODE_EXTS = {".py", ".js", ".ts", ".jsx", ".tsx", ".svelte", ".go", ".rs", ".c", ".h",
".cpp", ".java", ".rb", ".sh", ".sql", ".css", ".html", ".toml", ".yaml", ".yml"}
SKIP_DIRS = {"node_modules", "__pycache__", "dist", "build", "target", "vendor"}
def code_chunks(window=40):
"""Yield (id, text) per ~40-line block of every source file in the repo."""
# ponytail: fixed-line windows, not syntax-aware chunks; upgrade per-language if recall disappoints
for p in sorted(ROOT.rglob("*")):
parts = p.relative_to(ROOT).parts
if (not p.is_file() or p.suffix not in CODE_EXTS
or any(s.startswith(".") or s in SKIP_DIRS for s in parts[:-1])
or p.stat().st_size > 200_000):
continue
lines = p.read_text(errors="replace").splitlines()
for i in range(0, len(lines), window):
block = "\n".join(lines[i:i + window]).strip()
if block:
yield f"{rel(p)}:L{i + 1}", f"{rel(p)}\n{block}"
def _post(url, payload):
req = urllib.request.Request(url, json.dumps(payload).encode(),
{"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=30) as r:
return json.load(r)
def embed(texts):
vecs = []
for i in range(0, len(texts), 4): # TEI max_client_batch_size=4
vecs += _post(EMBED_URL + "/embed", {"inputs": texts[i:i + 4]})
return vecs
def cosine(a, b):
dot = sum(x * y for x, y in zip(a, b))
return dot / ((math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(x * x for x in b))) or 1)
def build_index(texts):
"""Return {chunk_id: vector}, re-embedding only changed chunks. Cached on disk."""
cache_path = ROOT / ".codebase_index/embeddings.json"
cache = json.loads(cache_path.read_text()) if cache_path.exists() else {}
out, todo = {}, []
for cid, text in texts.items():
h = hashlib.sha1(text.encode()).hexdigest()
if cid in cache and cache[cid]["hash"] == h:
out[cid] = cache[cid]
else:
todo.append((cid, h, text))
if todo:
for (cid, h, _), vec in zip(todo, embed([t for _, _, t in todo])):
out[cid] = {"hash": h, "vec": vec}
cache_path.write_text(json.dumps(out))
return {cid: rec["vec"] for cid, rec in out.items()}
def search(query, n=5):
texts = dict(chunks())
texts.update(code_chunks())
try:
index = build_index(texts)
qvec = embed(["Represent this sentence for searching relevant passages: " + query])[0]
top = sorted(index, key=lambda cid: cosine(qvec, index[cid]), reverse=True)[:12]
ranked = _post(RERANK_URL + "/rerank",
{"query": query, "texts": [texts[cid][:1500] for cid in top]})
except urllib.error.URLError as e:
sys.exit(f"embedding/rerank server unreachable ({e}) — are 8081/8082 up?")
for r in sorted(ranked, key=lambda r: -r["score"])[:n]:
print(f"{r['score']:.3f} {top[r['index']]}")
def main(argv=None):
ap = argparse.ArgumentParser(prog="pm", description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
n = sub.add_parser("new")
n.add_argument("kind", choices=("task", "epic", "doc"))
n.add_argument("slug")
n.add_argument("--epic")
s = sub.add_parser("status")
s.add_argument("slug")
s.add_argument("value")
sub.add_parser("list")
sub.add_parser("sync")
sub.add_parser("validate")
l = sub.add_parser("log")
l.add_argument("message")
q = sub.add_parser("search")
q.add_argument("query")
q.add_argument("-n", type=int, default=5)
a = ap.parse_args(argv)
if a.cmd == "new":
new(a.kind, a.slug, a.epic)
elif a.cmd == "status":
set_status(a.slug, a.value)
elif a.cmd == "list":
list_cmd()
elif a.cmd == "sync":
sync()
print("synced")
elif a.cmd == "validate":
errs = validate()
for e in errs:
print(e, file=sys.stderr)
print("ok" if not errs else f"{len(errs)} problem(s)")
sys.exit(1 if errs else 0)
elif a.cmd == "log":
log(a.message)
elif a.cmd == "search":
search(a.query, a.n)
if __name__ == "__main__":
main()