marketplace/skills/issue/scripts/issue.py

#!/usr/bin/env python3
r"""
issue.py — what an issue IS. The domain layer.

Not a command; the module every other issue script builds on. It knows the
canonical markdown format, the label taxonomy, validation, and the dependency
graph. It knows NOTHING about any tracker: no Gitea, no `tea`, no logins, no HTTP, no
issue numbers. The layering rule is mechanically checkable — every import in
this directory is stdlib, and `subprocess` is not among them:

    grep -rhn '^import\|^from' skills/issue/scripts/ | sort -u

Delete skills/sync/ entirely and this layer keeps working: issues that live
only on this machine are first-class, not drafts on their way somewhere.

Identity is a slug derived from the title, and it is the only identity the
domain has. The file name is the id:

    tmp/issues/wire-sqlc-appclick.md

    ---
    id: wire-sqlc-appclick
    state: open
    labels: [type/task, tech/sql]
    assignees: [naudachu]
    milestone: v0.2
    depends: [migrate-schema]
    origin: gitea
    gitea: owner/repo#42
    synced: 2026-08-07T18:40:00Z
    ---
    # Wire sqlc into the appclick repo layer

    ## Summary
    ...

Keys above `origin:` are owned here. Everything below is written by the sync
layer; this module carries those keys through load/save verbatim and never
reads them. That passthrough is what lets one file represent both a local
issue and a synced one without the domain learning a second vocabulary.

Every metadata field is one line and lists are inline, so plain grep works
without a parser:

    grep -l 'labels:.*type/bug' tmp/issues/*.md
    grep -ln 'depends:.*migrate-schema' tmp/issues/*.md   # who depends on it
"""
import collections
import os
import re

ISSUE_ROOT = os.path.join("tmp", "issues")

# Domain-owned metadata, in render order. Foreign keys render after these,
# sorted, so the sync layer can add fields without touching this list.
DOMAIN_KEYS = ["id", "state", "labels", "assignees", "milestone", "depends", "origin"]
LIST_KEYS = {"labels", "assignees", "depends"}
STATES = ("open", "closed")

# `origin` is "does this issue exist anywhere but here" — a fact about the
# work, so it is owned here. Its value is `local` or a tracker's name; what
# that name means, and the handle that goes with it (`gitea: owner/repo#42`),
# stay foreign keys this layer carries but never reads.
LOCAL = "local"

# type/* is mandatory and exclusive; severity/* is optional and exclusive;
# tech/* and comp/* are free-form. Colors are NOT here — a hex code is how
# Gitea paints a chip, which makes it the sync layer's business.
TYPES = {
    "bug": "Something behaves incorrectly in existing code",
    "task": "Implementation of new functionality",
    "refactor": "Internal restructuring; behavior must not change",
    "test": "Writing or fixing tests",
    "feature": "Container: several issues delivering one unit of business value",
    "draft": "Idea captured for later; not ready for work",
}
SEVERITIES = ("low", "medium", "high", "showstopper", "critical")
EXCLUSIVE_NS = ("type/", "severity/")

# Sections every type must carry. type/draft is exempt from acceptance criteria.
REQUIRED_SECTIONS = ["## Summary", "## Spec"]
AC_SECTION = "## Acceptance criteria"
DEPENDS_SECTION = "## Depends on"
# Per-type sections from the templates — absence is a warning, not a stop.
EXPECTED_SECTIONS = {
    "bug": ["## Steps to reproduce", "## Expected", "## Actual", "## Environment"],
    "task": ["## Motivation"],
    "refactor": ["## Motivation", "## Invariants"],
    "test": ["## Motivation", "## Test cases"],
    "feature": ["## Motivation", "## Issues"],
    "draft": ["## Notes"],
}

TITLE_PREFIX = re.compile(
    r'^\s*(\[[^\]]+\]|(fix|feat|feature|bug|task|test|chore|refactor)\s*:)', re.I)
