forked from lei.chen/myxce
feat(tooling): add cc-sdd to Gitea sync skill
Syncs Kiro specs to Gitea milestones + one issue per phase, and buglog.json entries to Gitea issues. Idempotent (safe to re-run). Validated against nexa/xce: admin-config-tool spec + 14 bugs synced.
This commit is contained in:
parent
18e7a0e9e1
commit
bf6a9b3405
2 changed files with 1073 additions and 0 deletions
198
.claude/skills/gitea-sync/SKILL.md
Normal file
198
.claude/skills/gitea-sync/SKILL.md
Normal file
|
|
@ -0,0 +1,198 @@
|
|||
---
|
||||
name: gitea-sync
|
||||
description: Sync cc-sdd specs and OpenWolf buglog to Gitea milestones/issues in nexa/xce
|
||||
argument-hint: [--dry-run] [--spec <name>] [--bugs] [--since <date>]
|
||||
triggers:
|
||||
- /gitea-sync
|
||||
- sync to gitea
|
||||
- gitea sync
|
||||
- synchronize specs to gitea
|
||||
---
|
||||
|
||||
# Gitea Sync Skill
|
||||
|
||||
Sync cc-sdd specs (`.kiro/specs/`) and OpenWolf buglog (`.wolf/buglog.json`) to
|
||||
Gitea issues in `nexa/xce`. Files are ground truth — Gitea is a read-only mirror.
|
||||
Sync state (issue numbers) is stored in source files for idempotency.
|
||||
|
||||
## Invocation
|
||||
|
||||
```
|
||||
/gitea-sync # sync all specs + all bugs
|
||||
/gitea-sync --spec admin-config-tool # sync one spec
|
||||
/gitea-sync --bugs --since 2026-07-18 # sync recent bugs only
|
||||
/gitea-sync --dry-run # preview without making changes
|
||||
```
|
||||
|
||||
## Procedure
|
||||
|
||||
### 0. Parse arguments
|
||||
|
||||
- `--dry-run`: only report what would change, don't call MCP write tools
|
||||
- `--spec <name>`: sync only one spec directory
|
||||
- `--bugs`: sync bugs only (skip specs)
|
||||
- `--since <YYYY-MM-DD>`: only sync bugs with `last_seen >=` this date
|
||||
|
||||
### 1. Ensure labels exist
|
||||
|
||||
These labels must exist in `nexa/xce`. Use `mcp__gitea__label_read` to list
|
||||
existing labels, then `mcp__gitea__label_write` with method `create_repo_label`
|
||||
for any missing ones:
|
||||
|
||||
| Name | Color | Description |
|
||||
|------|-------|-------------|
|
||||
| `phase:requirements` | `6C5CE7` | Spec requirements phase |
|
||||
| `phase:design` | `6C5CE7` | Spec design phase |
|
||||
| `phase:tasks` | `6C5CE7` | Spec tasks phase |
|
||||
| `phase:implementation` | `F39C12` | Spec implementation phase |
|
||||
| `phase:complete` | `27AE60` | Spec complete |
|
||||
| `spec` | `3498DB` | Spec-tracked issue |
|
||||
| `bug` | `E74C3C` | Bug-tracked issue |
|
||||
| `bug:fixed` | `27AE60` | Bug resolved |
|
||||
| `blocked` | `E74C3C` | Blocked on dependency |
|
||||
|
||||
### 2. Sync Specs
|
||||
|
||||
For each spec directory in `.kiro/specs/` (or just `--spec <name>`):
|
||||
|
||||
1. Read `spec.json`. If it has `gitea.milestone_id` and `gitea.issues` with
|
||||
all three keys (`requirements_design`, `implementation`, `validation`),
|
||||
skip creation — the spec is already synced. Check phase states and
|
||||
update issue open/closed state if needed (step 2d).
|
||||
|
||||
2. If not synced:
|
||||
a. **Create milestone**: Use `mcp__gitea__milestone_write` method `create`.
|
||||
Title = spec name. Description = one-line summary + phase + blockers.
|
||||
b. **Create three phase issues** under that milestone via
|
||||
`mcp__gitea__issue_write` method `create`:
|
||||
|
||||
| Key | Title | Labels | Initial State |
|
||||
|-----|-------|--------|---------------|
|
||||
| `requirements_design` | `[{spec}] Requirements & Design` | spec, phase:requirements, phase:design | closed if both approved, else open |
|
||||
| `implementation` | `[{spec}] Implementation` | spec, phase:implementation | open |
|
||||
| `validation` | `[{spec}] Validation & Closure` | spec, phase:complete | open |
|
||||
|
||||
Body format for each issue (build from spec files):
|
||||
|
||||
**Requirements & Design body**: Phase status, requirements summary table
|
||||
(extract `## Requirements Summary` from `requirements.md`), blockers list,
|
||||
link to spec files.
|
||||
|
||||
**Implementation body**: Tasks generated/ready status, task checklist
|
||||
(extract `- [ ]` / `- [x]` from `tasks.md` if it exists), blockers list.
|
||||
|
||||
**Validation body**: Acceptance gate checklist, deferred future-spec items.
|
||||
|
||||
c. **Write sync state** back to `spec.json`:
|
||||
```json
|
||||
"gitea": {
|
||||
"milestone_id": <id>,
|
||||
"issues": {
|
||||
"requirements_design": <number>,
|
||||
"implementation": <number>,
|
||||
"validation": <number>
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
d. **Phase state updates** (for already-synced specs): Check `spec.json`
|
||||
approvals. If requirements AND design are approved → close the
|
||||
requirements_design issue via `mcp__gitea__issue_write` method `update`
|
||||
with `state: "closed"`. If implementation approved → close
|
||||
implementation and validation issues. If all approved → close milestone
|
||||
via `mcp__gitea__milestone_write` method `update`.
|
||||
|
||||
### 3. Sync Bugs
|
||||
|
||||
Read `.wolf/buglog.json`. For each entry:
|
||||
|
||||
1. **Skip** if:
|
||||
- Entry has `gitea_issue` field (already synced) AND bug was already fixed
|
||||
when synced (no state change needed).
|
||||
- Filter: if `--since <date>` provided, skip entries with
|
||||
`last_seen < <date>`.
|
||||
|
||||
2. **Create issue** for entries without `gitea_issue`:
|
||||
Use `mcp__gitea__issue_write` method `create`.
|
||||
|
||||
**Title**: `[<bug-id>] <first sentence of error_message>` (max 120 chars)
|
||||
|
||||
**Body**:
|
||||
```
|
||||
## Error
|
||||
<error_message>
|
||||
|
||||
## Root Cause
|
||||
<root_cause>
|
||||
|
||||
## Fix
|
||||
<fix> (or "*Not yet fixed*" if empty)
|
||||
|
||||
## Affected Files
|
||||
- <file> (split on commas)
|
||||
|
||||
## Tags
|
||||
`tag1` `tag2` ...
|
||||
|
||||
## Related Bugs
|
||||
- <related>
|
||||
|
||||
---
|
||||
*Synced from `.wolf/buglog.json`*
|
||||
```
|
||||
|
||||
**Labels**: Start with `["bug"]`. If `fix` is populated, add `bug:fixed`.
|
||||
Map known tags to labels: `asan`/`ubsan` → create `bug:asan` label,
|
||||
`SEGV`/`double-free`/`UAF` → `bug:crash`, `race`/`toctou` → `bug:race`,
|
||||
`ha`/`raft`/`election`/`failover` → `bug:ha`, `data-loss` → `bug:data-loss`.
|
||||
|
||||
**State**: `"closed"` if `fix` is populated, `"open"` otherwise.
|
||||
|
||||
**Milestone**: None (bugs span specs).
|
||||
|
||||
3. **Close fixed bugs**: For entries WITH `gitea_issue` where `fix` is now
|
||||
populated (was empty before — we track by checking if the Gitea issue is
|
||||
still open). Use `mcp__gitea__issue_write` method `update` with
|
||||
`state: "closed"`.
|
||||
|
||||
4. **Write `gitea_issue`** back to the buglog entry: add
|
||||
`"gitea_issue": <issue_number>` to the entry in `.wolf/buglog.json`.
|
||||
Read the full file, modify the matching entry, write back with
|
||||
`json.dumps(data, indent=2)`.
|
||||
|
||||
### 4. Report
|
||||
|
||||
Print summary: "Specs: N synced, M skipped | Bugs: X created, Y closed, Z skipped"
|
||||
|
||||
## Important Rules
|
||||
|
||||
- **Files are ground truth**. Never read Gitea state to update local files.
|
||||
The gitea_issue/milestone_id fields in local files only store sync mappings.
|
||||
- **Idempotent**. Items with existing `gitea_issue` or `gitea` state are skipped
|
||||
unless their phase/fix state changed.
|
||||
- **Labels**: Check existing labels before creating. Use label IDs (not names)
|
||||
in issue create calls — fetch with `mcp__gitea__label_read` first.
|
||||
- **Rate limiting**: Batch issue creation — create issues sequentially
|
||||
(a few per turn) to stay within API limits.
|
||||
- **Buglog is large** (~13K lines). Always use `--since` filter for
|
||||
practical runs (e.g., `--since 2026-07-18`). Without `--since`, sync ALL bugs.
|
||||
- **Private repo**: `nexa/xce` is private. MCP auth must have access.
|
||||
|
||||
## Gitea MCP Tool Reference
|
||||
|
||||
| Action | Tool | Method |
|
||||
|--------|------|--------|
|
||||
| List labels | `mcp__gitea__label_read` | `list_repo_labels` |
|
||||
| Create label | `mcp__gitea__label_write` | `create_repo_label` |
|
||||
| List milestones | `mcp__gitea__milestone_read` | `list` |
|
||||
| Create milestone | `mcp__gitea__milestone_write` | `create` |
|
||||
| List issues | `mcp__gitea__issue_read` | `get` (single) or use list |
|
||||
| Create issue | `mcp__gitea__issue_write` | `create` |
|
||||
| Update issue | `mcp__gitea__issue_write` | `update` |
|
||||
|
||||
For listing issues by milestone, use `mcp__gitea__issue_read` with `method: "get"`
|
||||
and `state: "all"`. The Gitea MCP may have limited filtering — for bulk listing,
|
||||
check the milestone's open_issues/closed_issues counters.
|
||||
|
||||
Note: issue `state: "closed"` may not work on `create` — if the issue is created
|
||||
as `"open"`, immediately update it with `method: "update"` + `state: "closed"`.
|
||||
875
scripts/gitea-sync.py
Executable file
875
scripts/gitea-sync.py
Executable file
|
|
@ -0,0 +1,875 @@
|
|||
#!/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
|
||||
|
||||
|
||||
# ── 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"},
|
||||
# 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",
|
||||
}
|
||||
|
||||
|
||||
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) as resp:
|
||||
return resp.status, json.loads(resp.read().decode("utf-8"))
|
||||
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
|
||||
|
||||
|
||||
# ── 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, labels)."""
|
||||
request("PATCH", f"/issues/{issue_number}", fields)
|
||||
|
||||
|
||||
# ── 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 bug_labels(bug_entry):
|
||||
"""Derive Gitea labels from buglog tags."""
|
||||
labels = ["bug"]
|
||||
tags = [t.lower() for t in bug_entry.get("tags", [])]
|
||||
|
||||
# Fixed bugs get bug:fixed label
|
||||
fix = bug_entry.get("fix", "")
|
||||
if fix and fix.strip():
|
||||
labels.append("bug:fixed")
|
||||
|
||||
# Map known tag patterns to labels
|
||||
for tag in tags:
|
||||
for pattern, label in TAG_TO_LABEL.items():
|
||||
if pattern.lower() in tag:
|
||||
if label not in labels:
|
||||
labels.append(label)
|
||||
|
||||
# Ensure any needed labels exist in Gitea
|
||||
# (bug:asan, bug:crash, bug:data-loss, bug:ha, bug:race may not exist yet)
|
||||
extra_labels = {
|
||||
"bug:asan": ("9B59B6", "ASAN/UBSAN caught bug"),
|
||||
"bug:crash": ("C0392B", "SEGV/UAF/double-free crash"),
|
||||
"bug:data-loss": ("C0392B", "Committed data loss bug"),
|
||||
"bug:ha": ("E67E22", "HA/Raft/consensus bug"),
|
||||
"bug:race": ("F39C12", "Data race / TOCTOU bug"),
|
||||
}
|
||||
for lbl in labels:
|
||||
if lbl in extra_labels and lbl not in LABEL_NAMES:
|
||||
pass # handled below
|
||||
|
||||
return labels, extra_labels
|
||||
|
||||
|
||||
LABEL_NAMES = {l["name"] for l in LABELS}
|
||||
|
||||
|
||||
# ── Body Formatting ──────────────────────────────────────────────────────────
|
||||
|
||||
def bug_body(bug_entry):
|
||||
"""Format a buglog entry as a Gitea issue body."""
|
||||
parts = []
|
||||
|
||||
error = bug_entry.get("error_message", "")
|
||||
if error:
|
||||
parts.append("## Error\n")
|
||||
parts.append(error.strip())
|
||||
parts.append("")
|
||||
|
||||
root_cause = bug_entry.get("root_cause", "")
|
||||
if root_cause:
|
||||
parts.append("## Root Cause\n")
|
||||
parts.append(root_cause.strip())
|
||||
parts.append("")
|
||||
|
||||
fix = bug_entry.get("fix", "")
|
||||
if fix and fix.strip():
|
||||
parts.append("## Fix\n")
|
||||
parts.append(fix.strip())
|
||||
parts.append("")
|
||||
else:
|
||||
parts.append("## Fix\n")
|
||||
parts.append("*Not yet fixed*")
|
||||
parts.append("")
|
||||
|
||||
file_ = bug_entry.get("file", "")
|
||||
if file_:
|
||||
parts.append("## Affected Files\n")
|
||||
for f in file_.split(","):
|
||||
f = f.strip()
|
||||
if f:
|
||||
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 = bug_entry.get("error_message", "")
|
||||
# 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_approved = spec_data.get("approvals", {}).get("tasks", {}).get("approved", False)
|
||||
ready = spec_data.get("ready_for_implementation", False)
|
||||
parts.append(f"**Tasks generated**: {tasks_approved}")
|
||||
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 "complete" in phase or (
|
||||
approvals.get("implementation") and
|
||||
approvals["implementation"].get("approved")
|
||||
):
|
||||
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,
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
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']}"
|
||||
|
||||
# Determine if this phase issue should start closed
|
||||
state = "open"
|
||||
if phase_def["key"] == "requirements_design":
|
||||
approvals = spec_data.get("approvals", {})
|
||||
req_ok = approvals.get("requirements", {}).get("approved", False)
|
||||
des_ok = approvals.get("design", {}).get("approved", False)
|
||||
if req_ok and des_ok:
|
||||
state = "closed"
|
||||
|
||||
# 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,
|
||||
}
|
||||
_write_json(spec_json_path, spec_data)
|
||||
|
||||
|
||||
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
|
||||
|
||||
approvals = spec_data.get("approvals", {})
|
||||
req_ok = approvals.get("requirements", {}).get("approved", False)
|
||||
des_ok = approvals.get("design", {}).get("approved", False)
|
||||
tasks_ok = approvals.get("tasks", {}).get("approved", False)
|
||||
impl_ok = approvals.get("implementation", {}).get("approved", False)
|
||||
|
||||
issues = gitea_state.get("issues", {})
|
||||
|
||||
# Build desired states for each phase issue
|
||||
desired = {}
|
||||
rd_issue = issues.get("requirements_design")
|
||||
if rd_issue:
|
||||
desired[rd_issue] = "closed" if (req_ok and des_ok) else "open"
|
||||
impl_issue = issues.get("implementation")
|
||||
if impl_issue:
|
||||
desired[impl_issue] = "closed" if impl_ok else "open"
|
||||
val_issue = issues.get("validation")
|
||||
if val_issue:
|
||||
desired[val_issue] = "closed" if impl_ok else "open"
|
||||
|
||||
# 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
|
||||
all_approved = req_ok and des_ok and tasks_ok and impl_ok
|
||||
milestone_id = gitea_state.get("milestone_id")
|
||||
if all_approved and milestone_id:
|
||||
all_milestones = api.list_milestones(state="all")
|
||||
ms = all_milestones.get(spec_name)
|
||||
if ms and ms.get("state") != "closed":
|
||||
api.update_milestone(milestone_id, state="closed")
|
||||
if not quiet:
|
||||
print(f" closed milestone '{spec_name}'")
|
||||
|
||||
|
||||
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):
|
||||
"""Sync buglog entries to Gitea issues.
|
||||
|
||||
Args:
|
||||
since: Only sync bugs with last_seen >= this date (ISO format: YYYY-MM-DD).
|
||||
Helps avoid flooding Gitea with ancient auto-detected bugs.
|
||||
"""
|
||||
if not BUGLOG_PATH.exists():
|
||||
warn(f"buglog not found: {BUGLOG_PATH}")
|
||||
return
|
||||
|
||||
buglog = json.loads(BUGLOG_PATH.read_text())
|
||||
created = 0
|
||||
closed = 0
|
||||
skipped = 0
|
||||
|
||||
for i, bug in enumerate(buglog):
|
||||
bid = bug.get("id", f"bug-{i:04d}")
|
||||
|
||||
# Apply --since filter
|
||||
if since:
|
||||
last_seen = bug.get("last_seen", "")
|
||||
if last_seen < since:
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
gitea_issue_num = bug.get("gitea_issue")
|
||||
|
||||
if gitea_issue_num:
|
||||
# Already synced — check if state changed
|
||||
fix_now = bool(bug.get("fix", "").strip())
|
||||
# We don't track previous fix state, so check Gitea
|
||||
try:
|
||||
_, issue = request("GET", f"/issues/{gitea_issue_num}")
|
||||
if issue and issue.get("state") == "open" and fix_now:
|
||||
if dry_run:
|
||||
print(f" [DRY-RUN] close bug #{gitea_issue_num}: {bid}")
|
||||
else:
|
||||
api.update_issue(gitea_issue_num, state="closed")
|
||||
# Add bug:fixed label via API
|
||||
request(
|
||||
"PUT",
|
||||
f"/issues/{gitea_issue_num}/labels",
|
||||
{"labels": [api.list_labels().get("bug:fixed", {}).get("id", 0)]}
|
||||
)
|
||||
if not quiet:
|
||||
print(f" closed #{gitea_issue_num}: {bid}")
|
||||
closed += 1
|
||||
else:
|
||||
skipped += 1
|
||||
except Exception as e:
|
||||
warn(f" {bid}: could not check issue #{gitea_issue_num}: {e}")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
# New bug — create issue
|
||||
title = bug_title(bug)
|
||||
body = bug_body(bug)
|
||||
fix = bug.get("fix", "").strip()
|
||||
state = "closed" if fix else "open"
|
||||
|
||||
if dry_run:
|
||||
print(f" [DRY-RUN] create {'CLOSED' if state == 'closed' else 'OPEN'} "
|
||||
f"issue: {title}")
|
||||
created += 1
|
||||
continue
|
||||
|
||||
# Determine labels
|
||||
labels, extra = bug_labels(bug)
|
||||
for ename, (ecolor, edesc) in extra.items():
|
||||
if ename in labels:
|
||||
api.ensure_label(ename, ecolor, edesc)
|
||||
|
||||
label_ids = _resolve_label_ids(api, labels)
|
||||
|
||||
issue = api.create_issue(title, body, labels=label_ids, state=state)
|
||||
bug["gitea_issue"] = issue["number"]
|
||||
state_mark = " (closed)" if state == "closed" else ""
|
||||
if not quiet:
|
||||
print(f" created #{issue['number']}: {title}{state_mark}")
|
||||
created += 1
|
||||
|
||||
# Write back updated buglog if any new issues were created
|
||||
if created > 0 and not dry_run:
|
||||
_write_json(BUGLOG_PATH, buglog)
|
||||
|
||||
if not quiet:
|
||||
print(f"Bugs: {created} created, {closed} closed, {skipped} skipped")
|
||||
|
||||
|
||||
# ── 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="Only sync bugs with last_seen >= DATE (YYYY-MM-DD)")
|
||||
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:
|
||||
sync_bugs(api, dry_run=args.dry_run, quiet=args.quiet, since=args.since)
|
||||
|
||||
if not args.quiet:
|
||||
if args.dry_run:
|
||||
print("Done (dry-run — no changes made)")
|
||||
else:
|
||||
print("Done")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue