#!/usr/bin/env python3 """Publish a single staging ticket's current content to the production repo. #186: gabriel shares tickets on social media as part of developing in public, but doesn't want to send people to staging infrastructure. This is deliberately one-way and on-demand — not a tracker sync. Given this same session found a phantom-success write bug and an issue-numbering collision in staging's own issue tracking (#182/#183/#184), an always-on bidirectional mirror would just add a second place for that class of bug to bite. Instead: copy one ticket's current content to production, on request, idempotently. Explicitly out of scope: comments, assignees, state transitions, and any kind of two-way sync — see #186 for the full rationale. Usage:: python3 publish_issue.py 185 --json python3 publish_issue.py 185 --production-hub https://musehub.ai --json """ from __future__ import annotations import argparse import json import re import subprocess import sys from typing import Protocol _STAGING_URL_RE = re.compile(r"https?://staging\.musehub\.ai\S*") # Bare cross-references like "#182" — these are staging-only numbers and # won't resolve to the same ticket (or anything at all) on production. # Requires the digits immediately after '#' with no intervening whitespace, # so markdown headings ("## Summary") never match. _BARE_ISSUE_REF_RE = re.compile(r"(? str: """A stable, greppable marker appended to every published body. Used to find an existing production mirror on re-publish, so re-running this tool updates in place instead of creating a duplicate issue. """ return f"_Mirrored from staging#{staging_number}._" class HubClient(Protocol): def read_issue(self, number: int, hub: str) -> dict: ... def list_issues(self, hub: str, state: str = "all") -> list[dict]: ... def create_issue(self, hub: str, *, title: str, body: str, labels: list[str]) -> dict: ... def update_issue(self, hub: str, number: int, *, title: str, body: str) -> dict: ... def set_labels(self, hub: str, number: int, labels: list[str]) -> None: ... class MuseHubCliClient: """Real implementation — shells out to `muse hub issue ...`.""" def _run(self, args: list[str]) -> dict: proc = subprocess.run( ["muse", "hub", *args, "--json"], capture_output=True, text=True, ) if proc.returncode != 0: raise RuntimeError(f"muse hub {' '.join(args)} failed: {proc.stderr.strip()}") return json.loads(proc.stdout) def read_issue(self, number: int, hub: str) -> dict: return self._run(["issue", "read", str(number), "--hub", hub]) def list_issues(self, hub: str, state: str = "all") -> list[dict]: return self._run(["issue", "list", "--hub", hub, "--state", state]).get("issues", []) def create_issue(self, hub: str, *, title: str, body: str, labels: list[str]) -> dict: args = ["issue", "create", "--hub", hub, "--title", title, "--body", body] for label in labels: args += ["--label", label] return self._run(args) def update_issue(self, hub: str, number: int, *, title: str, body: str) -> dict: return self._run([ "issue", "update", str(number), "--hub", hub, "--title", title, "--body", body, ]) def set_labels(self, hub: str, number: int, labels: list[str]) -> None: if not labels: return self._run(["issue", "label", str(number), "--hub", hub, "--set", *labels]) def find_existing_mirror(client: HubClient, production_hub: str, staging_number: int) -> dict | None: marker = build_mirror_marker(staging_number) for issue in client.list_issues(production_hub, state="all"): if marker in (issue.get("body") or ""): return issue return None def publish_issue( client: HubClient, staging_number: int, *, staging_hub: str, production_hub: str, ) -> dict: """Copy staging issue ``staging_number``'s current content to ``production_hub``. Idempotent: a second call for the same ``staging_number`` updates the previously-created mirror in place rather than creating a duplicate. """ src = client.read_issue(staging_number, staging_hub) warnings: list[str] = [] body_text = src.get("body") or "" if _STAGING_URL_RE.search(body_text): warnings.append( "Source body contains a staging.musehub.ai URL — review before sharing publicly." ) bare_refs = sorted(set(_BARE_ISSUE_REF_RE.findall(body_text)), key=lambda r: int(r[1:])) if bare_refs: warnings.append( f"Source body contains bare issue references ({', '.join(bare_refs)}) that may " "not resolve to the same tickets on production — review before sharing publicly." ) marker = build_mirror_marker(staging_number) mirrored_body = f"{src['body']}\n\n{marker}" labels = src.get("labels", []) existing = find_existing_mirror(client, production_hub, staging_number) if existing is not None: client.update_issue(production_hub, existing["number"], title=src["title"], body=mirrored_body) client.set_labels(production_hub, existing["number"], labels) return { "action": "updated", "number": existing["number"], "url": existing.get("url"), "warnings": warnings, } created = client.create_issue(production_hub, title=src["title"], body=mirrored_body, labels=labels) return { "action": "created", "number": created["number"], "url": created.get("url"), "warnings": warnings, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("staging_number", type=int, help="Issue number on the staging hub to publish.") parser.add_argument("--staging-hub", default=DEFAULT_STAGING_HUB) parser.add_argument("--production-hub", default=DEFAULT_PRODUCTION_HUB) parser.add_argument("--json", action="store_true", help="Emit JSON (the only supported output form).") args = parser.parse_args(argv) client = MuseHubCliClient() try: result = publish_issue( client, args.staging_number, staging_hub=args.staging_hub, production_hub=args.production_hub, ) except (RuntimeError, KeyError) as e: print(f"❌ {e}", file=sys.stderr) return 1 for w in result["warnings"]: print(f"⚠️ {w}", file=sys.stderr) print(json.dumps(result)) return 0 if __name__ == "__main__": sys.exit(main())