CYRILLIC = re.compile(r'[а-яё]', re.I)
SLUG_OK = re.compile(r'^[a-z0-9]+(-[a-z0-9]+)*$')


# --------------------------------------------------------------------------
# identity
# --------------------------------------------------------------------------

def slugify(text, maxlen=48):
    """Title -> id. Titles are English by format rule, so ASCII is enough;
    anything else is dropped rather than transliterated."""
    s = re.sub(r'[^a-z0-9]+', '-', (text or "").lower()).strip("-")
    if len(s) > maxlen:
        s = s[:maxlen].rsplit("-", 1)[0] or s[:maxlen]
    return s.strip("-") or "issue"


def unique_id(root, base, taken=()):
    """`base`, or base-2, base-3… when the slug is already used."""
    used = set(taken) | set(all_ids(root))
    if base not in used:
        return base
    for i in range(2, 1000):
        cand = "%s-%d" % (base, i)
        if cand not in used:
            return cand
    raise ValueError("cannot allocate an id for %r" % base)


# --------------------------------------------------------------------------
# metadata block
# --------------------------------------------------------------------------

def parse_meta(text):
    """Split a file into (meta, title, body).

    meta values are strings, or lists for the inline `[a, b]` form. title is
    the first `# ` heading below the block and is stripped out of body."""
    meta, rest = {}, text
    if text.startswith("---"):
        end = text.find("\n---", 3)
        if end != -1:
            for line in text[3:end].strip().splitlines():
                if ":" not in line:
                    continue
                k, v = line.split(":", 1)
                k, v = k.strip(), v.strip()
                if v.startswith("[") and v.endswith("]"):
                    v = [x.strip() for x in v[1:-1].split(",") if x.strip()]
                elif k in LIST_KEYS:
                    v = [x.strip() for x in v.split(",") if x.strip()]
                meta[k] = v
            rest = text[end + 4:]
    rest = rest.lstrip("\n")

    title = ""
    m = re.match(r'^#\s+(.+?)\s*\n', rest)
    if m:
        title = m.group(1).strip()
        rest = rest[m.end():].lstrip("\n")
    return meta, title, rest


def render_meta(meta):
    """Domain keys in DOMAIN_KEYS order, foreign keys after them, sorted.
    Lists stay on one line so grep sees them whole."""
    lines = ["---"]
    foreign = sorted(k for k in meta if k not in DOMAIN_KEYS)
    for k in DOMAIN_KEYS + foreign:
        if k not in meta:
            continue
        v = meta[k]
        if isinstance(v, (list, tuple)):
            v = "[%s]" % ", ".join(str(x) for x in v)
        lines.append("%s: %s" % (k, v))
    lines.append("---")
    return "\n".join(lines)


# --------------------------------------------------------------------------
# the issue
# --------------------------------------------------------------------------

class Issue(object):
    """One unit of work. `extra` holds metadata this layer does not own."""

    def __init__(self, id="", title="", body="", state="open", labels=None,
                 assignees=None, milestone="", depends=None, origin=LOCAL,
                 extra=None):
        self.id = id
        self.title = title
        self.body = body
        self.state = state or "open"
        self.labels = list(labels or [])
        self.assignees = list(assignees or [])
        self.milestone = milestone or ""
        self.depends = list(depends or [])
        self.origin = origin or LOCAL
        self.extra = dict(extra or {})

    @property
    def is_local(self):
        """True while this issue exists nowhere but here — a durable state,
        not a pending one."""
        return self.origin == LOCAL

    # -- taxonomy views ----------------------------------------------------

    @property
    def type(self):
        for l in self.labels:
            if l.startswith("type/"):
                return l.split("/", 1)[1]
        return ""

    @property
    def severity(self):
        for l in self.labels:
            if l.startswith("severity/"):
                return l.split("/", 1)[1]
        return ""

    # -- serialization -----------------------------------------------------

    @classmethod
    def from_text(cls, text, id=None):
        meta, title, body = parse_meta(text)
        extra = {k: v for k, v in meta.items() if k not in DOMAIN_KEYS}

        def lst(key):
            v = meta.get(key) or []
            return [v] if isinstance(v, str) else list(v)

        ms = meta.get("milestone") or ""
        return cls(id=id or meta.get("id") or "",
                   title=title, body=body.strip(),
                   state=meta.get("state") or "open",
                   labels=lst("labels"), assignees=lst("assignees"),
                   milestone="" if ms == "none" else ms,
                   depends=lst("depends"),
                   origin=meta.get("origin") or LOCAL, extra=extra)

    def to_text(self):
        meta = dict(self.extra)
        meta.update({
            "id": self.id,
            "state": self.state,
            "labels": self.labels,
            "assignees": self.assignees,
            "milestone": self.milestone or "none",
            "depends": self.depends,
            "origin": self.origin,
        })
        body = self.body.strip() or "(no body)"
        return "%s\n# %s\n\n%s\n" % (render_meta(meta), self.title, body)


