diff --git a/skills/.curated/gh-address-comments/SKILL.md b/skills/.curated/gh-address-comments/SKILL.md new file mode 100644 index 0000000..4dd7029 --- /dev/null +++ b/skills/.curated/gh-address-comments/SKILL.md @@ -0,0 +1,23 @@ +--- +name: gh-address-comments +description: Help address review/issue comments on the open GitHub PR for the current branch using gh CLI; verify gh auth first and prompt the user to run oai_gh if not logged in. +--- + +# PR Comment Handler + +Guide to find the open PR for the current branch and address its comments with gh CLI. Run all `gh` commands with elevated network access. + +Prereq: run `oai_gh` once to authenticate, then run `gh auth status` with escalated permissions (include workflow/repo scopes) so `gh` commands succeed. If sandboxing blocks `gh auth status`, rerun it with `sandbox_permissions=require_escalated`. + +## 1) Inspect comments needing attention +- Run scripts/fetch_comments.py which will print out all the comments and review threads on the PR + +## 2) Ask the user for clarification +- Number all the review threads and comments and provide a short summary of what would be required to apply a fix for it +- Ask the user which numbered comments should be addressed + +## 3) If user chooses comments +- Apply fixes for the selected comments + +Notes: +- If gh hits auth/rate issues mid-run, prompt the user to re-run `oai_gh`, then retry. diff --git a/skills/.curated/gh-address-comments/scripts/fetch_comments.py b/skills/.curated/gh-address-comments/scripts/fetch_comments.py new file mode 100644 index 0000000..9081625 --- /dev/null +++ b/skills/.curated/gh-address-comments/scripts/fetch_comments.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +""" +Fetch all PR conversation comments + reviews + review threads (inline threads) +for the PR associated with the current git branch, by shelling out to: + + gh api graphql + +Requires: + - `gh auth login` already set up + - current branch has an associated (open) PR + +Usage: + python fetch_comments.py > pr_comments.json +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from typing import Any + +QUERY = """\ +query( + $owner: String!, + $repo: String!, + $number: Int!, + $commentsCursor: String, + $reviewsCursor: String, + $threadsCursor: String +) { + repository(owner: $owner, name: $repo) { + pullRequest(number: $number) { + number + url + title + state + + # Top-level "Conversation" comments (issue comments on the PR) + comments(first: 100, after: $commentsCursor) { + pageInfo { hasNextPage endCursor } + nodes { + id + body + createdAt + updatedAt + author { login } + } + } + + # Review submissions (Approve / Request changes / Comment), with body if present + reviews(first: 100, after: $reviewsCursor) { + pageInfo { hasNextPage endCursor } + nodes { + id + state + body + submittedAt + author { login } + } + } + + # Inline review threads (grouped), includes resolved state + reviewThreads(first: 100, after: $threadsCursor) { + pageInfo { hasNextPage endCursor } + nodes { + id + isResolved + isOutdated + path + line + diffSide + startLine + startDiffSide + originalLine + originalStartLine + resolvedBy { login } + comments(first: 100) { + nodes { + id + body + createdAt + updatedAt + author { login } + } + } + } + } + } + } +} +""" + + +def _run(cmd: list[str], stdin: str | None = None) -> str: + p = subprocess.run(cmd, input=stdin, capture_output=True, text=True) + if p.returncode != 0: + raise RuntimeError(f"Command failed: {' '.join(cmd)}\n{p.stderr}") + return p.stdout + + +def _run_json(cmd: list[str], stdin: str | None = None) -> dict[str, Any]: + out = _run(cmd, stdin=stdin) + try: + return json.loads(out) + except json.JSONDecodeError as e: + raise RuntimeError(f"Failed to parse JSON from command output: {e}\nRaw:\n{out}") from e + + +def _ensure_gh_authenticated() -> None: + try: + _run(["gh", "auth", "status"]) + except RuntimeError: + print("use oai_gh to authenticate gh cli", file=sys.stderr) + raise RuntimeError("gh auth status failed; use oai_gh to authenticate gh cli") from None + + +def gh_pr_view_json(fields: str) -> dict[str, Any]: + # fields is a comma-separated list like: "number,headRepositoryOwner,headRepository" + return _run_json(["gh", "pr", "view", "--json", fields]) + + +def get_current_pr_ref() -> tuple[str, str, int]: + """ + Resolve the PR for the current branch (whatever gh considers associated). + Works for cross-repo PRs too, by reading head repository owner/name. + """ + pr = gh_pr_view_json("number,headRepositoryOwner,headRepository") + owner = pr["headRepositoryOwner"]["login"] + repo = pr["headRepository"]["name"] + number = int(pr["number"]) + return owner, repo, number + + +def gh_api_graphql( + owner: str, + repo: str, + number: int, + comments_cursor: str | None = None, + reviews_cursor: str | None = None, + threads_cursor: str | None = None, +) -> dict[str, Any]: + """ + Call `gh api graphql` using -F variables, avoiding JSON blobs with nulls. + Query is passed via stdin using query=@- to avoid shell newline/quoting issues. + """ + cmd = [ + "gh", + "api", + "graphql", + "-F", + "query=@-", + "-F", + f"owner={owner}", + "-F", + f"repo={repo}", + "-F", + f"number={number}", + ] + if comments_cursor: + cmd += ["-F", f"commentsCursor={comments_cursor}"] + if reviews_cursor: + cmd += ["-F", f"reviewsCursor={reviews_cursor}"] + if threads_cursor: + cmd += ["-F", f"threadsCursor={threads_cursor}"] + + return _run_json(cmd, stdin=QUERY) + + +def fetch_all(owner: str, repo: str, number: int) -> dict[str, Any]: + conversation_comments: list[dict[str, Any]] = [] + reviews: list[dict[str, Any]] = [] + review_threads: list[dict[str, Any]] = [] + + comments_cursor: str | None = None + reviews_cursor: str | None = None + threads_cursor: str | None = None + + pr_meta: dict[str, Any] | None = None + + while True: + payload = gh_api_graphql( + owner=owner, + repo=repo, + number=number, + comments_cursor=comments_cursor, + reviews_cursor=reviews_cursor, + threads_cursor=threads_cursor, + ) + + if "errors" in payload and payload["errors"]: + raise RuntimeError(f"GitHub GraphQL errors:\n{json.dumps(payload['errors'], indent=2)}") + + pr = payload["data"]["repository"]["pullRequest"] + if pr_meta is None: + pr_meta = { + "number": pr["number"], + "url": pr["url"], + "title": pr["title"], + "state": pr["state"], + "owner": owner, + "repo": repo, + } + + c = pr["comments"] + r = pr["reviews"] + t = pr["reviewThreads"] + + conversation_comments.extend(c.get("nodes") or []) + reviews.extend(r.get("nodes") or []) + review_threads.extend(t.get("nodes") or []) + + comments_cursor = c["pageInfo"]["endCursor"] if c["pageInfo"]["hasNextPage"] else None + reviews_cursor = r["pageInfo"]["endCursor"] if r["pageInfo"]["hasNextPage"] else None + threads_cursor = t["pageInfo"]["endCursor"] if t["pageInfo"]["hasNextPage"] else None + + if not (comments_cursor or reviews_cursor or threads_cursor): + break + + assert pr_meta is not None + return { + "pull_request": pr_meta, + "conversation_comments": conversation_comments, + "reviews": reviews, + "review_threads": review_threads, + } + + +def main() -> None: + _ensure_gh_authenticated() + owner, repo, number = get_current_pr_ref() + result = fetch_all(owner, repo, number) + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/skills/.curated/gh-fix-ci/SKILL.md b/skills/.curated/gh-fix-ci/SKILL.md new file mode 100644 index 0000000..6300956 --- /dev/null +++ b/skills/.curated/gh-fix-ci/SKILL.md @@ -0,0 +1,69 @@ +--- +name: gh-fix-ci +description: Inspect GitHub PR checks with gh, pull failing GitHub Actions logs, summarize failure context, then create a fix plan and implement after user approval. Use when a user asks to debug or fix failing PR CI/CD checks on GitHub Actions and wants a plan + code changes; for external checks (e.g., Buildkite), only report the details URL and mark them out of scope. +--- + +# Gh Pr Checks Plan Fix + +## Overview + +Use gh to locate failing PR checks, fetch GitHub Actions logs for actionable failures, summarize the failure snippet, then propose a fix plan and implement after explicit approval. +- Depends on the `plan` skill for drafting and approving the fix plan. + +Prereq: run `oai_gh` once to authenticate, then run `gh auth status` with escalated permissions (include workflow/repo scopes) so `gh` commands succeed. If sandboxing blocks `gh auth status`, rerun it with `sandbox_permissions=require_escalated`. + +## Inputs + +- `repo`: path inside the repo (default `.`) +- `pr`: PR number or URL (optional; defaults to current branch PR) +- `gh` authentication for the repo host + +## Quick start + +- `python "/scripts/inspect_pr_checks.py" --repo "." --pr ""` +- Add `--json` if you want machine-friendly output for summarization. + +## Workflow + +1. Verify gh authentication. + - Run `gh auth status` in the repo with escalated scopes (workflow/repo) after running `oai_gh`. + - If sandboxed auth status fails, rerun the command with `sandbox_permissions=require_escalated` to allow network/keyring access. + - If unauthenticated, ask the user to log in before proceeding. +2. Resolve the PR. + - Prefer the current branch PR: `gh pr view --json number,url`. + - If the user provides a PR number or URL, use that directly. +3. Inspect failing checks (GitHub Actions only). + - Preferred: run the bundled script (handles gh field drift and job-log fallbacks): + - `python "/scripts/inspect_pr_checks.py" --repo "." --pr ""` + - Add `--json` for machine-friendly output. + - Manual fallback: + - `gh pr checks --json name,state,bucket,link,startedAt,completedAt,workflow` + - If a field is rejected, rerun with the available fields reported by `gh`. + - For each failing check, extract the run id from `detailsUrl` and run: + - `gh run view --json name,workflowName,conclusion,status,url,event,headBranch,headSha` + - `gh run view --log` + - If the run log says it is still in progress, fetch job logs directly: + - `gh api "/repos///actions/jobs//logs" > ""` +4. Scope non-GitHub Actions checks. + - If `detailsUrl` is not a GitHub Actions run, label it as external and only report the URL. + - Do not attempt Buildkite or other providers; keep the workflow lean. +5. Summarize failures for the user. + - Provide the failing check name, run URL (if any), and a concise log snippet. + - Call out missing logs explicitly. +6. Create a plan. + - Use the `plan` skill to draft a concise plan and request approval. +7. Implement after approval. + - Apply the approved plan, summarize diffs/tests, and ask about opening a PR. +8. Recheck status. + - After changes, suggest re-running the relevant tests and `gh pr checks` to confirm. + +## Bundled Resources + +### scripts/inspect_pr_checks.py + +Fetch failing PR checks, pull GitHub Actions logs, and extract a failure snippet. Exits non-zero when failures remain so it can be used in automation. + +Usage examples: +- `python "/scripts/inspect_pr_checks.py" --repo "." --pr "123"` +- `python "/scripts/inspect_pr_checks.py" --repo "." --pr "https://github.com/org/repo/pull/123" --json` +- `python "/scripts/inspect_pr_checks.py" --repo "." --max-lines 200 --context 40` diff --git a/skills/.curated/gh-fix-ci/scripts/inspect_pr_checks.py b/skills/.curated/gh-fix-ci/scripts/inspect_pr_checks.py new file mode 100755 index 0000000..5ca5298 --- /dev/null +++ b/skills/.curated/gh-fix-ci/scripts/inspect_pr_checks.py @@ -0,0 +1,505 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +from pathlib import Path +from typing import Any, Iterable, Sequence + +FAILURE_CONCLUSIONS = { + "failure", + "cancelled", + "timed_out", + "action_required", +} + +FAILURE_STATES = { + "failure", + "error", + "cancelled", + "timed_out", + "action_required", +} + +FAILURE_BUCKETS = {"fail"} + +FAILURE_MARKERS = ( + "error", + "fail", + "failed", + "traceback", + "exception", + "assert", + "panic", + "fatal", + "timeout", + "segmentation fault", +) + +DEFAULT_MAX_LINES = 160 +DEFAULT_CONTEXT_LINES = 30 +PENDING_LOG_MARKERS = ( + "still in progress", + "log will be available when it is complete", +) + + +class GhResult: + def __init__(self, returncode: int, stdout: str, stderr: str): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def run_gh_command(args: Sequence[str], cwd: Path) -> GhResult: + process = subprocess.run( + ["gh", *args], + cwd=cwd, + text=True, + capture_output=True, + ) + return GhResult(process.returncode, process.stdout, process.stderr) + + +def run_gh_command_raw(args: Sequence[str], cwd: Path) -> tuple[int, bytes, str]: + process = subprocess.run( + ["gh", *args], + cwd=cwd, + capture_output=True, + ) + stderr = process.stderr.decode(errors="replace") + return process.returncode, process.stdout, stderr + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Inspect failing GitHub PR checks, fetch GitHub Actions logs, and extract a " + "failure snippet." + ), + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("--repo", default=".", help="Path inside the target Git repository.") + parser.add_argument( + "--pr", default=None, help="PR number or URL (defaults to current branch PR)." + ) + parser.add_argument("--max-lines", type=int, default=DEFAULT_MAX_LINES) + parser.add_argument("--context", type=int, default=DEFAULT_CONTEXT_LINES) + parser.add_argument("--json", action="store_true", help="Emit JSON instead of text output.") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + repo_root = find_git_root(Path(args.repo)) + if repo_root is None: + print("Error: not inside a Git repository.", file=sys.stderr) + return 1 + + if not ensure_gh_available(repo_root): + return 1 + + pr_value = resolve_pr(args.pr, repo_root) + if pr_value is None: + return 1 + + checks = fetch_checks(pr_value, repo_root) + if checks is None: + return 1 + + failing = [c for c in checks if is_failing(c)] + if not failing: + print(f"PR #{pr_value}: no failing checks detected.") + return 0 + + results = [] + for check in failing: + results.append( + analyze_check( + check, + repo_root=repo_root, + max_lines=max(1, args.max_lines), + context=max(1, args.context), + ) + ) + + if args.json: + print(json.dumps({"pr": pr_value, "results": results}, indent=2)) + else: + render_results(pr_value, results) + + return 1 + + +def find_git_root(start: Path) -> Path | None: + result = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], + cwd=start, + text=True, + capture_output=True, + ) + if result.returncode != 0: + return None + return Path(result.stdout.strip()) + + +def ensure_gh_available(repo_root: Path) -> bool: + result = run_gh_command(["auth", "status"], cwd=repo_root) + if result.returncode == 0: + return True + message = (result.stderr or result.stdout or "").strip() + print(message or "Error: gh not authenticated.", file=sys.stderr) + return False + + +def resolve_pr(pr_value: str | None, repo_root: Path) -> str | None: + if pr_value: + return pr_value + result = run_gh_command(["pr", "view", "--json", "number"], cwd=repo_root) + if result.returncode != 0: + message = (result.stderr or result.stdout or "").strip() + print(message or "Error: unable to resolve PR.", file=sys.stderr) + return None + try: + data = json.loads(result.stdout or "{}") + except json.JSONDecodeError: + print("Error: unable to parse PR JSON.", file=sys.stderr) + return None + number = data.get("number") + if not number: + print("Error: no PR number found.", file=sys.stderr) + return None + return str(number) + + +def fetch_checks(pr_value: str, repo_root: Path) -> list[dict[str, Any]] | None: + primary_fields = ["name", "state", "conclusion", "detailsUrl", "startedAt", "completedAt"] + result = run_gh_command( + ["pr", "checks", pr_value, "--json", ",".join(primary_fields)], + cwd=repo_root, + ) + if result.returncode != 0: + message = "\n".join(filter(None, [result.stderr, result.stdout])).strip() + available_fields = parse_available_fields(message) + if available_fields: + fallback_fields = [ + "name", + "state", + "bucket", + "link", + "startedAt", + "completedAt", + "workflow", + ] + selected_fields = [field for field in fallback_fields if field in available_fields] + if not selected_fields: + print("Error: no usable fields available for gh pr checks.", file=sys.stderr) + return None + result = run_gh_command( + ["pr", "checks", pr_value, "--json", ",".join(selected_fields)], + cwd=repo_root, + ) + if result.returncode != 0: + message = (result.stderr or result.stdout or "").strip() + print(message or "Error: gh pr checks failed.", file=sys.stderr) + return None + else: + print(message or "Error: gh pr checks failed.", file=sys.stderr) + return None + try: + data = json.loads(result.stdout or "[]") + except json.JSONDecodeError: + print("Error: unable to parse checks JSON.", file=sys.stderr) + return None + if not isinstance(data, list): + print("Error: unexpected checks JSON shape.", file=sys.stderr) + return None + return data + + +def is_failing(check: dict[str, Any]) -> bool: + conclusion = normalize_field(check.get("conclusion")) + if conclusion in FAILURE_CONCLUSIONS: + return True + state = normalize_field(check.get("state") or check.get("status")) + if state in FAILURE_STATES: + return True + bucket = normalize_field(check.get("bucket")) + return bucket in FAILURE_BUCKETS + + +def analyze_check( + check: dict[str, Any], + repo_root: Path, + max_lines: int, + context: int, +) -> dict[str, Any]: + url = check.get("detailsUrl") or check.get("link") or "" + run_id = extract_run_id(url) + job_id = extract_job_id(url) + base: dict[str, Any] = { + "name": check.get("name", ""), + "detailsUrl": url, + "runId": run_id, + "jobId": job_id, + } + + if run_id is None: + base["status"] = "external" + base["note"] = "No GitHub Actions run id detected in detailsUrl." + return base + + metadata = fetch_run_metadata(run_id, repo_root) + log_text, log_error, log_status = fetch_check_log( + run_id=run_id, + job_id=job_id, + repo_root=repo_root, + ) + + if log_status == "pending": + base["status"] = "log_pending" + base["note"] = log_error or "Logs are not available yet." + if metadata: + base["run"] = metadata + return base + + if log_error: + base["status"] = "log_unavailable" + base["error"] = log_error + if metadata: + base["run"] = metadata + return base + + snippet = extract_failure_snippet(log_text, max_lines=max_lines, context=context) + base["status"] = "ok" + base["run"] = metadata or {} + base["logSnippet"] = snippet + base["logTail"] = tail_lines(log_text, max_lines) + return base + + +def extract_run_id(url: str) -> str | None: + if not url: + return None + for pattern in (r"/actions/runs/(\d+)", r"/runs/(\d+)"): + match = re.search(pattern, url) + if match: + return match.group(1) + return None + + +def extract_job_id(url: str) -> str | None: + if not url: + return None + match = re.search(r"/actions/runs/\d+/job/(\d+)", url) + if match: + return match.group(1) + match = re.search(r"/job/(\d+)", url) + if match: + return match.group(1) + return None + + +def fetch_run_metadata(run_id: str, repo_root: Path) -> dict[str, Any] | None: + fields = [ + "conclusion", + "status", + "workflowName", + "name", + "event", + "headBranch", + "headSha", + "url", + ] + result = run_gh_command(["run", "view", run_id, "--json", ",".join(fields)], cwd=repo_root) + if result.returncode != 0: + return None + try: + data = json.loads(result.stdout or "{}") + except json.JSONDecodeError: + return None + if not isinstance(data, dict): + return None + return data + + +def fetch_check_log( + run_id: str, + job_id: str | None, + repo_root: Path, +) -> tuple[str, str, str]: + log_text, log_error = fetch_run_log(run_id, repo_root) + if not log_error: + return log_text, "", "ok" + + if is_log_pending_message(log_error) and job_id: + job_log, job_error = fetch_job_log(job_id, repo_root) + if job_log: + return job_log, "", "ok" + if job_error and is_log_pending_message(job_error): + return "", job_error, "pending" + if job_error: + return "", job_error, "error" + return "", log_error, "pending" + + if is_log_pending_message(log_error): + return "", log_error, "pending" + + return "", log_error, "error" + + +def fetch_run_log(run_id: str, repo_root: Path) -> tuple[str, str]: + result = run_gh_command(["run", "view", run_id, "--log"], cwd=repo_root) + if result.returncode != 0: + error = (result.stderr or result.stdout or "").strip() + return "", error or "gh run view failed" + return result.stdout, "" + + +def fetch_job_log(job_id: str, repo_root: Path) -> tuple[str, str]: + repo_slug = fetch_repo_slug(repo_root) + if not repo_slug: + return "", "Error: unable to resolve repository name for job logs." + endpoint = f"/repos/{repo_slug}/actions/jobs/{job_id}/logs" + returncode, stdout_bytes, stderr = run_gh_command_raw(["api", endpoint], cwd=repo_root) + if returncode != 0: + message = (stderr or stdout_bytes.decode(errors="replace")).strip() + return "", message or "gh api job logs failed" + if is_zip_payload(stdout_bytes): + return "", "Job logs returned a zip archive; unable to parse." + return stdout_bytes.decode(errors="replace"), "" + + +def fetch_repo_slug(repo_root: Path) -> str | None: + result = run_gh_command(["repo", "view", "--json", "nameWithOwner"], cwd=repo_root) + if result.returncode != 0: + return None + try: + data = json.loads(result.stdout or "{}") + except json.JSONDecodeError: + return None + name_with_owner = data.get("nameWithOwner") + if not name_with_owner: + return None + return str(name_with_owner) + + +def normalize_field(value: Any) -> str: + if value is None: + return "" + return str(value).strip().lower() + + +def parse_available_fields(message: str) -> list[str]: + if "Available fields:" not in message: + return [] + fields: list[str] = [] + collecting = False + for line in message.splitlines(): + if "Available fields:" in line: + collecting = True + continue + if not collecting: + continue + field = line.strip() + if not field: + continue + fields.append(field) + return fields + + +def is_log_pending_message(message: str) -> bool: + lowered = message.lower() + return any(marker in lowered for marker in PENDING_LOG_MARKERS) + + +def is_zip_payload(payload: bytes) -> bool: + return payload.startswith(b"PK") + + +def extract_failure_snippet(log_text: str, max_lines: int, context: int) -> str: + lines = log_text.splitlines() + if not lines: + return "" + + marker_index = find_failure_index(lines) + if marker_index is None: + return "\n".join(lines[-max_lines:]) + + start = max(0, marker_index - context) + end = min(len(lines), marker_index + context) + window = lines[start:end] + if len(window) > max_lines: + window = window[-max_lines:] + return "\n".join(window) + + +def find_failure_index(lines: Sequence[str]) -> int | None: + for idx in range(len(lines) - 1, -1, -1): + lowered = lines[idx].lower() + if any(marker in lowered for marker in FAILURE_MARKERS): + return idx + return None + + +def tail_lines(text: str, max_lines: int) -> str: + if max_lines <= 0: + return "" + lines = text.splitlines() + return "\n".join(lines[-max_lines:]) + + +def render_results(pr_number: str, results: Iterable[dict[str, Any]]) -> None: + results_list = list(results) + print(f"PR #{pr_number}: {len(results_list)} failing checks analyzed.") + for result in results_list: + print("-" * 60) + print(f"Check: {result.get('name', '')}") + if result.get("detailsUrl"): + print(f"Details: {result['detailsUrl']}") + run_id = result.get("runId") + if run_id: + print(f"Run ID: {run_id}") + job_id = result.get("jobId") + if job_id: + print(f"Job ID: {job_id}") + status = result.get("status", "unknown") + print(f"Status: {status}") + + run_meta = result.get("run", {}) + if run_meta: + branch = run_meta.get("headBranch", "") + sha = (run_meta.get("headSha") or "")[:12] + workflow = run_meta.get("workflowName") or run_meta.get("name") or "" + conclusion = run_meta.get("conclusion") or run_meta.get("status") or "" + print(f"Workflow: {workflow} ({conclusion})") + if branch or sha: + print(f"Branch/SHA: {branch} {sha}") + if run_meta.get("url"): + print(f"Run URL: {run_meta['url']}") + + if result.get("note"): + print(f"Note: {result['note']}") + + if result.get("error"): + print(f"Error fetching logs: {result['error']}") + continue + + snippet = result.get("logSnippet") or "" + if snippet: + print("Failure snippet:") + print(indent_block(snippet, prefix=" ")) + else: + print("No snippet available.") + print("-" * 60) + + +def indent_block(text: str, prefix: str = " ") -> str: + return "\n".join(f"{prefix}{line}" for line in text.splitlines()) + + +if __name__ == "__main__": + raise SystemExit(main())