#!/usr/bin/env python3 """ xce-gitea-sync: Sync cc-sdd specs and OpenWolf buglog to Gitea issues. Reads .kiro/specs// (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 # 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()