# --------------------------------------------------------------------------
# body sections
# --------------------------------------------------------------------------

def section_body(body, header):
    """Text under `header`, up to the next `## ` heading."""
    out, active = [], False
    for line in (body or "").splitlines():
        if line.startswith("## "):
            if active:
                break
            active = line.strip() == header
            continue
        if active:
            out.append(line)
    return "\n".join(out).strip()


def body_dep_refs(body):
    """Tokens referenced from `## Depends on` / `## Issues` only — never from
    prose, or a graph walk would drag in half the backlog. Returns whatever was
    written there (slugs, and `#N` on issues that came from a tracker)."""
    out, active = [], False
    for line in (body or "").splitlines():
        if line.startswith("## "):
            active = line.strip() in (DEPENDS_SECTION, "## Issues")
            continue
        if not active:
            continue
        for tok in re.findall(r'#(\d+)|\b([a-z0-9]+(?:-[a-z0-9]+)+)\b', line):
            ref = ("#" + tok[0]) if tok[0] else tok[1]
            if ref not in out:
                out.append(ref)
    return out


# --------------------------------------------------------------------------
# checkboxes
# --------------------------------------------------------------------------

# A checkbox is the one part of a body that is *state* and not prose, so the
# format gives it markup of its own (references/format.md:163-164). It is item
# markup, not a property of one section: `## Acceptance criteria` is the usual
# home, but a type/feature keeps its children as checkboxes under `## Issues`
# (format.md:275-277). The scan is therefore over the whole text and the
# heading is only recorded, never required.
CHECKBOX_RE = re.compile(
    r'^(?P<indent>[ \t]*)(?P<marker>[-*+]|\d+[.)])[ \t]+'
    r'\[(?P<box>[ xX])\](?=[ \t]|$)(?P<text>.*)$')
# Any list item — a sibling ends the item above it, checkbox or not.
LIST_ITEM_RE = re.compile(r'^[ \t]*([-*+]|\d+[.)])([ \t]|$)')
FENCE_RE = re.compile(r'^[ \t]{0,3}(`{3,}|~{3,})')

Checkbox = collections.namedtuple(
    "Checkbox", "index line end_line checked text section")


