Rongsong Shen
ff4e2cb84e
gitea-sync.py: --ids syncs named bugs regardless of dates (OR'd with --since). Bulk triage bumped last_seen/status_updated across most of the buglog, so --since 2026-09-01 matched ~1150 bugs while recently fixed older bugs (bug-1798..1804, 1714, 1754, 1850, 1854) could never be reached. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1033 lines
38 KiB
Python
Executable file
1033 lines
38 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
xce-gitea-sync: Sync cc-sdd specs and OpenWolf buglog to Gitea issues.
|
|
|
|
Reads .kiro/specs/<feature>/ (spec.json, brief.md, requirements.md, tasks.md)
|
|
and .wolf/buglog.json as ground truth. Creates/updates Gitea milestones, issues,
|
|
and labels to mirror the local state.
|
|
|
|
Sync is one-way: files -> Gitea. Never Gitea -> files.
|
|
Sync state (issue numbers) is stored in the source files for idempotency.
|
|
|
|
Usage:
|
|
python scripts/gitea-sync.py # sync all specs + bugs
|
|
python scripts/gitea-sync.py --spec <name> # sync one spec
|
|
python scripts/gitea-sync.py --bugs # sync bugs only
|
|
python scripts/gitea-sync.py --dry-run # show what would change
|
|
|
|
Environment:
|
|
GITEA_URL - base URL (default: https://gitea.shenrs.eu)
|
|
GITEA_TOKEN - API token (required: write:issue, write:milestone scopes)
|
|
GITEA_OWNER - repo owner (default: nexa)
|
|
GITEA_REPO - repo name (default: xce)
|
|
"""
|
|
|
|
import argparse
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from buglog import BuglogEditError, set_fields # noqa: E402
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def die(msg):
|
|
print(f"ERROR: {msg}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def warn(msg):
|
|
print(f"WARN: {msg}", file=sys.stderr)
|
|
|
|
|
|
def _env(key, default=None, required=False):
|
|
val = os.environ.get(key, default)
|
|
if required and not val:
|
|
die(f"{key} environment variable is required")
|
|
return val
|
|
|
|
|
|
def _discover_gitea_token():
|
|
"""Discover Gitea API token from multiple sources.
|
|
|
|
Priority:
|
|
1. GITEA_TOKEN environment variable
|
|
2. ~/.claude.json → mcpServers.gitea.env.GITEA_ACCESS_TOKEN
|
|
3. GITEA_ACCESS_TOKEN environment variable
|
|
"""
|
|
# Explicit override
|
|
token = _env("GITEA_TOKEN")
|
|
if token:
|
|
return token
|
|
|
|
# Try Claude Code MCP config
|
|
claude_config_path = Path.home() / ".claude.json"
|
|
if claude_config_path.exists():
|
|
try:
|
|
config = json.loads(claude_config_path.read_text())
|
|
gitea_cfg = config.get("mcpServers", {}).get("gitea", {})
|
|
token = gitea_cfg.get("env", {}).get("GITEA_ACCESS_TOKEN")
|
|
if token:
|
|
return token
|
|
except (json.JSONDecodeError, OSError):
|
|
pass
|
|
|
|
# MCP env var
|
|
return _env("GITEA_ACCESS_TOKEN")
|
|
|
|
|
|
# ── Configuration ────────────────────────────────────────────────────────────
|
|
|
|
GITEA_URL = _env("GITEA_URL", "https://gitea.shenrs.eu").rstrip("/")
|
|
GITEA_TOKEN = _discover_gitea_token()
|
|
GITEA_OWNER = _env("GITEA_OWNER", "nexa")
|
|
GITEA_REPO = _env("GITEA_REPO", "xce")
|
|
|
|
API_BASE = f"{GITEA_URL}/api/v1"
|
|
REPO_API = f"{API_BASE}/repos/{GITEA_OWNER}/{GITEA_REPO}"
|
|
|
|
# Project roots (relative to script or cwd)
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
PROJECT_ROOT = SCRIPT_DIR.parent
|
|
SPECS_DIR = PROJECT_ROOT / ".kiro" / "specs"
|
|
BUGLOG_PATH = PROJECT_ROOT / ".wolf" / "buglog.json"
|
|
|
|
|
|
# ── Label Definitions ────────────────────────────────────────────────────────
|
|
|
|
LABELS = [
|
|
# Spec phase labels
|
|
{"name": "phase:requirements", "color": "6C5CE7",
|
|
"description": "Spec requirements phase — EARS entries, acceptance criteria"},
|
|
{"name": "phase:design", "color": "6C5CE7",
|
|
"description": "Spec design phase — architecture, data flow, API contracts"},
|
|
{"name": "phase:tasks", "color": "6C5CE7",
|
|
"description": "Spec tasks phase — implementation task breakdown"},
|
|
{"name": "phase:implementation", "color": "F39C12",
|
|
"description": "Spec implementation phase — code in progress"},
|
|
{"name": "phase:complete", "color": "27AE60",
|
|
"description": "Spec complete — all tasks done, validated, merged"},
|
|
# Top-level type labels
|
|
{"name": "spec", "color": "3498DB",
|
|
"description": "Spec-driven development tracked issue"},
|
|
{"name": "bug", "color": "E74C3C",
|
|
"description": "Bug found during development"},
|
|
{"name": "bug:fixed", "color": "27AE60",
|
|
"description": "Bug resolved and fix committed"},
|
|
{"name": "bug:partial", "color": "F39C12",
|
|
"description": "Partly fixed, remainder still open"},
|
|
{"name": "bug:investigating", "color": "E67E22",
|
|
"description": "Real failure, root cause not found yet"},
|
|
{"name": "bug:cannot-reproduce", "color": "95A5A6",
|
|
"description": "Reproduction attempted and failed"},
|
|
{"name": "bug:wontfix", "color": "7F8C8D",
|
|
"description": "Not a bug / accepted limitation"},
|
|
{"name": "bug:duplicate", "color": "BDC3C7",
|
|
"description": "Duplicate of another bug"},
|
|
# Bug category labels (from buglog tags, see TAG_TO_LABEL)
|
|
{"name": "bug:asan", "color": "9B59B6",
|
|
"description": "ASAN/UBSAN caught bug"},
|
|
{"name": "bug:crash", "color": "C0392B",
|
|
"description": "SEGV/UAF/double-free crash"},
|
|
{"name": "bug:data-loss", "color": "C0392B",
|
|
"description": "Committed data loss bug"},
|
|
{"name": "bug:ha", "color": "E67E22",
|
|
"description": "HA/Raft/consensus bug"},
|
|
{"name": "bug:race", "color": "F39C12",
|
|
"description": "Data race / TOCTOU bug"},
|
|
# State labels
|
|
{"name": "blocked", "color": "E74C3C",
|
|
"description": "Blocked on external dependency or upstream spec"},
|
|
]
|
|
|
|
# Map buglog tags to Gitea labels (prefix-based matching)
|
|
TAG_TO_LABEL = {
|
|
"asan": "bug:asan",
|
|
"ubsan": "bug:asan",
|
|
"SEGV": "bug:crash",
|
|
"double-free": "bug:crash",
|
|
"heap-use-after-free": "bug:crash",
|
|
"UAF": "bug:crash",
|
|
"data-loss": "bug:data-loss",
|
|
"race": "bug:race",
|
|
"toctou": "bug:race",
|
|
"data-race": "bug:race",
|
|
"ha": "bug:ha",
|
|
"raft": "bug:ha",
|
|
"election": "bug:ha",
|
|
"failover": "bug:ha",
|
|
}
|
|
|
|
# buglog `status` -> (Gitea issue state, status label). The buglog's `status`
|
|
# field is the only source of truth for open/closed (see
|
|
# .claude/rules/buglog-status.md); never infer it from `fix`, which often says
|
|
# "NOT FIXED" / "PENDING" / "PARTIALLY FIXED". An entry with no status (e.g.
|
|
# created by an OpenWolf hook) is mirrored as open and reported for triage.
|
|
STATUS_MAP = {
|
|
"open": ("open", None),
|
|
"partial": ("open", "bug:partial"),
|
|
"investigating": ("open", "bug:investigating"),
|
|
"fixed": ("closed", "bug:fixed"),
|
|
"cannot-reproduce": ("closed", "bug:cannot-reproduce"),
|
|
"wontfix": ("closed", "bug:wontfix"),
|
|
"duplicate": ("closed", "bug:duplicate"),
|
|
}
|
|
|
|
|
|
# When several buglog entries share one Gitea issue (e.g. a user-reported
|
|
# issue whose cause was split into several bugs), the issue takes the status
|
|
# of its "most open" entry, first in this list.
|
|
STATUS_PRIORITY = ["open", "investigating", "partial", None,
|
|
"cannot-reproduce", "duplicate", "wontfix", "fixed"]
|
|
|
|
|
|
def bug_status(bug_entry):
|
|
"""The entry's status, or None when it has none / an unknown one."""
|
|
status = bug_entry.get("status")
|
|
return status if status in STATUS_MAP else None
|
|
|
|
|
|
def bug_state(bug_entry):
|
|
return STATUS_MAP.get(bug_status(bug_entry), ("open", None))[0]
|
|
|
|
|
|
def _text(value):
|
|
"""buglog text fields are a string or, in some entries, a list of strings."""
|
|
if isinstance(value, list):
|
|
return "\n\n".join(str(v).strip() for v in value if str(v).strip())
|
|
return (value or "").strip()
|
|
|
|
|
|
def api_url(path):
|
|
"""Build full API URL. All paths are repo-scoped."""
|
|
return f"{REPO_API}/{path.lstrip('/')}"
|
|
|
|
|
|
def request(method, path, body=None):
|
|
"""Make a Gitea API request. Returns (status, response_body) or None on 404."""
|
|
url = api_url(path)
|
|
data = json.dumps(body).encode("utf-8") if body else None
|
|
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Authorization", f"token {GITEA_TOKEN}")
|
|
req.add_header("Content-Type", "application/json")
|
|
req.add_header("Accept", "application/json")
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
raw = resp.read().decode("utf-8")
|
|
# DELETE (and some PUT/POST) answer 204 with an empty body.
|
|
return resp.status, (json.loads(raw) if raw.strip() else None)
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
return None, None
|
|
body_text = ""
|
|
try:
|
|
body_text = e.read().decode("utf-8")
|
|
except Exception:
|
|
pass
|
|
raise RuntimeError(
|
|
f"Gitea API error: {method} {path} -> {e.code}\n{body_text}"
|
|
) from e
|
|
except urllib.error.URLError as e:
|
|
raise RuntimeError(
|
|
f"Gitea API unreachable: {method} {path}\n{e.reason}"
|
|
) from e
|
|
except TimeoutError as e:
|
|
raise RuntimeError(f"Gitea API timed out: {method} {path}") from e
|
|
|
|
|
|
# ── GiteaAPI ─────────────────────────────────────────────────────────────────
|
|
|
|
class GiteaAPI:
|
|
"""Minimal Gitea REST API wrapper for sync operations."""
|
|
|
|
def __init__(self):
|
|
self._labels_cache = None
|
|
|
|
# Labels
|
|
|
|
def list_labels(self):
|
|
if self._labels_cache is None:
|
|
_, labels = request("GET", "/labels")
|
|
self._labels_cache = {l["name"]: l for l in (labels or [])}
|
|
return self._labels_cache
|
|
|
|
def ensure_label(self, name, color, description=""):
|
|
existing = self.list_labels()
|
|
if name in existing:
|
|
return existing[name]
|
|
warn(f" + label '{name}'")
|
|
_, label = request("POST", "/labels", {
|
|
"name": name, "color": color, "description": description
|
|
})
|
|
if label:
|
|
self._labels_cache[name] = label
|
|
return label
|
|
|
|
# Milestones
|
|
|
|
def list_milestones(self, state="all"):
|
|
_, milestones = request("GET", f"/milestones?state={state}")
|
|
return {m["title"]: m for m in (milestones or [])}
|
|
|
|
def create_milestone(self, title, description="", state="open"):
|
|
_, milestone = request("POST", "/milestones", {
|
|
"title": title, "description": description, "state": state
|
|
})
|
|
return milestone
|
|
|
|
def update_milestone(self, milestone_id, **fields):
|
|
request("PATCH", f"/milestones/{milestone_id}", fields)
|
|
|
|
# Issues
|
|
|
|
def list_issues(self, milestone_id=None, state="all"):
|
|
"""List issues, optionally filtered by milestone."""
|
|
params = [f"state={state}"]
|
|
if milestone_id is not None:
|
|
params.append(f"milestone={milestone_id}")
|
|
url = "?" + "&".join(params)
|
|
_, issues = request("GET", f"/issues{url}")
|
|
return issues or []
|
|
|
|
def create_issue(self, title, body, milestone_id=None, labels=None,
|
|
state="open"):
|
|
payload = {"title": title, "body": body, "state": state}
|
|
if milestone_id:
|
|
payload["milestone"] = milestone_id
|
|
if labels:
|
|
payload["labels"] = [int(l) if isinstance(l, (int, float)) else l
|
|
for l in labels]
|
|
_, issue = request("POST", "/issues", payload)
|
|
return issue
|
|
|
|
def update_issue(self, issue_number, **fields):
|
|
"""Update issue fields (state, title, body, milestone)."""
|
|
request("PATCH", f"/issues/{issue_number}", fields)
|
|
|
|
def add_labels(self, issue_number, label_ids):
|
|
request("POST", f"/issues/{issue_number}/labels", {"labels": label_ids})
|
|
|
|
def remove_label(self, issue_number, label_id):
|
|
request("DELETE", f"/issues/{issue_number}/labels/{label_id}")
|
|
|
|
def add_comment(self, issue_number, body):
|
|
request("POST", f"/issues/{issue_number}/comments", {"body": body})
|
|
|
|
|
|
# ── Label Sync ───────────────────────────────────────────────────────────────
|
|
|
|
def ensure_all_labels(api):
|
|
"""Create any missing labels. Runs before issue/milestone creation."""
|
|
for lbl in LABELS:
|
|
api.ensure_label(lbl["name"], lbl["color"], lbl.get("description", ""))
|
|
|
|
|
|
# ── Bug Tag → Label Mapping ──────────────────────────────────────────────────
|
|
|
|
def status_label(status):
|
|
return STATUS_MAP.get(status, (None, None))[1]
|
|
|
|
|
|
STATUS_LABELS = {lbl for _, lbl in STATUS_MAP.values() if lbl}
|
|
|
|
|
|
def bug_labels(bug_entry):
|
|
"""Derive Gitea labels from the entry's status and tags."""
|
|
labels = ["bug"]
|
|
if status_label(bug_status(bug_entry)):
|
|
labels.append(status_label(bug_status(bug_entry)))
|
|
|
|
# Map known tag patterns to labels
|
|
for tag in (t.lower() for t in bug_entry.get("tags", [])):
|
|
for pattern, label in TAG_TO_LABEL.items():
|
|
if pattern.lower() in tag and label not in labels:
|
|
labels.append(label)
|
|
return labels
|
|
|
|
|
|
|
|
|
|
# ── Body Formatting ──────────────────────────────────────────────────────────
|
|
|
|
def bug_body(bug_entry):
|
|
"""Format a buglog entry as a Gitea issue body."""
|
|
parts = []
|
|
|
|
status = bug_entry.get("status") or "unknown (needs triage)"
|
|
note = _text(bug_entry.get("status_note"))
|
|
parts.append(f"**Status:** {status}" + (f" — {note}" if note else ""))
|
|
if bug_entry.get("duplicate_of"):
|
|
parts.append(f"Duplicate of {bug_entry['duplicate_of']}")
|
|
if bug_entry.get("repro_attempts"):
|
|
parts.append(f"Repro attempts: {_text(bug_entry['repro_attempts'])}")
|
|
parts.append("")
|
|
|
|
error = _text(bug_entry.get("error_message"))
|
|
if error:
|
|
parts.append("## Error\n")
|
|
parts.append(error)
|
|
parts.append("")
|
|
|
|
root_cause = _text(bug_entry.get("root_cause"))
|
|
if root_cause:
|
|
parts.append("## Root Cause\n")
|
|
parts.append(root_cause)
|
|
parts.append("")
|
|
|
|
fix = _text(bug_entry.get("fix"))
|
|
if fix:
|
|
parts.append("## Fix\n")
|
|
parts.append(fix)
|
|
parts.append("")
|
|
else:
|
|
parts.append("## Fix\n")
|
|
parts.append("*Not yet fixed*")
|
|
parts.append("")
|
|
|
|
files = [f.strip() for f in (bug_entry.get("file") or "").split(",")]
|
|
files += [str(f).strip() for f in bug_entry.get("files") or []]
|
|
files = [f for f in dict.fromkeys(files) if f]
|
|
if files:
|
|
parts.append("## Affected Files\n")
|
|
for f in files:
|
|
parts.append(f"- `{f}`")
|
|
parts.append("")
|
|
|
|
tags = bug_entry.get("tags", [])
|
|
if tags:
|
|
parts.append("## Tags\n")
|
|
parts.append(" ".join(f"`{t}`" for t in tags))
|
|
parts.append("")
|
|
|
|
related = bug_entry.get("related_bugs", [])
|
|
if related:
|
|
parts.append("## Related Bugs\n")
|
|
for rb in related:
|
|
parts.append(f"- {rb}")
|
|
parts.append("")
|
|
|
|
parts.append("---")
|
|
parts.append(f"*Synced from `.wolf/buglog.json` "
|
|
f"({bug_entry.get('id', '?')})*")
|
|
|
|
return "\n".join(parts)
|
|
|
|
|
|
def bug_title(bug_entry):
|
|
"""Build a short title from a buglog entry."""
|
|
bid = bug_entry.get("id", "bug-?")
|
|
error = _text(bug_entry.get("error_message"))
|
|
# Drop a leading status banner ("OPEN, NOT FIXED -- ...", "PRIORITY: LOW --
|
|
# ...") -- status lives in labels/state, not the title.
|
|
error = re.sub(r"^(?:\s*(?:OPEN|NOT (?:YET )?FIXED|NOT A CRASH[^,.]*|"
|
|
r"NOT YET ROOT-CAUSED|DEAD-CODE-ONLY|PRIORITY: \w+|"
|
|
r"low priority|pre-existing)\b\s*(?:,|--|\.|;)?)+\s*",
|
|
"", error, flags=re.I) or error
|
|
# Take first sentence or first 100 chars
|
|
first_line = error.split("\n")[0].strip()
|
|
first_sentence = re.split(r'[.。]\s+', first_line)[0].strip()
|
|
if not first_sentence:
|
|
first_sentence = first_line
|
|
if len(first_sentence) > 120:
|
|
first_sentence = first_sentence[:117] + "..."
|
|
return f"[{bid}] {first_sentence}"
|
|
|
|
|
|
def spec_requirements_body(spec_name, spec_data, brief_path, req_path, design_path):
|
|
"""Build body for the Requirements & Design phase issue."""
|
|
parts = []
|
|
parts.append(f"# {spec_name}: Requirements & Design")
|
|
parts.append("")
|
|
|
|
# Phase status from spec.json
|
|
approvals = spec_data.get("approvals", {})
|
|
req_approved = approvals.get("requirements", {}).get("approved", False)
|
|
design_approved = approvals.get("design", {}).get("approved", False)
|
|
phase = spec_data.get("phase", "unknown")
|
|
|
|
parts.append(f"**Phase**: `{phase}`")
|
|
parts.append(f"**Requirements**: {'✅ Approved' if req_approved else '⏳ Pending'}")
|
|
parts.append(f"**Design**: {'✅ Approved' if design_approved else '⏳ Pending'}")
|
|
parts.append("")
|
|
|
|
# Extract requirements summary from requirements.md if available
|
|
if req_path.exists():
|
|
req_text = req_path.read_text()
|
|
# Find the requirements summary table
|
|
table_match = re.search(
|
|
r'## Requirements Summary\s*\n(.*?)(?=\n## |\Z)',
|
|
req_text, re.DOTALL
|
|
)
|
|
if table_match:
|
|
parts.append("## Requirements")
|
|
parts.append("")
|
|
parts.append(table_match.group(1).strip())
|
|
parts.append("")
|
|
|
|
# Blockers
|
|
blocked_on = spec_data.get("blocked_on", [])
|
|
if blocked_on:
|
|
parts.append("## Blockers")
|
|
parts.append("")
|
|
for b in blocked_on:
|
|
parts.append(f"- {b}")
|
|
parts.append("")
|
|
|
|
parts.append("---")
|
|
parts.append(f"See: `.kiro/specs/{spec_name}/`")
|
|
|
|
return "\n".join(parts)
|
|
|
|
|
|
def spec_implementation_body(spec_name, spec_data, tasks_path):
|
|
"""Build body for the Implementation phase issue."""
|
|
parts = []
|
|
parts.append(f"# {spec_name}: Implementation")
|
|
parts.append("")
|
|
|
|
# Phase status
|
|
tasks = spec_data.get("approvals", {}).get("tasks", {})
|
|
ready = spec_data.get("ready_for_implementation", False)
|
|
parts.append(f"**Phase**: {spec_data.get('phase', 'unknown')}")
|
|
parts.append(f"**Tasks generated**: {tasks.get('generated', False)}")
|
|
parts.append(f"**Tasks approved**: {tasks.get('approved', False)}")
|
|
parts.append(f"**Ready for implementation**: {ready}")
|
|
parts.append("")
|
|
|
|
# Task checklist from tasks.md
|
|
if tasks_path.exists():
|
|
tasks_text = tasks_path.read_text()
|
|
# Extract - [ ] / - [x] checklist items
|
|
tasks = re.findall(r'^(\s*- \[([ x])\]\s+.+)$', tasks_text, re.MULTILINE)
|
|
if tasks:
|
|
parts.append("## Tasks")
|
|
parts.append("")
|
|
for line, status in tasks:
|
|
parts.append(line.strip())
|
|
parts.append("")
|
|
|
|
# Blockers
|
|
blocked_on = spec_data.get("blocked_on", [])
|
|
if blocked_on:
|
|
parts.append("## Blockers")
|
|
parts.append("")
|
|
for b in blocked_on:
|
|
parts.append(f"- {b}")
|
|
parts.append("")
|
|
|
|
parts.append("---")
|
|
parts.append(f"See: `.kiro/specs/{spec_name}/tasks.md`")
|
|
|
|
return "\n".join(parts)
|
|
|
|
|
|
def spec_validation_body(spec_name, spec_data):
|
|
"""Build body for the Validation & Closure phase issue."""
|
|
parts = []
|
|
parts.append(f"# {spec_name}: Validation & Closure")
|
|
parts.append("")
|
|
|
|
phase = spec_data.get("phase", "unknown")
|
|
approvals = spec_data.get("approvals", {})
|
|
open_phases = [
|
|
k for k, v in approvals.items()
|
|
if v.get("generated") and not v.get("approved")
|
|
]
|
|
|
|
if phase == "superseded":
|
|
parts.append(f"**Status**: Superseded by {spec_data.get('superseded_by', '?')}")
|
|
elif phase in DONE_PHASES:
|
|
parts.append("**Status**: ✅ Complete")
|
|
elif open_phases:
|
|
parts.append(f"**Status**: ⏳ Pending — awaiting: {', '.join(open_phases)}")
|
|
else:
|
|
parts.append("**Status**: ⏳ Future")
|
|
|
|
parts.append("")
|
|
parts.append("## Acceptance Gates")
|
|
parts.append("")
|
|
parts.append("- [ ] All requirements verified")
|
|
parts.append("- [ ] Full test suite passes")
|
|
parts.append("- [ ] HA soak validation passes")
|
|
parts.append("- [ ] Code review sign-off")
|
|
parts.append("- [ ] Spec marked complete in spec.json")
|
|
parts.append("")
|
|
|
|
blocked_on = spec_data.get("blocked_on", [])
|
|
if blocked_on:
|
|
parts.append("## Blockers")
|
|
parts.append("")
|
|
for b in blocked_on:
|
|
parts.append(f"- {b}")
|
|
parts.append("")
|
|
|
|
parts.append("---")
|
|
parts.append(f"See: `.kiro/specs/{spec_name}/`")
|
|
|
|
return "\n".join(parts)
|
|
|
|
|
|
# ── Spec Sync ────────────────────────────────────────────────────────────────
|
|
|
|
PHASE_DEFS = [
|
|
{
|
|
"key": "requirements_design",
|
|
"title_suffix": "Requirements & Design",
|
|
"labels": ["spec", "phase:requirements", "phase:design"],
|
|
"body_fn": spec_requirements_body,
|
|
"body_fn_extra": lambda n, d: spec_requirements_body(
|
|
n, d,
|
|
SPECS_DIR / n / "brief.md",
|
|
SPECS_DIR / n / "requirements.md",
|
|
SPECS_DIR / n / "design.md",
|
|
),
|
|
},
|
|
{
|
|
"key": "implementation",
|
|
"title_suffix": "Implementation",
|
|
"labels": ["spec", "phase:implementation"],
|
|
"body_fn": lambda n, d: spec_implementation_body(
|
|
n, d, SPECS_DIR / n / "tasks.md"
|
|
),
|
|
"body_fn_extra": None,
|
|
},
|
|
{
|
|
"key": "validation",
|
|
"title_suffix": "Validation & Closure",
|
|
"labels": ["spec", "phase:complete"],
|
|
"body_fn": lambda n, d: spec_validation_body(n, d),
|
|
"body_fn_extra": None,
|
|
},
|
|
]
|
|
|
|
|
|
# spec.json `phase` values that mean the spec's work is finished. There is no
|
|
# implementation approval in spec.json (approvals cover requirements/design/
|
|
# tasks only), and no tool distinguishes "implemented" from
|
|
# "implementation-complete", so both count as done.
|
|
DONE_PHASES = {"implementation-complete", "implemented", "superseded"}
|
|
|
|
|
|
def _spec_desired_states(spec_data):
|
|
"""Desired Gitea state per phase-issue key, plus the milestone's."""
|
|
approvals = spec_data.get("approvals", {})
|
|
req_ok = approvals.get("requirements", {}).get("approved", False)
|
|
des_ok = approvals.get("design", {}).get("approved", False)
|
|
done = spec_data.get("phase") in DONE_PHASES
|
|
closed = lambda cond: "closed" if cond else "open"
|
|
return {
|
|
"requirements_design": closed((req_ok and des_ok) or done),
|
|
"implementation": closed(done),
|
|
"validation": closed(done),
|
|
}, closed(done)
|
|
|
|
|
|
def sync_spec(api, spec_name, dry_run=False, quiet=False):
|
|
"""Sync one spec directory to Gitea milestone + phase issues."""
|
|
spec_dir = SPECS_DIR / spec_name
|
|
spec_json_path = spec_dir / "spec.json"
|
|
|
|
if not spec_json_path.exists():
|
|
if not quiet:
|
|
print(f"SKIP {spec_name}: no spec.json")
|
|
return
|
|
|
|
spec_data = json.loads(spec_json_path.read_text())
|
|
gitea_state = spec_data.get("gitea", {})
|
|
|
|
# Check for already-synced state
|
|
if gitea_state.get("milestone_id") and gitea_state.get("issues"):
|
|
if all(
|
|
k in gitea_state["issues"]
|
|
for k in ["requirements_design", "implementation", "validation"]
|
|
):
|
|
# Already synced — check if phase states need update
|
|
_sync_spec_phase_states(api, spec_name, spec_data, gitea_state,
|
|
dry_run, quiet)
|
|
return
|
|
|
|
# Need to create milestone + issues (or adopt existing)
|
|
if not quiet:
|
|
print(f"SYNC spec: {spec_name}")
|
|
|
|
if dry_run:
|
|
print(f" [DRY-RUN] create milestone '{spec_name}'")
|
|
for phase_def in PHASE_DEFS:
|
|
title = f"[{spec_name}] {phase_def['title_suffix']}"
|
|
print(f" [DRY-RUN] create issue '{title}'")
|
|
return
|
|
|
|
# Try to adopt existing milestone
|
|
milestone = None
|
|
existing_milestones = api.list_milestones()
|
|
if spec_name in existing_milestones:
|
|
milestone = existing_milestones[spec_name]
|
|
if not quiet:
|
|
print(f" adopting existing milestone #{milestone['id']}")
|
|
else:
|
|
milestone = api.create_milestone(
|
|
spec_name,
|
|
description=_spec_milestone_desc(spec_name, spec_data),
|
|
state="open"
|
|
)
|
|
if not quiet:
|
|
print(f" created milestone #{milestone['id']}")
|
|
|
|
if not milestone:
|
|
warn(f" could not find or create milestone for '{spec_name}'")
|
|
return
|
|
|
|
# Create phase issues
|
|
milestone_id = milestone["id"]
|
|
phase_issues = {}
|
|
|
|
for phase_def in PHASE_DEFS:
|
|
title = f"[{spec_name}] {phase_def['title_suffix']}"
|
|
|
|
state = _spec_desired_states(spec_data)[0][phase_def["key"]]
|
|
|
|
# Build body
|
|
if phase_def.get("body_fn_extra"):
|
|
body = phase_def["body_fn_extra"](spec_name, spec_data)
|
|
else:
|
|
body = phase_def["body_fn"](spec_name, spec_data)
|
|
|
|
# Determine labels (use label IDs if available)
|
|
label_ids = _resolve_label_ids(api, phase_def["labels"])
|
|
|
|
issue = api.create_issue(
|
|
title, body, milestone_id=milestone_id,
|
|
labels=label_ids, state=state
|
|
)
|
|
phase_issues[phase_def["key"]] = issue["number"]
|
|
if not quiet:
|
|
state_mark = " (closed)" if state == "closed" else ""
|
|
print(f" created issue #{issue['number']}: {title}{state_mark}")
|
|
|
|
# Write sync state back to spec.json
|
|
spec_data["gitea"] = {
|
|
"milestone_id": milestone_id,
|
|
"issues": phase_issues,
|
|
}
|
|
_append_top_level_key(spec_json_path, "gitea", spec_data["gitea"])
|
|
|
|
|
|
def _sync_spec_phase_states(api, spec_name, spec_data, gitea_state,
|
|
dry_run, quiet):
|
|
"""Update issue states + milestone based on current approvals."""
|
|
if dry_run:
|
|
# In dry-run mode, just report what would be checked
|
|
print(f" [DRY-RUN] check phase states for '{spec_name}'")
|
|
return
|
|
|
|
issues = gitea_state.get("issues", {})
|
|
by_key, milestone_state = _spec_desired_states(spec_data)
|
|
desired = {issues[k]: st for k, st in by_key.items() if issues.get(k)}
|
|
|
|
# Fetch existing issues once
|
|
existing_issues = api.list_issues(
|
|
milestone_id=gitea_state.get("milestone_id"), state="all"
|
|
)
|
|
|
|
# Update any issues whose state doesn't match desired
|
|
for issue_num, target_state in desired.items():
|
|
current = next(
|
|
(i for i in existing_issues if i["number"] == issue_num), None
|
|
)
|
|
if current and current.get("state") != target_state:
|
|
api.update_issue(issue_num, state=target_state)
|
|
if not quiet:
|
|
label = str(issue_num)
|
|
print(f" issue #{issue_num}: "
|
|
f"{current['state']} -> {target_state}")
|
|
|
|
# Update milestone state
|
|
milestone_id = gitea_state.get("milestone_id")
|
|
if milestone_id:
|
|
ms = api.list_milestones(state="all").get(spec_name)
|
|
if ms and ms.get("state") != milestone_state:
|
|
api.update_milestone(milestone_id, state=milestone_state)
|
|
if not quiet:
|
|
print(f" milestone '{spec_name}': "
|
|
f"{ms['state']} -> {milestone_state}")
|
|
|
|
|
|
def _spec_milestone_desc(spec_name, spec_data):
|
|
"""Build milestone description from spec data."""
|
|
phase = spec_data.get("phase", "unknown")
|
|
blocked = spec_data.get("blocked_on", [])
|
|
created = spec_data.get("created_at", "")
|
|
parts = [f"Spec-driven development for {spec_name}."]
|
|
parts.append(f"Phase: {phase}")
|
|
if blocked:
|
|
parts.append(f"Blocked on: {', '.join(blocked)}")
|
|
parts.append(f"Created: {created}")
|
|
return "\n".join(parts)
|
|
|
|
|
|
def _resolve_label_ids(api, label_names):
|
|
"""Convert label names to IDs for issue creation."""
|
|
existing = api.list_labels()
|
|
ids = []
|
|
for name in label_names:
|
|
if name in existing:
|
|
ids.append(existing[name]["id"])
|
|
return ids
|
|
|
|
|
|
# ── Bug Sync ─────────────────────────────────────────────────────────────────
|
|
|
|
def sync_bugs(api, dry_run=False, quiet=False, since=None, ids=None):
|
|
"""Sync buglog entries to Gitea issues.
|
|
|
|
Open/closed and the status label come from each entry's `status` (see
|
|
STATUS_MAP). `gitea_synced_status` records the status last mirrored, so an
|
|
entry is only touched when its status changed -- including going back to
|
|
open (reopen). Sync state is written back per entry, surgically, so a
|
|
failure part-way through never loses an already-created issue number.
|
|
|
|
Args:
|
|
since: ISO date (YYYY-MM-DD). Create issues only for bugs with
|
|
last_seen >= since; update already-synced ones whose
|
|
last_seen or status_updated >= since.
|
|
ids: set of bug ids to sync regardless of dates. Combined with
|
|
since, a bug is synced if it matches either -- bulk triage
|
|
has bumped last_seen/status_updated across most of the log,
|
|
so dates alone cannot pick out recently fixed older bugs.
|
|
"""
|
|
if not BUGLOG_PATH.exists():
|
|
warn(f"buglog not found: {BUGLOG_PATH}")
|
|
return
|
|
|
|
buglog = json.loads(BUGLOG_PATH.read_text())["bugs"]
|
|
created = closed = reopened = updated = skipped = 0
|
|
needs_status = []
|
|
|
|
# Aggregate status per already-synced issue (see STATUS_PRIORITY).
|
|
issue_status = {}
|
|
for bug in buglog:
|
|
num = bug.get("gitea_issue")
|
|
if num:
|
|
prev = issue_status.get(num, "fixed")
|
|
cur = bug_status(bug)
|
|
issue_status[num] = min(prev, cur, key=STATUS_PRIORITY.index)
|
|
done_issues = set()
|
|
|
|
for bug in buglog:
|
|
bid = bug.get("id")
|
|
if not bid:
|
|
continue
|
|
|
|
# --since: new issues only for recently seen bugs; status changes
|
|
# (status_updated) are still mirrored for already-synced ones.
|
|
if since or ids:
|
|
recent = False
|
|
if since:
|
|
recent = (bug.get("last_seen") or "") >= since
|
|
if bug.get("gitea_issue"):
|
|
recent = recent or (bug.get("status_updated") or "") >= since
|
|
if ids and bid in ids:
|
|
recent = True
|
|
if not recent:
|
|
skipped += 1
|
|
continue
|
|
|
|
status = bug_status(bug)
|
|
if status is None:
|
|
needs_status.append(bid)
|
|
synced_status = status or "unknown"
|
|
state = bug_state(bug)
|
|
issue_num = bug.get("gitea_issue")
|
|
|
|
if issue_num and bug.get("gitea_synced_status") == synced_status:
|
|
skipped += 1
|
|
continue
|
|
|
|
if issue_num:
|
|
# Status changed since the last sync (or was never recorded:
|
|
# issues synced before the status field existed were closed
|
|
# whenever `fix` was non-empty, so re-assert state for them).
|
|
# The issue itself follows the aggregate status of every entry
|
|
# linked to it; only status labels are touched, so labels a
|
|
# person put on the issue survive.
|
|
agg = issue_status[issue_num]
|
|
agg_state = STATUS_MAP.get(agg, ("open", None))[0]
|
|
if dry_run:
|
|
if issue_num not in done_issues:
|
|
print(f" [DRY-RUN] #{issue_num} ({bid}): -> {agg_state}"
|
|
f" ({agg or 'unknown'})")
|
|
done_issues.add(issue_num)
|
|
continue
|
|
try:
|
|
if issue_num not in done_issues:
|
|
_, issue = request("GET", f"/issues/{issue_num}")
|
|
if issue is None:
|
|
warn(f" {bid}: issue #{issue_num} not found, skipping")
|
|
skipped += 1
|
|
continue
|
|
labels = api.list_labels()
|
|
want = status_label(agg)
|
|
for lbl in issue.get("labels") or []:
|
|
if lbl["name"] in STATUS_LABELS and lbl["name"] != want:
|
|
api.remove_label(issue_num, lbl["id"])
|
|
have = {lbl["name"] for lbl in issue.get("labels") or []}
|
|
if want and want not in have and want in labels:
|
|
api.add_labels(issue_num, [labels[want]["id"]])
|
|
if issue.get("state") != agg_state:
|
|
api.update_issue(issue_num, state=agg_state)
|
|
note = _text(bug.get("status_note"))
|
|
api.add_comment(issue_num, f"{bid} status: **{synced_status}**"
|
|
+ (f" — {note}" if note else ""))
|
|
if agg_state == "closed":
|
|
closed += 1
|
|
else:
|
|
reopened += 1
|
|
if not quiet:
|
|
print(f" #{issue_num} {bid}: {issue['state']} -> "
|
|
f"{agg_state} ({agg or 'unknown'})")
|
|
else:
|
|
updated += 1
|
|
done_issues.add(issue_num)
|
|
set_fields(BUGLOG_PATH, bid, {"gitea_synced_status": synced_status})
|
|
except (RuntimeError, BuglogEditError) as e:
|
|
warn(f" {bid}: could not update issue #{issue_num}: {e}")
|
|
skipped += 1
|
|
continue
|
|
|
|
# New bug -- create issue
|
|
title = bug_title(bug)
|
|
if dry_run:
|
|
print(f" [DRY-RUN] create {state.upper()} issue: {title}")
|
|
created += 1
|
|
continue
|
|
|
|
label_ids = _resolve_label_ids(api, bug_labels(bug))
|
|
issue = api.create_issue(title, bug_body(bug), labels=label_ids)
|
|
if state == "closed":
|
|
# Gitea may ignore `state` on create; close explicitly.
|
|
api.update_issue(issue["number"], state="closed")
|
|
set_fields(BUGLOG_PATH, bid, {"gitea_issue": issue["number"],
|
|
"gitea_synced_status": synced_status})
|
|
if not quiet:
|
|
mark = " (closed)" if state == "closed" else ""
|
|
print(f" created #{issue['number']}: {title}{mark}")
|
|
created += 1
|
|
|
|
if not quiet:
|
|
print(f"Bugs: {created} created, {closed} closed, {reopened} reopened, "
|
|
f"{updated} relabelled, {skipped} skipped")
|
|
if needs_status:
|
|
warn(f"{len(needs_status)} bug(s) have no status -- triage per "
|
|
f".claude/rules/buglog-status.md: {' '.join(needs_status)}")
|
|
|
|
|
|
def _append_top_level_key(path, key, value):
|
|
"""Add `key` as the last member of a JSON file's top-level object without
|
|
re-serializing the rest (spec.json files use hand-formatted inline arrays
|
|
that a json.dumps rewrite would reflow). Falls back to a full rewrite when
|
|
the key already exists or the file doesn't end in a plain '}' line."""
|
|
text = path.read_text()
|
|
data = json.loads(text)
|
|
lines = text.rstrip("\n").split("\n")
|
|
if key in data or lines[-1] != "}":
|
|
data[key] = value
|
|
_write_json(path, data)
|
|
return
|
|
body = json.dumps(value, indent=2, ensure_ascii=False).split("\n")
|
|
last = len(lines) - 2
|
|
if lines[last].strip() and not lines[last].rstrip().endswith(("{", ",")):
|
|
lines[last] = lines[last].rstrip() + ","
|
|
lines[-1:] = [f" {json.dumps(key)}: {body[0]}"] + [" " + b for b in body[1:]] + ["}"]
|
|
out = "\n".join(lines) + "\n"
|
|
if json.loads(out) != {**data, key: value}:
|
|
raise RuntimeError(f"{path}: surgical write did not round-trip")
|
|
path.write_text(out)
|
|
|
|
|
|
# ── JSON Writing ─────────────────────────────────────────────────────────────
|
|
|
|
def _write_json(path, data):
|
|
"""Write JSON back to file, preserving newline style to match existing."""
|
|
text = json.dumps(data, indent=2, ensure_ascii=False)
|
|
# Preserve trailing newline
|
|
if not text.endswith("\n"):
|
|
text += "\n"
|
|
path.write_text(text)
|
|
|
|
|
|
# ── Main ─────────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Sync cc-sdd specs and OpenWolf buglog to Gitea issues"
|
|
)
|
|
parser.add_argument("--dry-run", action="store_true",
|
|
help="Show what would change, don't call API")
|
|
parser.add_argument("--spec", metavar="NAME",
|
|
help="Sync only one spec (by directory name)")
|
|
parser.add_argument("--bugs", action="store_true",
|
|
help="Sync bugs only")
|
|
parser.add_argument("--since", metavar="DATE",
|
|
help="Create issues only for bugs last seen >= DATE; update synced ones changed since (YYYY-MM-DD)")
|
|
parser.add_argument("--ids", metavar="ID[,ID...]",
|
|
help="Also sync these bug ids regardless of --since (comma-separated, e.g. bug-1798,bug-1799)")
|
|
parser.add_argument("--quiet", action="store_true",
|
|
help="Suppress progress output, print only errors")
|
|
args = parser.parse_args()
|
|
|
|
# Validate environment
|
|
if not args.dry_run:
|
|
if not GITEA_TOKEN:
|
|
die("GITEA_TOKEN not set. Use --dry-run to preview without token.")
|
|
api = GiteaAPI()
|
|
ensure_all_labels(api)
|
|
else:
|
|
api = None
|
|
|
|
if not args.quiet:
|
|
print(f"Gitea: {GITEA_URL}/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}")
|
|
if args.dry_run:
|
|
print("Mode: DRY-RUN (no changes)")
|
|
|
|
# Sync specs
|
|
if not args.bugs:
|
|
if args.spec:
|
|
spec_dirs = [args.spec]
|
|
elif SPECS_DIR.exists():
|
|
spec_dirs = [
|
|
d.name for d in SPECS_DIR.iterdir()
|
|
if d.is_dir() and (d / "spec.json").exists()
|
|
]
|
|
else:
|
|
spec_dirs = []
|
|
|
|
for spec_name in sorted(spec_dirs):
|
|
sync_spec(api, spec_name, dry_run=args.dry_run, quiet=args.quiet)
|
|
|
|
# Sync bugs
|
|
if not args.spec:
|
|
ids = ({i.strip() for i in args.ids.split(",") if i.strip()}
|
|
if args.ids else None)
|
|
sync_bugs(api, dry_run=args.dry_run, quiet=args.quiet,
|
|
since=args.since, ids=ids)
|
|
|
|
if not args.quiet:
|
|
if args.dry_run:
|
|
print("Done (dry-run — no changes made)")
|
|
else:
|
|
print("Done")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|