"""TASKS.md as data: parse into sections and items, change, render back. Pure text in, text out; no file access. Format: docs/design.md (section 3). """ from __future__ import annotations import difflib import re from dataclasses import dataclass, field SECTIONS = { "awaiting": "Awaiting your decision", "pending": "Pending", "human": "Needs human", "deferred": "Deferred", } TASK_KEYS = ("pending", "human", "deferred") EFFORTS = ("<1h", "1h", "5h", "10h", "100h") INDENT = " " ID_RE = re.compile(r"^[ta]-[a-z0-9]+(?:-[a-z0-9]+)*$") LINK_RE = re.compile(r"\[\[([^\]|#]+)\]\]") HEADER_RE = re.compile( r"^- \*\*(?P[^*\s]+)\*\*" r"(?: \[P(?P\d)\])?" r"(?: \((?!in progress:|blocked:)(?P[^)]*)\))?" r"(?: \((?P(?:in progress|blocked):[^)]*)\))?" r": (?P.*)$" ) ITEM_START = "- **" AFTER_RE = re.compile(r"^\s*(?:-\s+)?After:\s*(.*)$") REF_RE = re.compile(r"^\s*(?:-\s+)?Ref:\s*(.*)$") SLICES_RE = re.compile(r"^\s*-\s+Slices:\s*(.*)$") MODEL_RE = re.compile(r"^\s*(?:-\s+)?Model:\s*(.*)$") MODELS = ("haiku", "sonnet", "opus") # low → high; no Model line = opus DONE_RE = re.compile(r"^\s*(?:-\s+)?Done:\s*(.*)$") HUMAN_DONE_RE = re.compile( r"\bowner(?:'s)?\s+(?:approves?|approval|tests?|checks?|runs?|says?|confirms?|verif(?:y|ies)|decides?|reviews?|signs?|sees?|ok)\b" r"|\breport to\b|\bconfirm with\b|\bplease confirm\b|^\s*confirm\b", re.I) SESSIONS_RE = re.compile(r"^\s*(?:-\s+)?Sessions:\s*(.*)$") SESSIONS = ("parallel", "solo", "owner") # no Sessions line = parallel CLOUD_RE = re.compile(r"^\s*(?:-\s+)?Cloud:\s*(.*)$") CLOUDS = ("yes", "no") # no Cloud line = decided by the fit rules (wflib/cloud.py) LEADING_ID_RE = re.compile(r"^([ta]-[a-z0-9]+(?:-[a-z0-9]+)*):\s+(.*)$") class TaskError(ValueError): pass @dataclass class Item: id: str prio: int | None = None effort: str | None = None interactive: bool = False status: str | None = None text: str = "" body: list[str] = field(default_factory=list) error: str | None = None raw: str | None = None line: int | None = None # 1-based line of the header in the parsed text def header(self) -> str: if self.error: return self.raw or "" out = f"- **{self.id}**" if self.prio is not None: out += f" [P{self.prio}]" if self.effort is not None: out += f" ({self.effort}{', interactive' if self.interactive else ''})" if self.status: out += f" ({self.status})" return f"{out}: {self.text}" def lines(self) -> list[str]: return [self.header(), *self.body] @property def title(self) -> str: return self.text.split(". ", 1)[0].rstrip(".").strip() @property def goal(self) -> str: return self.text.split(". ", 1)[1].strip() if ". " in self.text else "" @property def blocked_on(self) -> str | None: if self.status and self.status.startswith("blocked:"): found = LINK_RE.findall(self.status) return found[0] if found else None return None def _line(self, pattern: re.Pattern) -> int | None: return next((i for i, l in enumerate(self.body) if pattern.match(l)), None) @property def after(self) -> list[str]: i = self._line(AFTER_RE) return LINK_RE.findall(self.body[i]) if i is not None else [] @property def slices(self) -> list[str]: i = self._line(SLICES_RE) return LINK_RE.findall(self.body[i]) if i is not None else [] @property def model(self) -> str: i = self._line(MODEL_RE) words = MODEL_RE.match(self.body[i]).group(1).split() if i is not None else [] return words[0] if words else "opus" @property def done_text(self) -> str | None: """Text of the Done line, None without one.""" i = self._line(DONE_RE) return DONE_RE.match(self.body[i]).group(1) if i is not None else None @property def runner_ready(self) -> bool: """Headless worker can finish it: has a Done line, not owner-bound, Done not about the owner/confirming.""" done = self.done_text return bool(done and done.strip()) and self.sessions != "owner" and not HUMAN_DONE_RE.search(done) @property def human_done_match(self) -> str | None: """The Done phrase that makes it not runner-ready (human action), None if none.""" m = HUMAN_DONE_RE.search(self.done_text or "") return m.group(0) if m else None @property def sessions(self) -> str: """parallel | solo (no other live session) | owner (owner present); header ', interactive' = owner.""" i = self._line(SESSIONS_RE) words = SESSIONS_RE.match(self.body[i]).group(1).split() if i is not None else [] if words and words[0] in SESSIONS: return words[0] return "owner" if self.interactive else "parallel" @property def cloud(self) -> str | None: """yes | no | None (no Cloud line, or an unknown word).""" i = self._line(CLOUD_RE) words = CLOUD_RE.match(self.body[i]).group(1).split() if i is not None else [] return words[0] if words and words[0] in CLOUDS else None @property def refs(self) -> list[tuple[str, str | None]]: i = self._line(REF_RE) if i is None: return [] out = [] for part in split_refs(REF_RE.match(self.body[i]).group(1)): target = re.sub(r"\s*\(.*$", "", part).strip() if target: path, _, anchor = target.partition("#") out.append((path, anchor or None)) return out def split_refs(text: str) -> list[str]: """Split on commas that are not inside parentheses.""" parts, depth, cur = [], 0, "" for ch in text: if ch == "(": depth += 1 elif ch == ")": depth = max(0, depth - 1) if ch == "," and depth == 0: parts.append(cur) cur = "" else: cur += ch return [p.strip() for p in parts + [cur] if p.strip()] @dataclass class Section: heading: str key: str | None prefix: list[str] = field(default_factory=list) items: list[Item] = field(default_factory=list) suffix: list[str] = field(default_factory=list) line: int | None = None # 1-based line of the heading suffix_line: int | None = None def lines(self) -> list[str]: out = [f"## {self.heading}", *self.prefix] for item in self.items: out += [*item.lines(), ""] return out + self.suffix @dataclass class Doc: preamble: list[str] sections: list[Section] newline: str = "\n" def section(self, key: str) -> Section: for s in self.sections: if s.key == key: return s raise TaskError(f"no '## {SECTIONS[key]}' section") def all_items(self) -> list[Item]: return [i for s in self.sections for i in s.items] def ids(self) -> set[str]: return {i.id for i in self.all_items() if i.id} def find(self, id: str) -> tuple[Section, int]: for s in self.sections: for n, item in enumerate(s.items): if item.id == id: return s, n raise TaskError(unknown_id(id, self.ids())) def item(self, id: str) -> Item: s, n = self.find(id) return s.items[n] def unknown_id(id: str, known: set[str]) -> str: near = difflib.get_close_matches(id, sorted(known), n=3, cutoff=0.0) return f"unknown id '{id}'" + (f" (nearest: {', '.join(near)})" if near else "") def parse_header(line: str) -> Item: m = HEADER_RE.match(line) if not m: found = re.match(r"^- \*\*([^*]+)\*\*", line) return Item(id=found.group(1).strip() if found else "", raw=line, error="bad header: want '- **id** [Pn] (effort) [(status)]: Title. Goal.'") effort, interactive = m.group("effort"), False if effort is not None: parts = [p.strip() for p in effort.split(",")] interactive = "interactive" in parts[1:] effort = parts[0] status = m.group("status") return Item(id=m.group("id"), prio=int(m.group("prio")) if m.group("prio") else None, effort=effort, interactive=interactive, status=status.strip() if status else None, text=m.group("text").strip()) def _strip_blank_tail(lines: list[str]) -> list[str]: while lines and not lines[-1].strip(): lines.pop() return lines def _parse_section(heading: str, lines: list[str], at: int = 0) -> Section: """`at` = 1-based line number of the heading.""" key = next((k for k, h in SECTIONS.items() if h == heading), None) section = Section(heading, key, line=at) if key is None: section.prefix = lines return section for n, line in enumerate(lines, at + 1): if section.suffix: section.suffix.append(line) elif line.startswith(ITEM_START): section.items.append(parse_header(line)) section.items[-1].line = n elif not section.items: section.prefix.append(line) elif line.strip() and not line[0].isspace(): section.suffix.append(line) section.suffix_line = n else: section.items[-1].body.append(line) for item in section.items: _strip_blank_tail(item.body) return section def parse(text: str) -> Doc: newline = "\r\n" if "\r\n" in text else "\n" lines = _strip_blank_tail(text.replace("\r\n", "\n").split("\n")) starts = [i for i, l in enumerate(lines) if l.startswith("## ")] doc = Doc(lines[: starts[0]] if starts else lines, [], newline) for n, start in enumerate(starts): end = starts[n + 1] if n + 1 < len(starts) else len(lines) doc.sections.append(_parse_section(lines[start][3:].strip(), lines[start + 1:end], start + 1)) return doc def render(doc: Doc) -> str: lines = list(doc.preamble) for s in doc.sections: lines += s.lines() _strip_blank_tail(lines) return doc.newline.join(lines) + doc.newline # ---------------------------------------------------------------- ids def make_id(title: str, taken: set[str], prefix: str = "t-") -> str: words = [w for w in re.sub(r"[^a-z0-9]+", " ", title.lower().encode("ascii", "ignore").decode()).split() if w] slug = "" for w in words: longer = f"{slug}-{w}" if slug else w if len(longer) > 40: break slug = longer if not slug and words: slug = words[0][:40] if not slug: raise TaskError(f"no id can be made from title '{title}': give one with --id") base, n, id = prefix + slug, 2, prefix + slug while id in taken: id = f"{base}-{n}" n += 1 return id def slice_id(parent: str, taken: set[str]) -> str: n = 1 while f"{parent}-{n}" in taken: n += 1 return f"{parent}-{n}" def open_slices(doc: Doc, id: str) -> list[str]: """Open items listed in the Slices: line of `id`, then open `-N` items.""" pattern = re.compile(re.escape(id) + r"-\d+$") open_ = doc.ids() named = [s for s in doc.item(id).slices if s in open_] return named + [i.id for i in doc.all_items() if pattern.match(i.id) and i.id not in named] def parent_of(doc: Doc, id: str) -> str | None: for item in doc.all_items(): if id in item.slices: return item.id m = re.match(r"(.+)-\d+$", id) return m.group(1) if m and m.group(1) in doc.ids() else None def rename(doc: Doc, old: str, new: str, taken: set[str]) -> int: """Change the id of an open item and every [[old]] link; returns the link count. `taken` = ids used in the archive.""" item = doc.item(old) if not ID_RE.match(new): raise TaskError(f"bad id '{new}' (want t-… or a-…, lowercase a-z 0-9 -)") if new[:2] != old[:2]: raise TaskError(f"a rename keeps the kind ({old[:2]}…)") if new in doc.ids(): raise TaskError(f"id '{new}' already exists") if new in taken: raise TaskError(f"id '{new}' already used in the archive (ids are never reused)") link, n = re.compile(r"\[\[" + re.escape(old) + r"\]\]"), 0 for other in doc.all_items(): other.text, k = link.subn(f"[[{new}]]", other.text) n += k if other.status: other.status, k = link.subn(f"[[{new}]]", other.status) n += k for i, line in enumerate(other.body): other.body[i], k = link.subn(f"[[{new}]]", line) n += k item.id = new return n # ---------------------------------------------------------------- placing def _prio(item: Item) -> int: return 9 if item.prio is None else item.prio def place(section: Section, item: Item) -> int: """Insert rule: after the last item of the same or higher priority, and after every item named in the new item's After:.""" at = max((n + 1 for n, other in enumerate(section.items) if _prio(other) <= _prio(item)), default=0) deps = set(item.after) return max([at, *(n + 1 for n, other in enumerate(section.items) if other.id in deps)]) def _check_kind(item: Item, key: str) -> None: if key not in SECTIONS: raise TaskError(f"unknown section '{key}' (want {', '.join(SECTIONS)})") if item.id.startswith("a-") != (key == "awaiting"): raise TaskError("a- items belong in Awaiting, t- items in Pending / Needs human / Deferred") if key != "awaiting" and item.prio is None: raise TaskError(f"task '{item.id}' needs a priority [P0]-[P3]") def insert(doc: Doc, item: Item, key: str) -> None: if item.error: raise TaskError(item.error) if item.id in doc.ids(): raise TaskError(f"id '{item.id}' already exists") _check_kind(item, key) section = doc.section(key) section.items.insert(place(section, item), item) def remove(doc: Doc, id: str) -> Item: section, n = doc.find(id) return section.items.pop(n) def _section_key(doc: Doc, id: str) -> str: return doc.find(id)[0].key def set_prio(doc: Doc, id: str, prio: int) -> None: if prio not in (0, 1, 2, 3): raise TaskError("priority 0-3") key = _section_key(doc, id) item = doc.item(id) if item.id.startswith("a-"): raise TaskError("Awaiting items have no priority") remove(doc, id) item.prio = prio insert(doc, item, key) def move_to(doc: Doc, id: str, key: str) -> None: item = doc.item(id) _check_kind(item, key) doc.section(key) remove(doc, id) insert(doc, item, key) def move_rel(doc: Doc, id: str, other: str, before: bool, force: bool = False) -> None: if id == other: raise TaskError("cannot move an item relative to itself") item = doc.item(id) target, _ = doc.find(other) _check_kind(item, target.key) source, n = doc.find(id) source.items.pop(n) at = next(i for i, o in enumerate(target.items) if o.id == other) + (0 if before else 1) target.items.insert(at, item) def undo(): target.items.pop(at) source.items.insert(n, item) later = {o.id for o in target.items[at + 1:]} for dep in item.after: if dep in later: undo() raise TaskError(f"'{id}' is After: [[{dep}]], it cannot go before it") for o in target.items[:at]: if id in o.after: undo() raise TaskError(f"'{o.id}' is After: [[{id}]], '{id}' cannot go after it") prev = target.items[at - 1] if at else None nxt = target.items[at + 1] if at + 1 < len(target.items) else None if not force and ((prev and _prio(prev) > _prio(item)) or (nxt and _prio(nxt) < _prio(item))): undo() raise TaskError(f"moving '{id}' there breaks priority order (wf prio, or --force)") # ---------------------------------------------------------------- fields def set_status(doc: Doc, id: str, status: str | None) -> None: item = doc.item(id) if item.id.startswith("a-"): raise TaskError("Awaiting items have no status") if status: status = status.strip() if status.startswith("blocked:"): on = LINK_RE.findall(status) open_awaiting = {i.id for s in doc.sections if s.key == "awaiting" for i in s.items} if len(on) != 1 or on[0] not in open_awaiting: raise TaskError(f"'{on[0] if on else status}' is not an open Awaiting item") elif not status.startswith("in progress:"): raise TaskError("status is 'in progress: ' or 'blocked: [[a-id]]'") if ")" in status or "(" in status: raise TaskError("status note cannot contain parentheses (the status line ends in one); use - or :") item.status = status or None def _tail_start(item: Item) -> int: """Index where the trailing Sessions:/Model:/After:/Ref: lines of the body begin.""" n = len(item.body) while n and any(p.match(item.body[n - 1]) for p in (AFTER_RE, REF_RE, MODEL_RE, SESSIONS_RE, CLOUD_RE)): n -= 1 return n def _set_line(item: Item, pattern: re.Pattern, line: str | None, at_end: bool) -> None: i = item._line(pattern) if i is not None: if line is None: item.body.pop(i) else: item.body[i] = line elif line is not None: if at_end: item.body.append(line) else: ref = item._line(REF_RE) item.body.insert(len(item.body) if ref is None else ref, line) def set_fields(doc: Doc, id: str, *, title: str | None = None, effort: str | None = None, interactive: bool | None = None, after: list[str] | None = None, refs: list[str] | None = None, model: str | None = None, sessions: str | None = None, done: str | None = None, cloud: str | None = None) -> None: item = doc.item(id) if done is not None: done = done.strip() if (i := item._line(DONE_RE)) is not None: if done: item.body[i] = INDENT + "Done: " + done else: item.body.pop(i) elif done: item.body.insert(_tail_start(item), INDENT + "Done: " + done) if sessions is not None: if sessions not in ("", *SESSIONS): raise TaskError(f"sessions '{sessions}' (want {', '.join(SESSIONS)})") item.interactive = False line = INDENT + "Sessions: " + sessions if sessions else None if (i := item._line(SESSIONS_RE)) is not None: if line: item.body[i] = line else: item.body.pop(i) elif line: m = item._line(MODEL_RE) item.body.insert(m if m is not None and m >= _tail_start(item) else _tail_start(item), line) if cloud is not None: if cloud not in ("", *CLOUDS): raise TaskError(f"cloud '{cloud}' (want {', '.join(CLOUDS)})") line = INDENT + "Cloud: " + cloud if cloud else None if (i := item._line(CLOUD_RE)) is not None: if line: item.body[i] = line else: item.body.pop(i) elif line: item.body.insert(_tail_start(item), line) if model: if model not in MODELS: raise TaskError(f"model '{model}' (want {', '.join(MODELS)})") line = INDENT + "Model: " + model if (i := item._line(MODEL_RE)) is not None: item.body[i] = line else: item.body.insert(_tail_start(item), line) elif model is not None and (i := item._line(MODEL_RE)) is not None: item.body.pop(i) if title is not None: title = title.strip().rstrip(".") if not title: raise TaskError("empty title") if ". " in title or title.endswith(("?", "!")): item.text = title if title.endswith(("?", "!")) else f"{title}." else: item.text = f"{title}. {item.goal}" if item.goal else f"{title}." if effort is not None: if effort not in EFFORTS: raise TaskError(f"effort '{effort}' (want {', '.join(EFFORTS)})") item.effort = effort if interactive is not None: item.interactive = interactive if after is not None: line = INDENT + "- After: " + ", ".join(f"[[{a}]]" for a in after) if after else None _set_line(item, AFTER_RE, line, at_end=False) if refs is not None: _set_line(item, REF_RE, INDENT + "Ref: " + ", ".join(refs) if refs else None, at_end=True) def add_note(doc: Doc, id: str, line: str) -> None: item = doc.item(id) line = line.strip() if not line: raise TaskError("empty note") item.body.insert(_tail_start(item), f"{INDENT}- {line}") def _link_after(line: str) -> str: """'After: t-a, t-b' -> 'After: [[t-a]], [[t-b]]' (bare ids accepted, written as links).""" m = re.match(r"^(\s*(?:-\s+)?After:\s*)(.*)$", line) if not m: return line parts = [w for w in re.split(r"[\s,;]+", m.group(2)) if w] if not parts or not all(LINK_RE.fullmatch(w) or ID_RE.match(w) for w in parts): return line return m.group(1) + ", ".join(w if w.startswith("[[") else f"[[{w}]]" for w in parts) def indent_body(lines: list[str]) -> list[str]: import textwrap text = textwrap.dedent("\n".join(_link_after(l.rstrip()) for l in lines)) return _strip_blank_tail([INDENT + l if l.strip() else "" for l in text.split("\n")]) if text.strip() else [] def set_body(doc: Doc, id: str, lines: list[str]) -> None: item = doc.item(id) new = indent_body(lines) kept = [l for l in item.body[_tail_start(item):] if not any(p.match(l) and any(p.match(n) for n in new) for p in (AFTER_RE, REF_RE))] item.body = new + kept BOX_RE = re.compile(r"^(\s*- \[)([ xX])(\] )(.*)$") def tick(doc: Doc, id: str, which: str) -> str: item = doc.item(id) boxes = [(i, BOX_RE.match(l)) for i, l in enumerate(item.body) if BOX_RE.match(l)] if which.isdigit(): if not 1 <= int(which) <= len(boxes): raise TaskError(f"no box {which} in '{id}' ({len(boxes)} boxes)") hits = [boxes[int(which) - 1]] else: hits = [(i, m) for i, m in boxes if which.lower() in m.group(4).lower()] if not hits: raise TaskError(f"no box matches '{which}' in '{id}'") if len(hits) > 1: raise TaskError(f"'{which}' matches {len(hits)} boxes in '{id}'") i, m = hits[0] if m.group(2) != " ": raise TaskError(f"box already ticked: {m.group(4)}") item.body[i] = f"{m.group(1)}x{m.group(3)}{m.group(4)}" return m.group(4) def unblock(doc: Doc, a_id: str) -> list[str]: out = [] for item in doc.all_items(): if item.blocked_on == a_id: item.status = None out.append(item.id) return out # ---------------------------------------------------------------- pick def why_not(doc: Doc, item: Item, archived: set[str], held: dict[str, str] | None = None, others: int = 0, owner: bool = True) -> str | None: """Why a Pending item can't be picked now; None = pickable. held: id → holder of a live claim by another session; others: other live sessions in the project; owner: owner present.""" if item.error: return "bad header" if held and item.id in held: return f"in progress by {held[item.id]}" if item.sessions == "owner" and not owner: return "owner: needs the owner (wf next --owner)" if item.sessions == "solo" and others: return f"solo: {others} other live session{'s' if others > 1 else ''}" if item.status and item.status.startswith("blocked:"): return f"blocked: {item.blocked_on}" if open_slices(doc, item.id): return "open slices: " + ", ".join(open_slices(doc, item.id)) if [a for a in item.after if a not in archived]: return "after: " + ", ".join(a for a in item.after if a not in archived) return None def waiters(doc: Doc) -> dict[str, set[str]]: """Open task id → every open task (Pending, Needs human) that waits on it, transitively: via After:, and a parent waits on its open slices.""" items = {i.id: i for s in doc.sections if s.key in ("pending", "human") for i in s.items if not i.error} direct: dict[str, set[str]] = {id: set() for id in items} for item in items.values(): for dep in item.after: if dep in direct: direct[dep].add(item.id) for s in item.slices: if s in direct: direct[s].add(item.id) out: dict[str, set[str]] = {} for id in direct: seen, todo = set(), list(direct[id]) while todo: w = todo.pop() if w not in seen and w != id: seen.add(w) todo += direct[w] out[id] = seen return out def solo_running(doc: Doc, held: dict[str, str]) -> tuple[str, str] | None: """(id, holder) of a solo task in progress by another live session; held: see why_not.""" return next(((i.id, held[i.id]) for i in doc.section("pending").items if i.id in held and not i.error and i.sessions == "solo"), None) # ---------------------------------------------------------------- archive ARCHIVE_ID_RE = re.compile(r"^- .*?\*\*([ta]-[a-z0-9-]+)\*\*") def archive_ids(text: str) -> set[str]: return {m.group(1) for l in text.replace("\r\n", "\n").split("\n") if (m := ARCHIVE_ID_RE.match(l))} def archive_line(date: str, item: Item, entry: str) -> str: entry = " ".join(entry.split()) return f"- {date} **{item.id}** {item.title}" + (f" — {entry}" if entry else "") def archive_prepend(text: str, line: str) -> str: newline = "\r\n" if "\r\n" in text else "\n" lines = _strip_blank_tail(text.replace("\r\n", "\n").split("\n")) at = next((i for i, l in enumerate(lines) if l.startswith("- ")), None) if at is None: lines += ["", line] else: lines.insert(at, line) return newline.join(lines) + newline # ---------------------------------------------------------------- blocks def parse_block(text: str) -> Item: lines = _strip_blank_tail(text.replace("\r\n", "\n").strip("\n").split("\n")) if not lines or not lines[0].strip(): raise TaskError("empty item") first = lines[0].strip() if re.match(r"^\d+\. ", first): raise TaskError("old numbered format: write '- **id** [Pn] (effort): Title. Goal.'") item = parse_header(first) if item.error: raise TaskError(item.error) item.body = indent_body(lines[1:]) return item