#!/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 # -------------------------------------------------------------------------- # where the store lives # -------------------------------------------------------------------------- # `/tmp/issues`, absolute, resolved once at import. # # It used to be the relative `tmp/issues`, which made "the store" whatever # directory the shell happened to be standing in. One `cd` — and a `cd` outlives # the command that ran it — was enough for readers to report an empty store on a # full one and for writers to quietly build a second store beside the first. # # The anchor is THIS FILE, not the working directory. A script's own location is # a fact about the installation; cwd is a fact about the last `cd`. Walking up # from __file__ therefore hands every script in both layers the same answer no # matter where it is invoked from — including from inside tmp/issues itself. # # An explicit --out still wins over all of this, and is used exactly as typed: a # relative --out stays relative to cwd, because that is what the operator asked # for. There is no environment override; the store is where the repo is. STORE_PARTS = ("tmp", "issues") # `.git` is a directory in a normal clone and a FILE in a worktree — hence # exists(), not isdir(). AGENTS.md is the fallback for a plugin copied out of # git; the agents-sync hook only ever puts one at a repository root. REPO_MARKERS = (".git", "AGENTS.md") _HERE = os.path.dirname(os.path.abspath(__file__)) def repo_root(start): """Nearest ancestor of `start` (inclusive) carrying a repo marker, or None. Markers, not a fixed number of `..` hops: how deep this file sits below the root is an implementation detail of the repo layout, and the layout is not a promise.""" d = os.path.abspath(start) while True: if any(os.path.exists(os.path.join(d, m)) for m in REPO_MARKERS): return d parent = os.path.dirname(d) if parent == d: return None d = parent def store_root(start=None): """Absolute path of the issue store. `start` overrides the anchor and exists so the resolution can be exercised against a scratch tree. When these scripts are not inside a repository at all, cwd gets a turn; failing that the historical cwd-relative location stands, made absolute so an error message can name the directory it really looked in.""" for anchor in ([start] if start is not None else [_HERE, os.getcwd()]): root = repo_root(anchor) if root: return os.path.join(root, *STORE_PARTS) return os.path.abspath(os.path.join(*STORE_PARTS)) ISSUE_ROOT = store_root() # 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" ISSUES_SECTION = "## Issues" # Both sections name what an issue depends on, so both are edge sources and # both point the same way. In a `type/feature` that reads container -> child: # "the container is closed when its children are closed" IS a dependency. # "a child belongs to a feature" is membership, and membership has no place in # a dependency graph — which is why a child never names its container back. DEP_SECTIONS = (DEPENDS_SECTION, ISSUES_SECTION) # 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_SECTION], "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 complete state, not a pending one — and the state in which this file is the only copy of the work. An issue whose `origin` names somewhere else can be fetched from there again; this one cannot.""" 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_ref_sections(body): """[(section, ref)] for every reference under one of DEP_SECTIONS — never from prose, or a graph walk would drag in half the backlog. Refs are whatever was written there (slugs, and `#N` on issues that came from a tracker), deduplicated on first sight. The section is carried out with the ref so a caller can name the one the reader actually has in front of them: a container's children come from `## Issues`, and pointing at `## Depends on` would name a section that is not in the file.""" out, seen, section = [], set(), "" for line in (body or "").splitlines(): if line.startswith("## "): head = line.strip() section = head if head in DEP_SECTIONS else "" continue if not section: 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 seen: seen.add(ref) out.append((section, ref)) return out def body_dep_refs(body): """Just the refs, in order of first appearance.""" return [ref for _, ref in body_dep_ref_sections(body)] # -------------------------------------------------------------------------- # 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[ \t]*)(?P[-*+]|\d+[.)])[ \t]+' r'\[(?P[ xX])\](?=[ \t]|$)(?P.*)$') # 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. Name the section # the reference actually came from — for a container that is `## Issues`. listed = set(issue.depends) for section, ref in body_dep_ref_sections(issue.body): if not ref.startswith("#") and ref not in listed: warn.append("%s mentions %r but `depends:` does not list it" % (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 # -------------------------------------------------------------------------- class StoreMissing(Exception): """The store directory is not there. Deliberately a different answer from "the store is empty". One is a path that does not exist, the other is a repository with no issues filed yet, and conflating the two is exactly what made a missed directory look like an empty backlog.""" def __init__(self, root): self.root = root Exception.__init__(self, "store %s does not exist" % root) def store_exists(root): return os.path.isdir(root) def require_store(root): """Assert the store is there before reading or writing it.""" if not os.path.isdir(root): raise StoreMissing(root) return root def create_store(root): """Create the store; True when it actually made the directory. Only the commands that legitimately bootstrap a store call this — issue_new and pull — and both announce it. Nothing creates a store as a side effect of a write any more: a missing directory is something to report, not something to conjure.""" if os.path.isdir(root): return False os.makedirs(root) return True def store_error(root): """Why `root` cannot be read as a store, or None when it holds issues. The two messages are distinct on purpose — see StoreMissing.""" if not os.path.isdir(root): return ("store %s does not exist — nothing was created; pass --out to " "point elsewhere" % root) if not all_ids(root): return "store %s exists but is empty" % root return None def path_of(root, id): return os.path.join(root, "%s.md" % id) def all_ids(root): """Every issue in the store, by slug. An issue file is named by its slug and a slug has no dot in it (SLUG_OK), so `.comments.md` — the thread the sync layer parks beside an issue — is not one, and neither is anything else that grew a second extension. Without that rule `wire-sqlc.comments` reads as an issue called `wire-sqlc.comments`, and a bare `push.py` tries to file the comment thread as a unit of work.""" 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-")) and "." not in f[:-3]) def slug_files(root, id): """Every file the store holds under one slug — the issue and its sidecars. `.md` is the issue. Anything named `.` beside it is a companion another layer parked there (`.comments.md` is the one that exists today). `all_ids` already refuses to read those as issues because a slug has no dot in it; this is the same rule read the other way round. Which is how the domain can remove an issue *completely* without learning what any of those companions are: it does not need to know that a comment thread exists to know that a file named after this issue belongs to it and goes when it goes. The issue's own file comes first — it is the headline of any receipt printed from this list. A missing store is an empty list, not an error: nothing is there to remove. """ if not os.path.isdir(root): return [] own, sidecars = [], [] for name in sorted(os.listdir(root)): if not name.startswith("%s." % id): continue p = os.path.join(root, name) if not os.path.isfile(p): continue (own if name == "%s.md" % id else sidecars).append(p) return own + sidecars 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): require_store(root) 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