def checkboxes(text):
    """Every checkbox item in `text`, in document order.

    A pure function of the string it is given — no I/O, no store, no tracker.
    Pass an issue body (`Issue.body`) to get body-relative line numbers, or a
    whole file to get file-relative ones; nothing else changes.

    Returns a list of `Checkbox` namedtuples:

        index     1-based position in this list — what a user types to pick it
        line      1-based line of the `- [ ]` marker, in the text given
        end_line  1-based last line of the item, continuation lines included
        checked   True for `[x]` / `[X]`, False for `[ ]`
        text      the item's text; continuation lines joined with one space
        section   nearest preceding `## ` heading, "" above the first one

    Rules:

    - Only a line matching CHECKBOX_RE opens an item. A wrapped ("continuation")
      line is part of the item above it, never an item of its own; the item
      runs to the next blank line, heading, code fence, or list marker.
    - Fenced code blocks are skipped whole: `- [ ]` inside a ``` fence is an
      example of the markup, not a box anybody may tick.
    - `-`, `*`, `+` and `1.` markers all count, at any indentation, so nested
      lists are seen too.
    """
    lines = (text or "").splitlines()
    items, section, fence = [], "", ""
    for n, line in enumerate(lines, 1):
        m = FENCE_RE.match(line)
        if m:
            tok = m.group(1)
            if not fence:
                fence = tok
            elif tok[0] == fence[0] and len(tok) >= len(fence):
                fence = ""
            continue
        if fence:
            continue
        if line.startswith("## "):
            section = line.strip()
            continue
        if line.startswith("# "):
            section = ""
            continue
        m = CHECKBOX_RE.match(line)
        if not m:
            continue
        end, parts = n, [m.group("text").strip()]
        for k in range(n, len(lines)):  # lines[k] is line number k + 1
            nxt = lines[k]
            if (not nxt.strip() or nxt.startswith("#")
                    or FENCE_RE.match(nxt) or LIST_ITEM_RE.match(nxt)):
                break
            end = k + 1
            parts.append(nxt.strip())
        items.append(Checkbox(len(items) + 1, n, end,
                              m.group("box") != " ",
                              " ".join(p for p in parts if p), section))
    return items


def set_checkbox(text, item, checked=True):
    """Return `text` with one checkbox set to `checked`.

    Pure, and deliberately surgical: exactly one character of the input
    changes — the one between the brackets. Everything else, including
    trailing whitespace and the item's own wording, comes back byte for byte.
    That is the whole point of the function: ticking a box must not produce a
    diff wider than the state that changed.

    `item` is a `Checkbox` from `checkboxes(text)` — the same text, or the
    line number will point at the wrong line — or a 1-based line number.
    Already in the requested state is a no-op: `text` is returned unchanged,
    and an existing `[X]` keeps its capital.
    """
    line_no = item.line if isinstance(item, Checkbox) else int(item)
    off = 0
    for n, raw in enumerate(text.splitlines(True), 1):
        if n == line_no:
            m = CHECKBOX_RE.match(raw.rstrip("\r\n"))
            if not m:
                raise ValueError("line %d is not a checkbox item" % line_no)
            if (m.group("box") != " ") == bool(checked):
                return text
            box = off + m.start("box")
            return text[:box] + ("x" if checked else " ") + text[box + 1:]
        off += len(raw)
    raise ValueError("line %d is past the end of the text" % line_no)


def checkbox_progress(text):
    """(done, total) over every checkbox in `text`; (0, 0) when it has none.

    Computed on the fly, on purpose. Progress is not a metadata field: it is
    the body read back, and the body is the only place the state lives."""
    items = checkboxes(text)
    return sum(1 for c in items if c.checked), len(items)


# --------------------------------------------------------------------------
# validation
# --------------------------------------------------------------------------

