Rongsong Shen
643741fad5
sync_bugs() read buglog.json as a bare list and wrote it back the
same way, but the real format is {"version": 1, "bugs": [...]}. Any
real run would have crashed on the first bug entry (enumerate() over
a dict yields its keys, not bug objects) and, had it somehow gotten
past that, a write-back would have destroyed the wrapper. Never
actually run against production Gitea, per the buglog entry.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
877 lines
30 KiB
Python
Executable file
877 lines
30 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
|
|
|
|
|
|
# ── 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_wrapper = json.loads(BUGLOG_PATH.read_text())
|
|
buglog = buglog_wrapper["bugs"]
|
|
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:
|
|
buglog_wrapper["bugs"] = buglog
|
|
_write_json(BUGLOG_PATH, buglog_wrapper)
|
|
|
|
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()
|