ai-agent/backend/goal.py

"""GOAL.md — a directory's long-term direction file, and its checklist progress.

A project or service may keep a ``GOAL.md`` at its root: the north star the
scheduled goal-keeper agent (see ``cron.py``) reads every few hours to push work
forward. Its "what's left" list is a markdown task list, so the file doubles as a
progress meter — ``- [x]`` done vs. ``- [ ]`` open.

Both catalogs (``projects.py`` / ``svc.py``) surface it the same way: a
``{done, total}`` rollup on the summary (the card's ``x/n`` tag) and the full
markdown on the detail (the page's Goal section). A directory with no GOAL.md
simply has ``goal: None`` — like every catalog here, absence is a valid state.

Two further structures are mined out of the same file for the Goals page
(``/goals``), and both are *conventions the `goal` skill writes*, not new files:

**Milestones.** A ``## Horizons`` section's ``###`` subheadings name versions —
``### Short term — v0.2: read + send (now)``. Each becomes a milestone whose
progress is the checklist items under it, so a goal reads as a versioned
roadmap rather than one flat bar.

**Claims.** A ``## Being worked on`` section is the goal-keeper's mutual
exclusion: before starting an item an agent appends a claim line there, and
removes it when it lands. Two agents reading the same GOAL.md therefore pick
different items. A claim is one bullet::

    - [ai-agent] Bundle zipgo so "build me a site" ends in a hosted URL — @2026-07-14T09:12Z (conv abc123)

Only the free text matters to the parser; the trailing ``@<iso>`` (if present)
ages the claim so the UI can flag a stale one whose agent died.
"""
from __future__ import annotations

import re
import textwrap
from datetime import datetime, timezone
from pathlib import Path

GOAL_FILE = "GOAL.md"

MAX_GOAL = 40_000  # cap the payload so a detail response stays bounded

# How many checklist items a *board* card carries. The board renders every
# goal at once, so it can't ship every item of every one; this is enough to
# expand a card and act on what's next, and the detail page has the full list.
BOARD_ITEMS = 12

# A claim older than this is almost certainly an agent that died mid-item
# rather than one still working — the goal-keeper runs every 5h, and no single
# session outlives that by much. Surfaced as `stale`, never auto-removed: only
# an agent that reads the file knows whether the work actually landed.
STALE_CLAIM_SECS = 6 * 3600

# A markdown task-list item: optional indent, a bullet, then [ ] / [x].
# Matches "- [x] done", "  * [ ] open", "+ [X] done" — but not "[x]" mid-prose.
_TASK_RE = re.compile(r'^[ \t]*[-*+][ \t]+\[([ xX])\]', re.MULTILINE)

# The start of a checklist item: the bullet, its checkbox, and the rest of that
# first line. The item's *full* text usually continues onto following indented
# lines, so this only locates where an item begins — `parse_items` folds the
# continuation in (see `_ITEM_CONT_RE`). Trailing markdown emphasis is left
# intact: the frontend renders it.
_TASK_ITEM_RE = re.compile(r'^[ \t]*[-*+][ \t]+\[([ xX])\][ \t]*(.*?)[ \t]*$', re.MULTILINE)

# What ends a multi-line item: the next top-level bullet (checkbox or not) or a
# heading. Anything else — indented prose, nested bullets, blank lines — is
# continuation belonging to the item above, and is folded into its text.
_ITEM_END_RE = re.compile(r'^(?:[ \t]{0,1}[-*+][ \t]|#{1,6}[ \t])')

# A GOAL.md keeps its checklist text as multi-line bullets — the first line
# carries the checkbox and a bold lead, continuation lines are indented prose.
# For a Work prompt we want a short, single-line handle, so we take the first
# line and strip markdown emphasis/inline-code to a plain summary.
_EMPHASIS_RE = re.compile(r'(\*\*|__|\*|_|`)')

# The end of the lead sentence: ".", "!" or "?" followed by a space (or the end
# of the text). A version like "v0.2" has no space after its dot, so it survives.
_SENTENCE_RE = re.compile(r'[.!?](?=\s|$)')

# A bold lead opening the item — `**Fold the sidecar in.**` / `__Ship it.__`.
_BOLD_LEAD_RE = re.compile(r'^(?:\*\*|__)(.+?)(?:\*\*|__)')

# A bracket tag opening the item — `**[human]**`, `[blocked]`, `[human]`. It
# says what the item *needs*, not what it is, so a summary drops it. Only a
# short, single-word-ish tag: `[x]`-style prose in brackets is left alone.
_TAG_PREFIX_RE = re.compile(r'^(?:\*\*|__)?\[[^\]]{1,20}\](?:\*\*|__)?[ \t:—-]*')

# Guard rails for the lead: a "sentence" shorter than this is an abbreviation,
# not a summary, so we keep reading; past _MAX_LEAD we hard-cut with an ellipsis.
_MIN_LEAD = 24
_MAX_LEAD = 120

# Fenced code blocks are stripped before counting: a ``- [ ]`` inside a snippet
# (e.g. a README example) is sample text, not this file's own checklist.
_FENCE_RE = re.compile(r'^[ \t]*(```|~~~).*?^[ \t]*\1[ \t]*$', re.MULTILINE | re.DOTALL)

# An ATX heading: capture its level and its text.
_HEADING_RE = re.compile(r'^(#{1,6})[ \t]+(.+?)[ \t]*#*[ \t]*$', re.MULTILINE)

# The section whose ### subheadings are the version milestones.
_HORIZONS_RE = re.compile(r'^horizons?\b', re.IGNORECASE)

# The section whose ### subheadings group the actionable checkboxes by theme.
_WISHLIST_RE = re.compile(r'^wish\s*list\b', re.IGNORECASE)

# The section holding the goal-keeper's in-flight claims.
_WORKING_RE = re.compile(r'^being worked on\b', re.IGNORECASE)

# A version inside a milestone heading: "v0.3", "v1.0 / someday", "1.2.0".
_VERSION_RE = re.compile(r'\bv?(\d+\.\d+(?:\.\d+)?)\b')

# "(now)" marks the horizon currently being worked toward.
_NOW_RE = re.compile(r'\(\s*now\s*\)', re.IGNORECASE)

# A claim bullet, with an optional trailing "@<iso timestamp>".
_CLAIM_RE = re.compile(r'^[ \t]*[-*+][ \t]+(?:\[[ xX]\][ \t]+)?(.+?)[ \t]*$', re.MULTILINE)
_CLAIM_AT_RE = re.compile(r'@(\d{4}-\d{2}-\d{2}[T ][\d:]+(?:Z|[+-]\d{2}:?\d{2})?)')


def _strip_fences(text: str) -> str:
    return _FENCE_RE.sub("", text)


def count_tasks(text: str) -> tuple[int, int]:
    """(done, total) markdown checklist items in a GOAL.md body."""
    marks = _TASK_RE.findall(_strip_fences(text))
    done = sum(1 for m in marks if m in ("x", "X"))
    return done, len(marks)


def _plain(text: str) -> str:
    """A single-line, emphasis-stripped handle for a checklist item.

    A wishlist item is usually ``**Bold lead.** more prose…`` hard-wrapped over
    several lines, so the *physical* first line is a poor summary — it breaks
    mid-sentence ("The Claude Code CLI is a"). We instead unwrap the item to one
    line and cut at its first sentence: the bold lead when there is one, which is
    exactly the handle these items are written to carry. Markers ``*``/``_``/`` `
    `` are dropped so the Work prompt and tooltip read as plain text; the
    frontend still renders the original markdown for display.
    """
    flat = " ".join(text.split())

    # A leading bracket tag (`**[human]**`, `[blocked]`) marks *who or what* an
    # item needs, not what it is — moooo's GOAL.md opens several items that way.
    # Drop it before looking for the lead, or the summary reads "[human]" and
    # the actual task is lost.
    flat = _TAG_PREFIX_RE.sub("", flat).strip()

    # A bold lead (`**Fold the sidecar in.**`) is the handle the item was written
    # to carry, so prefer it — and read it *before* stripping emphasis, since
    # stripping is what would erase the boundary.
    # No length floor here: bolding it *is* the author saying "this is the
    # handle", which is a stronger signal than any heuristic we'd apply.
    bold = _BOLD_LEAD_RE.match(flat)
    if bold:
        lead = _EMPHASIS_RE.sub("", bold.group(1)).strip()
        if lead:
            return _clip(lead)

    plain = _EMPHASIS_RE.sub("", flat).strip()
    # Otherwise the first sentence — but only when it's a real lead rather than
    # an abbreviation or a stray "v0.2.", so require some substance behind it.
    m = _SENTENCE_RE.search(plain)
    if m and _MIN_LEAD <= m.end() <= _MAX_LEAD:
        return plain[:m.end()].strip()
    return _clip(plain)


