from __future__ import annotations
import argparse
import json
import os
import pathlib
import subprocess
import sys
from typing import Any
ROOT = pathlib.Path(__file__).resolve().parents[1]
DEFAULT_LABELS = ["sdk-contract", "lockstep", "automated"]
def run(cmd: list[str], *, input_text: str | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
proc = subprocess.run(cmd, input=input_text, text=True, capture_output=True, check=False)
if check and proc.returncode != 0:
raise RuntimeError(f"command failed: {' '.join(cmd)}\nSTDOUT:\n{proc.stdout}\nSTDERR:\n{proc.stderr}")
return proc
def load_contract(path: pathlib.Path) -> dict[str, Any]:
return json.loads(path.read_text())
def changed_sections(before: dict[str, Any] | None, after: dict[str, Any]) -> list[str]:
if before is None:
return sorted(after.keys())
keys = sorted(set(before) | set(after))
return [key for key in keys if before.get(key) != after.get(key)]
def existing_issue(repo: str, title: str, labels: list[str]) -> int | None:
query = ["gh", "issue", "list", "--repo", repo, "--state", "open", "--search", title, "--json", "number,title,labels"]
proc = run(query)
for issue in json.loads(proc.stdout):
if issue.get("title") != title:
continue
issue_labels = {label.get("name") for label in issue.get("labels", [])}
if set(labels).issubset(issue_labels):
return int(issue["number"])
return None
def ensure_labels(repo: str, labels: list[str], dry_run: bool) -> None:
for label in labels:
if dry_run:
print(f"dry-run: ensure label {repo}:{label}")
continue
run(["gh", "label", "create", label, "--repo", repo, "--color", "5319e7", "--description", "SDK contract lockstep automation"], check=False)
def open_or_update(repo: str, title: str, body: str, labels: list[str], dry_run: bool) -> None:
ensure_labels(repo, labels, dry_run)
if dry_run:
print(f"dry-run: would open/update issue in {repo}: {title}\n{body}\n")
return
number = existing_issue(repo, title, labels)
if number is None:
cmd = ["gh", "issue", "create", "--repo", repo, "--title", title, "--body-file", "-"]
for label in labels:
cmd.extend(["--label", label])
run(cmd, input_text=body)
else:
run(["gh", "issue", "comment", str(number), "--repo", repo, "--body-file", "-"], input_text=body)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--contract", default="schema/sdk-contract.json")
ap.add_argument("--previous-contract", help="Optional previous sdk-contract.json for changed-section summary")
ap.add_argument("--tag-or-sha", default=os.environ.get("GITHUB_REF_NAME") or os.environ.get("GITHUB_SHA") or "unknown")
ap.add_argument("--py-repo", default="styrene-lab/omegon-extension-py")
ap.add_argument("--ts-repo", default="styrene-lab/omegon-extension-ts")
ap.add_argument("--dry-run", action="store_true")
args = ap.parse_args()
contract_path = (ROOT / args.contract).resolve()
after = load_contract(contract_path)
before = load_contract(pathlib.Path(args.previous_contract)) if args.previous_contract else None
sections = changed_sections(before, after)
version = after.get("sdk_contract_version", "unknown")
title = f"Align SDK contract with omegon-extension-rs {args.tag_or_sha}"
body = f"""The canonical Rust SDK contract changed or was released.
Rust SDK reference: `{args.tag_or_sha}`
Contract version: `{version}`
Canonical artifact: `omegon-extension-rs/schema/sdk-contract.json`
Changed top-level sections:
{chr(10).join(f'- `{section}`' for section in sections)}
Required downstream work:
1. Copy or regenerate the SDK contract artifact from the Rust SDK.
2. Update exported constants/types/helpers to match the contract.
3. Run the downstream validation suite.
4. Run the cross-repo lockstep check if available.
Validation commands:
```bash
# Python SDK
.venv/bin/pytest -q
# TypeScript SDK
npm test
```
This issue is generated by the Rust SDK downstream lockstep workflow. It is idempotent by title + labels.
"""
for repo in [args.py_repo, args.ts_repo]:
open_or_update(repo, title, body, DEFAULT_LABELS, args.dry_run)
return 0
if __name__ == "__main__":
raise SystemExit(main())