def validate(issue, known_ids=None):
    """Return (errors, warnings). Errors mean the issue is not well-formed in
    the canonical format; warnings mean it deviates from its type template."""
    err, warn = [], []

    if not issue.id:
        err.append("no `id:` — the slug is the issue's identity")
    elif not SLUG_OK.match(issue.id):
        err.append("id %r is not a slug (lowercase, digits, single dashes)" % issue.id)

    if issue.state not in STATES:
        err.append("state %r must be one of: %s" % (issue.state, ", ".join(STATES)))

    types = [l for l in issue.labels if l.startswith("type/")]
    if len(types) != 1:
        err.append("need exactly one type/* label, found %d: %s"
                   % (len(types), ", ".join(types) or "none"))
    elif issue.type not in TYPES:
        err.append("unknown type %r — known: %s" % (issue.type, ", ".join(sorted(TYPES))))
    if len([l for l in issue.labels if l.startswith("severity/")]) > 1:
        err.append("at most one severity/* label")
    if issue.severity and issue.severity not in SEVERITIES:
        warn.append("unknown severity %r" % issue.severity)

    if not issue.title:
        err.append("no `# Title` heading below the metadata block")
    else:
        if TITLE_PREFIX.match(issue.title):
            err.append("title carries a type prefix (%r) — the type lives in the label"
                       % issue.title[:24])
        if CYRILLIC.search(issue.title):
            err.append("title must be English, imperative mood (prose stays Russian)")

    for h in REQUIRED_SECTIONS:
        if h not in issue.body:
            err.append("missing section %s" % h)
    if issue.type != "draft" and AC_SECTION not in issue.body:
        err.append("missing section %s" % AC_SECTION)
    if "## Spec" in issue.body and not section_body(issue.body, "## Spec"):
        err.append("## Spec is empty — put a repo path, a URL, or the literal `none`")

    for h in EXPECTED_SECTIONS.get(issue.type, []):
        if h not in issue.body:
            warn.append("type/%s template usually has %s" % (issue.type, h))

    if issue.id in issue.depends:
        err.append("depends on itself")
    if known_ids is not None:
        for d in issue.depends:
            if d not in known_ids:
                warn.append("depends on %r, which is not in the store" % d)

    # `depends:` is the machine-readable graph; the body section is prose for
    # humans. They drift silently unless something says so.
    listed = set(issue.depends)
    for ref in body_dep_refs(issue.body):
        if not ref.startswith("#") and ref not in listed:
            warn.append("%s mentions %r but `depends:` does not list it"
                        % (DEPENDS_SECTION, ref))

    # An unticked checkbox is never a finding — neither an error nor a
    # warning. `- [ ]` is work not done yet, which is the normal state of a
    # perfectly well-formed issue. Reading that state is issue_ac.py's job.

    return err, warn


# --------------------------------------------------------------------------
# store
# --------------------------------------------------------------------------

def path_of(root, id):
    return os.path.join(root, "%s.md" % id)


def all_ids(root):
    if not os.path.isdir(root):
        return []
    return sorted(f[:-3] for f in os.listdir(root)
                  if f.endswith(".md") and not f.startswith((".", "INDEX", "tree-")))


def load(root, id):
    with open(path_of(root, id)) as f:
        return Issue.from_text(f.read(), id=id)


def load_all(root):
    return {i: load(root, i) for i in all_ids(root)}


def save(root, issue):
    os.makedirs(root, exist_ok=True)
    p = path_of(root, issue.id)
    with open(p, "w") as f:
        f.write(issue.to_text())
    return p


# --------------------------------------------------------------------------
# dependency graph
# --------------------------------------------------------------------------

def graph(issues):
    """{id: [dep ids]} from the `depends:` metadata — the authoritative edge
    list. Body prose is never walked."""
    return {i: list(iss.depends) for i, iss in issues.items()}


def dependents(issues, id):
    """Who depends on `id` (the upward direction)."""
    return sorted(i for i, iss in issues.items() if id in iss.depends)


def topo_order(ids, edges):
    """Dependencies first. Cycles are broken deterministically rather than
    raising: a cycle is a data problem for the caller to report, not a reason
    to refuse to order the rest."""
    order, state = [], {}

    def visit(n):
        if state.get(n) == "done":
            return
        if state.get(n) == "open":
            return  # cycle — leave the back edge unresolved
        state[n] = "open"
        for d in edges.get(n, []):
            if d in edges:
                visit(d)
        state[n] = "done"
        order.append(n)

    for n in ids:
        visit(n)
    return order


def find_cycles(edges):
    """List of id lists, one per cycle found. Empty when the graph is a DAG."""
    cycles, state, stack = [], {}, []

    def visit(n):
        state[n] = "open"
        stack.append(n)
        for d in edges.get(n, []):
            if d not in edges:
                continue
            if state.get(d) == "open":
                cycles.append(stack[stack.index(d):] + [d])
            elif d not in state:
                visit(d)
        stack.pop()
        state[n] = "done"

    for n in edges:
        if n not in state:
            visit(n)
    return cycles