def _clip(s: str) -> str:
    """Hard-cut an over-long lead at a word boundary, with an ellipsis."""
    if len(s) <= _MAX_LEAD:
        return s
    cut = s[:_MAX_LEAD].rstrip()
    sp = cut.rfind(" ")
    return (cut[:sp] if sp > _MAX_LEAD // 2 else cut).rstrip(" ,;:—-") + "…"


def _fold_continuations(body: str, start_re: re.Pattern) -> list[tuple[re.Match, str]]:
    """Pair each item-start match in ``body`` with its **full** multi-line text.

    A GOAL.md keeps its items as multi-line bullets: the checkbox line carries a
    bold lead and the substance continues on following indented lines. Matching
    only the first line silently drops that substance — which is what made items
    read as truncated everywhere they were surfaced. So we walk the matches and
    extend each one's text to the next *top-level* bullet or heading
    (``_ITEM_END_RE``), keeping the indented prose in between.

    Returns ``(match, full_text)`` pairs in document order; ``full_text`` is the
    item's first-line text plus its continuation, dedented and stripped.
    """
    out: list[tuple[re.Match, str]] = []
    matches = list(start_re.finditer(body))
    for i, m in enumerate(matches):
        limit = matches[i + 1].start() if i + 1 < len(matches) else len(body)
        # Everything between this item's first line and the next item, cut short
        # at the first line that starts a new top-level block.
        tail_lines: list[str] = []
        for line in body[m.end():limit].splitlines():
            if line.strip() and _ITEM_END_RE.match(line):
                break
            tail_lines.append(line)
        first = m.group(m.re.groups).strip()
        tail = textwrap.dedent("\n".join(tail_lines)).strip()
        out.append((m, f"{first}\n{tail}".strip() if tail else first))
    return out


def parse_items(text: str) -> list[dict]:
    """Every checklist item in a GOAL.md body, in document order.

    Each is ``{text, plain, checked}``: ``text`` is the raw markdown of the whole
    item — its checkbox line *and* the indented prose continuing it, which is
    what the widget renders when you expand a row — ``plain`` a single-line
    emphasis-stripped summary (what the Work button sends the agent and shows as
    a tooltip), and ``checked`` whether it's ticked. Fenced code blocks are
    stripped first, so a ``- [ ]`` inside a snippet is not mistaken for a real
    item — same rule as ``count_tasks``, so the item list and the ``done/total``
    rollup always agree.
    """
    out: list[dict] = []
    for m, full in _fold_continuations(_strip_fences(text), _TASK_ITEM_RE):
        out.append({
            "text": full,
            "plain": _plain(full),
            "checked": m.group(1) in ("x", "X"),
        })
    return out


def _sections(text: str) -> list[dict]:
    """Flatten a markdown body into its headings and their bodies.

    Each entry is ``{level, title, body}`` where ``body`` is everything up to
    the next heading of *any* level — so a ``##`` section's body excludes its
    own ``###`` children, and each child is its own entry. That's what lets a
    milestone count only the checklist items written under it.
    """
    out: list[dict] = []
    heads = list(_HEADING_RE.finditer(text))
    for i, h in enumerate(heads):
        end = heads[i + 1].start() if i + 1 < len(heads) else len(text)
        out.append({"level": len(h.group(1)), "title": h.group(2).strip(),
                    "body": text[h.end():end]})
    return out


def _subsections_of(secs: list[dict], match: re.Pattern) -> list[dict]:
    """The ``###`` children of the first ``##`` section whose title matches."""
    out, inside = [], False
    for s in secs:
        if s["level"] <= 2:
            # We're inside the target section only while the last ## we saw was it.
            inside = s["level"] == 2 and bool(match.match(s["title"]))
            continue
        if inside and s["level"] == 3:
            out.append(s)
    return out


def _parse_milestones(secs: list[dict]) -> list[dict]:
    """The roadmap's stages — the ``###`` subheadings under ``## Horizons``.

    A horizon names a version ("### Short term — v0.2: read + send (now)") and
    is the unit the Goals page draws as a milestone marker.

    Its progress is the checklist written under it — but in practice a GOAL.md
    keeps its horizons as *prose* and puts every checkbox in ``## Wishlist``,
    grouped by theme rather than by version. So when a horizon carries no
    checklist of its own we leave it at 0/0 rather than inventing an
    attribution: the milestone is still a real stage on the bar (the page
    positions it against the goal's overall progress), it simply has no
    checklist to call its own. ``hasTasks`` says which of the two a milestone
    is, so the UI never renders a misleading "0/0 done".
    """
    out: list[dict] = []
    for s in _subsections_of(secs, _HORIZONS_RE):
        title = s["title"]
        done, total = count_tasks(s["body"])
        ver = _VERSION_RE.search(title)
        out.append({
            "title": title,
            "version": ver.group(1) if ver else None,
            "current": bool(_NOW_RE.search(title)),
            "done": done,
            "total": total,
            "hasTasks": total > 0,
        })
    return out


def _parse_themes(secs: list[dict]) -> list[dict]:
    """The wishlist's themed groups — the ``###`` subheadings under ``## Wishlist``.

    This is where the actionable checkboxes actually live ("### Reliability",
    "### Frontend split"), so it's the breakdown that tells you *what* is left,
    complementing the milestones' *when*. A flat wishlist with no subheadings
    yields ``[]`` and the page just shows the overall bar.
    """
    out: list[dict] = []
    for s in _subsections_of(secs, _WISHLIST_RE):
        done, total = count_tasks(s["body"])
        if total:
            out.append({"title": s["title"], "done": done, "total": total})
    return out


def _parse_claims(secs: list[dict], now: datetime) -> list[dict]:
    """In-flight goal-keeper claims — the bullets under ``## Being worked on``.

    An empty section (the steady state — every claim removed as its item
    landed) yields ``[]``, which is exactly what it means: nobody is working
    on this goal right now.
    """
    out: list[dict] = []
    for s in secs:
        if s["level"] != 2 or not _WORKING_RE.match(s["title"]):
            continue
        for _m, raw in _fold_continuations(_strip_fences(s["body"]), _CLAIM_RE):
            # Folded so a wrapped claim keeps all its words (the item summary it
            # embeds is what matches it to a checklist row), but flattened to one
            # line — a claim is a single record and the UI renders it inline.
            text = " ".join(raw.split())
            if not text:
                continue
            at = _CLAIM_AT_RE.search(text)
            since, stale = None, False
            if at:
                try:
                    ts = datetime.fromisoformat(at.group(1).replace("Z", "+00:00"))
                    if ts.tzinfo is None:
                        ts = ts.replace(tzinfo=timezone.utc)
                    since = ts.isoformat()
                    stale = (now - ts).total_seconds() > STALE_CLAIM_SECS
                except ValueError:
                    pass
            out.append({"text": text, "since": since, "stale": stale})
    return out


def parse(text: str, now: datetime | None = None) -> dict:
    """The full parse of a GOAL.md body: progress, milestones, themes, claims."""
    body = _strip_fences(text)
    secs = _sections(body)
    done, total = count_tasks(text)
    return {
        "done": done,
        "total": total,
        "milestones": _parse_milestones(secs),
        "themes": _parse_themes(secs),
        "claims": _parse_claims(secs, now or datetime.now(timezone.utc)),
    }


def _read(root: Path) -> str | None:
    p = root / GOAL_FILE
    if not p.is_file():
        return None
    try:
        text = p.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return None
    if len(text) > MAX_GOAL:
        text = text[:MAX_GOAL] + f"\n\n… [{len(text) - MAX_GOAL} more chars truncated]"
    return text


def goal_summary(root: Path) -> dict | None:
    """{done, total} for a directory's GOAL.md, or None when it has none."""
    text = _read(root)
    if text is None:
        return None
    done, total = count_tasks(text)
    return {"done": done, "total": total}


def goal_detail(root: Path, now: datetime | None = None) -> dict | None:
    """{done, total, content, items, claims} — the summary, full markdown, the
    flat checklist the detail page's checklist widget renders (each item with a
    Work button), and the in-flight ``## Being worked on`` claims so the widget
    can flag an item another agent is already on as loading."""
    text = _read(root)
    if text is None:
        return None
    done, total = count_tasks(text)
    secs = _sections(_strip_fences(text))
    return {
        "done": done,
        "total": total,
        "content": text,
        "items": parse_items(text),
        "claims": _parse_claims(secs, now or datetime.now(timezone.utc)),
    }


def goal_board(root: Path) -> dict | None:
    """What the Goals page needs: progress + milestones + claims + checklist.

    The full markdown is deliberately *not* included — the board renders many
    goals at once, and the detail pages already serve the file itself. The
    checklist *is*, so a card can expand into its individual items and put a
    Work button on each without leaving the board; it's capped at
    ``BOARD_ITEMS`` (open items first, since those are the ones you'd act on)
    to keep the payload bounded, with the overflow reported as a count.
    """
    text = _read(root)
    if text is None:
        return None
    out = parse(text)
    items = parse_items(text)
    # Open items lead: the card hides done ones by default, and a goal with 40
    # ticked boxes shouldn't spend its whole budget on them.
    ordered = [i for i in items if not i["checked"]] + [i for i in items if i["checked"]]
    out["items"] = ordered[:BOARD_ITEMS]
    out["itemsTruncated"] = max(0, len(ordered) - BOARD_ITEMS)
    return out