"""GitHub work, PR, security, and issue-mutation workflow tools.""" from __future__ import annotations from typing import Any, Literal, cast from core.tool_framework.tool_decorator import tool from integrations.github.client import GitHubApiError, GitHubRestClient, resolve_github_token from integrations.github.helpers import github_creds, github_source_available from integrations.github.tools.workflow import ( GitHubIssueMutationProposal, PullRequestStatus, SecurityAlert, WorkItem, build_issue_mutation_proposal, ) _HELP_WANTED_LABELS = {"help wanted", "good first issue", "up for grabs", "agent-ready"} _BLOCKING_MERGEABLE_STATES = {"blocked", "dirty", "behind", "unstable"} _FAILED_CHECK_CONCLUSIONS = { "failure", "cancelled", "timed_out", "action_required", "startup_failure", } _TERMINAL_CHECK_CONCLUSIONS = _FAILED_CHECK_CONCLUSIONS | {"success", "skipped", "neutral"} def _github_available(sources: dict[str, dict]) -> bool: gh = sources.get("github", {}) return bool( (github_source_available(sources) or resolve_github_token(None)) and gh.get("owner") and gh.get("repo") ) def _github_extract_params(sources: dict[str, dict]) -> dict[str, Any]: gh = sources.get("github", {}) if not gh: return {} return {"owner": gh.get("owner"), "repo": gh.get("repo"), **github_creds(gh)} def _labels(item: dict[str, Any]) -> list[str]: return [ str(label.get("name", "")).strip() for label in item.get("labels", []) if isinstance(label, dict) ] def _logins(items: Any) -> list[str]: if not isinstance(items, list): return [] return [ str(item.get("login", "")).strip() for item in items if isinstance(item, dict) and item.get("login") ] def _normalize_issue(item: dict[str, Any]) -> WorkItem: labels = _labels(item) assignees = _logins(item.get("assignees")) label_set = {label.lower() for label in labels} if assignees: work_status: Literal["taken", "up_for_grabs", "unassigned"] = "taken" elif label_set & _HELP_WANTED_LABELS: work_status = "up_for_grabs" else: work_status = "unassigned" return WorkItem( number=item.get("number") if isinstance(item.get("number"), int) else None, title=str(item.get("title", "")), state=str(item.get("state", "")), url=str(item.get("html_url", "")), author=str((item.get("user") or {}).get("login", "")), labels=labels, assignees=assignees, updated_at=str(item.get("updated_at", "")), work_status=work_status, ) def _count_work_items(items: list[dict[str, Any]]) -> dict[str, int]: return { "total": len(items), "taken": sum(1 for item in items if item.get("work_status") == "taken"), "up_for_grabs": sum(1 for item in items if item.get("work_status") == "up_for_grabs"), "unassigned": sum(1 for item in items if item.get("work_status") == "unassigned"), } @tool( name="list_github_work_items", source="github", description="List GitHub issues as engineering work items and classify them as taken, up for grabs, or unassigned.", use_cases=[ "Answering which GitHub issues are taken versus available", "Building engineering status reports from open issue state", "Finding unassigned or agent-ready work without mutating GitHub", ], anti_examples=["Creating, editing, or closing GitHub issues"], requires=["owner", "repo"], surfaces=("investigation", "chat"), side_effect_level="read_only", input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "state": {"type": "string", "enum": ["open", "closed", "all"]}, "labels": {"type": "string"}, "include_prs": {"type": "boolean"}, "per_page": {"type": "integer"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo"], }, is_available=_github_available, extract_params=_github_extract_params, ) def list_github_work_items( owner: str, repo: str, state: str = "open", labels: str = "", include_prs: bool = False, per_page: int = 50, github_token: str | None = None, **_kwargs: Any, ) -> dict[str, Any]: params: dict[str, Any] = {"state": state, "per_page": max(1, min(per_page, 100))} if labels.strip(): params["labels"] = labels.strip() try: raw_items = GitHubRestClient(github_token).paginate( f"/repos/{owner}/{repo}/issues", params=params ) except GitHubApiError as exc: return { "source": "github", "available": False, "error": str(exc), "items": [], "counts": _count_work_items([]), "side_effects": [], } items = [ _normalize_issue(item).to_dict() for item in raw_items if include_prs or "pull_request" not in item ] return { "source": "github", "available": True, "owner": owner, "repo": repo, "items": items, "counts": _count_work_items(items), "side_effects": [], } def _check_summary(check_runs: list[dict[str, Any]]) -> tuple[str, list[str]]: failed = [ str(run.get("name", "check")) for run in check_runs if str(run.get("conclusion") or "").lower() in _FAILED_CHECK_CONCLUSIONS ] pending = [ str(run.get("name", "check")) for run in check_runs if str(run.get("status") or "").lower() != "completed" or str(run.get("conclusion") or "").lower() not in _TERMINAL_CHECK_CONCLUSIONS ] if failed: return "failed", failed if pending: return "pending", pending return "passing", [] def _normalize_pull_request( pr: dict[str, Any], check_runs: list[dict[str, Any]] ) -> PullRequestStatus: check_status, check_names = _check_summary(check_runs) mergeable = pr.get("mergeable") if isinstance(pr.get("mergeable"), bool) else None mergeable_state = str(pr.get("mergeable_state") or "unknown").lower() reasons: list[str] = [] status: Literal["mergeable", "blocked", "unknown"] if pr.get("draft"): reasons.append("draft") if mergeable_state in _BLOCKING_MERGEABLE_STATES: reasons.append(f"mergeable_state={mergeable_state}") if check_status == "failed": reasons.append(f"failed checks: {', '.join(check_names)}") elif check_status == "pending": reasons.append(f"pending checks: {', '.join(check_names)}") if mergeable is None or mergeable_state == "unknown": reasons.append("mergeability unknown") status = "unknown" elif reasons or mergeable is False: status = "blocked" if mergeable is False and not any( reason.startswith("mergeable_state=") for reason in reasons ): reasons.append("mergeable=false") else: status = "mergeable" return PullRequestStatus( number=pr.get("number") if isinstance(pr.get("number"), int) else None, title=str(pr.get("title", "")), url=str(pr.get("html_url", "")), author=str((pr.get("user") or {}).get("login", "")), head_ref=str((pr.get("head") or {}).get("ref", "")), head_sha=str((pr.get("head") or {}).get("sha", "")), draft=bool(pr.get("draft")), mergeable=mergeable, mergeable_state=mergeable_state, check_status=check_status, status=status, mergeability=status, blocking_reasons=reasons, updated_at=str(pr.get("updated_at", "")), ) def _count_prs(prs: list[dict[str, Any]]) -> dict[str, int]: return { "total": len(prs), "mergeable": sum(1 for pr in prs if pr.get("status") == "mergeable"), "blocked": sum(1 for pr in prs if pr.get("status") == "blocked"), "unknown": sum(1 for pr in prs if pr.get("status") == "unknown"), "draft": sum(1 for pr in prs if pr.get("draft")), } @tool( name="summarize_github_pr_status", source="github", description="Summarize open GitHub pull requests, authoritative mergeability, checks, and blocking reasons.", use_cases=[ "Answering which PRs are mergeable, blocked, or unknown", "Finding failing or pending CI checks for active work", "Preparing engineering status updates without changing GitHub state", ], requires=["owner", "repo"], surfaces=("investigation", "chat"), side_effect_level="read_only", input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "state": {"type": "string", "enum": ["open", "closed", "all"]}, "per_page": {"type": "integer"}, "include_checks": {"type": "boolean"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo"], }, is_available=_github_available, extract_params=_github_extract_params, ) def summarize_github_pr_status( owner: str, repo: str, state: str = "open", per_page: int = 30, include_checks: bool = True, github_token: str | None = None, **_kwargs: Any, ) -> dict[str, Any]: client = GitHubRestClient(github_token) try: raw_prs = client.paginate( f"/repos/{owner}/{repo}/pulls", params={"state": state, "per_page": max(1, min(per_page, 100))}, ) prs: list[dict[str, Any]] = [] for list_pr in raw_prs: number = list_pr.get("number") if not isinstance(number, int): continue detail_pr = client.request("GET", f"/repos/{owner}/{repo}/pulls/{number}") if not isinstance(detail_pr, dict): detail_pr = list_pr sha = str((detail_pr.get("head") or {}).get("sha", "")) check_runs: list[dict[str, Any]] = [] if include_checks and sha: check_payload = client.request( "GET", f"/repos/{owner}/{repo}/commits/{sha}/check-runs", params={"per_page": 100}, ) if isinstance(check_payload, dict) and isinstance( check_payload.get("check_runs"), list ): check_runs = [ run for run in check_payload["check_runs"] if isinstance(run, dict) ] prs.append(_normalize_pull_request(detail_pr, check_runs).to_dict()) except GitHubApiError as exc: return { "source": "github", "available": False, "error": str(exc), "pull_requests": [], "counts": _count_prs([]), "side_effects": [], } return { "source": "github", "available": True, "owner": owner, "repo": repo, "pull_requests": prs, "counts": _count_prs(prs), "side_effects": [], } def _normalize_security_alert(alert_type: str, item: dict[str, Any]) -> SecurityAlert: summary = "" if alert_type == "dependabot": summary = str((item.get("security_advisory") or {}).get("summary", "")) elif alert_type == "secret_scanning": summary = str(item.get("secret_type", "")) elif alert_type == "code_scanning": summary = str((item.get("rule") or {}).get("description", "")) return SecurityAlert( type=alert_type, number=item.get("number"), state=str(item.get("state", "")), summary=summary, url=str(item.get("html_url", "")), ) _ALERT_ENDPOINTS = { "dependabot": "dependabot/alerts", "secret_scanning": "secret-scanning/alerts", "code_scanning": "code-scanning/alerts", } _ISSUE_MUTATION_OPERATIONS = {"create", "update", "close"} @tool( name="list_github_security_alerts", source="github", description="List GitHub Dependabot, secret-scanning, and code-scanning alerts when token scope allows it.", use_cases=[ "Surfacing repository security alerts during work triage", "Checking whether secret scanning or code scanning has open alerts", "Building a read-only engineering status report with security context", ], requires=["owner", "repo"], surfaces=("investigation", "chat"), side_effect_level="read_only", input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "alert_type": { "type": "string", "enum": ["all", "dependabot", "secret_scanning", "code_scanning"], }, "state": {"type": "string"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo"], }, is_available=_github_available, extract_params=_github_extract_params, ) def list_github_security_alerts( owner: str, repo: str, alert_type: str = "all", state: str = "open", github_token: str | None = None, **_kwargs: Any, ) -> dict[str, Any]: client = GitHubRestClient(github_token) selected = list(_ALERT_ENDPOINTS) if alert_type == "all" else [alert_type] alerts: list[dict[str, Any]] = [] errors: dict[str, str] = {} for kind in selected: endpoint = _ALERT_ENDPOINTS.get(kind) if endpoint is None: errors[kind] = f"Unsupported alert_type: {kind}" continue try: payload = client.paginate( f"/repos/{owner}/{repo}/{endpoint}", params={"state": state, "per_page": 100} ) except GitHubApiError as exc: errors[kind] = str(exc) continue alerts.extend(_normalize_security_alert(kind, item).to_dict() for item in payload) counts = { kind: sum(1 for alert in alerts if alert.get("type") == kind) for kind in _ALERT_ENDPOINTS } counts["total"] = len(alerts) return { "source": "github", "available": not errors or bool(alerts), "owner": owner, "repo": repo, "alerts": alerts, "counts": counts, "errors": errors, "side_effects": [], } @tool( name="propose_github_issue_mutation_from_slack", source="github", description="Build a read-only proposal for creating, updating, or closing a GitHub issue from an explicit Slack request.", use_cases=[ "Preparing a Slack-sourced GitHub issue change for human approval", "Rendering deterministic issue mutation payloads without mutating GitHub", ], anti_examples=["Directly mutating GitHub", "Inferring tasks from ambiguous Slack discussion"], surfaces=("chat",), side_effect_level="read_only", input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "operation": {"type": "string", "enum": ["create", "update", "close"]}, "issue_number": {"type": "integer"}, "slack_text": {"type": "string"}, "slack_url": {"type": "string"}, "title": {"type": "string"}, "labels": {"type": "array", "items": {"type": "string"}}, "assignees": {"type": "array", "items": {"type": "string"}}, }, "required": ["owner", "repo", "operation", "slack_text"], }, is_available=_github_available, extract_params=_github_extract_params, ) def propose_github_issue_mutation_from_slack( owner: str, repo: str, operation: Literal["create", "update", "close"], slack_text: str, slack_url: str = "", issue_number: int | None = None, title: str = "", labels: list[str] | None = None, assignees: list[str] | None = None, **_kwargs: Any, ) -> dict[str, Any]: if operation in {"update", "close"} and issue_number is None: return { "source": "github", "available": False, "error": f"issue_number is required for {operation}", "side_effects": [], } proposal = build_issue_mutation_proposal( owner=owner, repo=repo, operation=operation, issue_number=issue_number, slack_text=slack_text, slack_url=slack_url, title=title, labels=labels, assignees=assignees, ) return { "source": "github", "available": True, "proposal": proposal.to_dict(), "side_effects": [], } def _mutation_rejected(error: str) -> dict[str, Any]: return { "source": "github", "available": False, "executed": False, "error": error, "side_effect": "github_issue_mutation_rejected", } def _proposal_from_payload( payload: Any, ) -> tuple[GitHubIssueMutationProposal | None, str | None]: if not isinstance(payload, dict): return None, "proposal must be an object" required = ("proposal_id", "operation", "owner", "repo", "target", "payload") missing = [key for key in required if key not in payload] if missing: return None, f"proposal missing required field(s): {', '.join(missing)}" operation = payload.get("operation") if operation not in _ISSUE_MUTATION_OPERATIONS: return None, f"unsupported proposal operation: {operation}" target = payload.get("target") if not isinstance(target, dict): return None, "proposal target must be an object" mutation_payload = payload.get("payload") if not isinstance(mutation_payload, dict): return None, "proposal payload must be an object" proposal_id = str(payload.get("proposal_id") or "").strip() owner = str(payload.get("owner") or "").strip() repo = str(payload.get("repo") or "").strip() idempotency_marker = str(payload.get("idempotency_marker") or "").strip() if not proposal_id: return None, "proposal_id is required" if not owner or not repo: return None, "proposal owner and repo are required" if not idempotency_marker or proposal_id not in idempotency_marker: return None, "proposal idempotency_marker is missing or does not match proposal_id" return ( GitHubIssueMutationProposal( proposal_id=proposal_id, operation=cast(Literal["create", "update", "close"], operation), owner=owner, repo=repo, target=target, payload=mutation_payload, slack_url=str(payload.get("slack_url", "")), idempotency_marker=idempotency_marker, ), None, ) def _proposal_marker_text(proposal: GitHubIssueMutationProposal) -> str: if proposal.operation == "create": return str(proposal.payload.get("body", "")) return str(proposal.payload.get("comment_body", "")) def _validate_proposal_marker(proposal: GitHubIssueMutationProposal) -> str | None: if proposal.idempotency_marker not in _proposal_marker_text(proposal): return "proposal payload does not include its idempotency marker" return None def _quoted_search_term(term: str) -> str: cleaned = term.replace('"', "") return f'"{cleaned}"' def _search_issues_for_marker( client: GitHubRestClient, *, owner: str, repo: str, marker: str, search_area: Literal["body", "comments"], ) -> list[dict[str, Any]]: result = client.request( "GET", "/search/issues", params={ "q": (f"repo:{owner}/{repo} is:issue in:{search_area} {_quoted_search_term(marker)}") }, ) if not isinstance(result, dict): return [] items = result.get("items") if not isinstance(items, list): return [] return [item for item in items if isinstance(item, dict)] def _marker_exists_on_issue( client: GitHubRestClient, *, owner: str, repo: str, issue_number: int, marker: str, ) -> bool: return any( item.get("number") == issue_number for item in _search_issues_for_marker( client, owner=owner, repo=repo, marker=marker, search_area="comments", ) ) @tool( name="execute_github_issue_mutation", source="github", description="Execute a GitHub issue mutation proposal. Not exposed to investigation.", use_cases=["Executing a previously rendered GitHub issue mutation proposal"], anti_examples=[ "Creating proposals", "Running during investigations", ], surfaces=("chat",), side_effect_level="mutating", input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "proposal": {"type": "object"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo", "proposal"], }, is_available=_github_available, extract_params=_github_extract_params, ) def execute_github_issue_mutation( owner: str, repo: str, proposal: dict[str, Any], github_token: str | None = None, **_kwargs: Any, ) -> dict[str, Any]: client = GitHubRestClient(github_token) parsed, parse_error = _proposal_from_payload(proposal) if parsed is None: return _mutation_rejected(parse_error or "invalid proposal") if parsed.owner != owner or parsed.repo != repo: return _mutation_rejected("proposal owner/repo does not match request") marker_error = _validate_proposal_marker(parsed) if marker_error is not None: return _mutation_rejected(marker_error) try: if parsed.operation == "create": existing_items = _search_issues_for_marker( client, owner=owner, repo=repo, marker=parsed.idempotency_marker, search_area="body", ) if existing_items: return { "source": "github", "available": True, "executed": False, "side_effect": "existing_github_issue", "issue": existing_items[0], } issue = client.request("POST", f"/repos/{owner}/{repo}/issues", body=parsed.payload) return { "source": "github", "available": True, "executed": True, "side_effect": "created_github_issue", "issue": issue, } issue_number = parsed.target.get("issue_number") if not isinstance(issue_number, int): return _mutation_rejected("proposal target.issue_number is required") client.request("GET", f"/repos/{owner}/{repo}/issues/{issue_number}") comment_body = str(parsed.payload.get("comment_body", "")) comment_already_recorded = _marker_exists_on_issue( client, owner=owner, repo=repo, issue_number=issue_number, marker=parsed.idempotency_marker, ) if comment_body and not comment_already_recorded: client.request( "POST", f"/repos/{owner}/{repo}/issues/{issue_number}/comments", body={"body": comment_body}, ) if parsed.operation == "update": patch_body = { key: parsed.payload[key] for key in ("title", "labels", "assignees") if key in parsed.payload } issue = ( client.request( "PATCH", f"/repos/{owner}/{repo}/issues/{issue_number}", body=patch_body ) if patch_body else {"number": issue_number} ) return { "source": "github", "available": True, "executed": True, "side_effect": "updated_github_issue", "issue": issue, "comment_already_recorded": comment_already_recorded, } issue = client.request( "PATCH", f"/repos/{owner}/{repo}/issues/{issue_number}", body={"state": "closed", "state_reason": "completed"}, ) return { "source": "github", "available": True, "executed": True, "side_effect": "closed_github_issue", "issue": issue, "comment_already_recorded": comment_already_recorded, } except GitHubApiError as exc: return { "source": "github", "available": False, "executed": False, "error": str(exc), "side_effect": f"{parsed.operation}_github_issue_failed", }