refactor: split issue domain from Gitea transport

An issue was a Gitea row that happened to be cached locally: its identity
was the tracker's number (42.md), its dependencies were tracker numbers
(depends: [#12]), and a local issue existed only as a draft that push
deleted on success. Nothing could be planned or tracked without a tracker.

Split into layers, with knowledge flowing one way:

  skills/issue  DOMAIN  what an issue is: format, validation, dep graph
        ^               offline; stdlib imports only, no subprocess
        | imports
  skills/sync   BRIDGE  map.py    md <-> Gitea JSON, pure, no I/O
                        _gitea.py login pin, api, pagination, filters
  skills/use    REFERENCE  tea CLI docs for non-issue entities

skills/issue never imports skills/sync. Delete the sync layer and the
domain keeps working.

Identity is now a slug derived from the title (wire-sqlc-appclick.md) and
is stable across retitles and pushes. Tracker numbers live in a `gitea:`
field, never in a file name and never in `depends:`; the pair is indexed
in .remote.json, which is a cache over the files, not a second source of
truth.

Behavior changes:

- Pushing is additive. The file is never deleted; it gains gitea:/url:/
  synced: and origin: flips from local to gitea. `origin: local` is a
  durable state, not a pending one.
- Pushes go in topological order so dependencies get numbers first.
- The dependency graph is computed offline from `depends:` metadata; body
  prose is passed through unchanged in both directions rather than being
  rewritten between slugs and #N.
- `origin` is domain-owned (whether work exists elsewhere is a fact about
  the work); the handle and how to reach it stay with sync.

Script moves:

  issue_get.py   -> sync/pull.py
  issue_push.py  -> sync/push.py
  issue_list.py  -> sync/remote.py
  issue_index.py -> issue/issue_index.py
  _tea.py        -> split into issue/issue.py, sync/map.py, sync/_gitea.py

New: issue/issue_new.py, issue/issue_check.py, issue/issue_tree.py, and
sync/comment.py — comment posting was the last issue operation still
hand-rolled through raw `tea api`.

references/issue-format.md moves to skills/issue/references/format.md;
label hex colors move out of it into map.py, since a color is how a
tracker paints a chip, not what an issue is.

Verified: offline path end to end (new, check, tree, index, push
--dry-run) and read-only against Gitea (remote listing, pull with
mapping, comment guard). Write paths of push.py and comment.py are not
exercised here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
naudachu
2026-08-09 23:37:32 +05:00
parent 335b0bbd54
commit 091dceec1d
24 changed files with 2504 additions and 1284 deletions
+445
View File
@@ -0,0 +1,445 @@
#!/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 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
# --------------------------------------------------------------------------
# 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))
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