"""GitHub Actions workflow investigation tools - MCP-direct implementation.""" from __future__ import annotations import json from typing import Any, cast from core.tool_framework.tool_decorator import tool from core.tool_framework.utils.code_host_unavailable import code_host_unavailable_payload from integrations.github.helpers import ( github_creds, github_source_available, normalize_github_tool_result, resolve_github_mcp_config, ) from integrations.github.mcp import call_github_mcp_tool def _extract_json_text(result: dict[str, Any]) -> dict[str, Any] | str | None: text = str(result.get("text") or "").strip() if not text: return None try: parsed = json.loads(text) return cast(dict[str, Any], parsed) except json.JSONDecodeError: return text def _extract_list(result: dict[str, Any], key: str) -> list[dict[str, Any]]: """Extract list of items from MCP tool result.""" json_result = _extract_json_text(result) items: list[Any] = [] if isinstance(json_result, dict): value = json_result.get(key) if isinstance(value, list): items = value return [item for item in items if isinstance(item, dict)] def _extract_workflow_jobs(result: dict[str, Any]) -> list[dict[str, Any]]: """Extract workflow jobs from MCP tool result.""" json_result = _extract_json_text(result) if isinstance(json_result, dict) and "jobs" in json_result: jobs_raw = json_result["jobs"] if isinstance(jobs_raw, dict) and "jobs" in jobs_raw: target_list = jobs_raw["jobs"] else: target_list = jobs_raw if isinstance(target_list, list): return [_normalize_job(job) for job in target_list if isinstance(job, dict)] return [] def _extract_log_text(result: dict[str, Any]) -> str: """Extract log text from MCP tool result.""" json_result = _extract_json_text(result) if isinstance(json_result, dict) and "logs_content" in json_result: return str(json_result["logs_content"] or "").strip() return str(result.get("text") or "").strip() def _normalize_step(step: dict[str, Any]) -> dict[str, Any]: """Normalize step data.""" return { "name": step.get("name", ""), "status": step.get("status", ""), "conclusion": step.get("conclusion", ""), "number": step.get("number"), "started_at": step.get("started_at", ""), "completed_at": step.get("completed_at", ""), } def _normalize_job(job: dict[str, Any]) -> dict[str, Any]: """Normalize job data including steps.""" steps_raw = job.get("steps") steps: list[dict[str, Any]] = [] if isinstance(steps_raw, list): for item in steps_raw: if isinstance(item, dict): steps.append(_normalize_step(item)) return { "id": job.get("id"), "run_id": job.get("run_id"), "name": job.get("name", ""), "status": job.get("status", ""), "conclusion": job.get("conclusion", ""), "started_at": job.get("started_at", ""), "completed_at": job.get("completed_at", ""), "runner_name": job.get("runner_name", ""), "runner_group_name": job.get("runner_group_name", ""), "labels": job.get("labels", []), "html_url": job.get("html_url", ""), "steps": steps, } def _normalize_run(run: dict[str, Any]) -> dict[str, Any]: """Normalize workflow run data.""" actor_raw = run.get("actor") actor = actor_raw if isinstance(actor_raw, dict) else {} triggering_actor_raw = run.get("triggering_actor") triggering_actor = triggering_actor_raw if isinstance(triggering_actor_raw, dict) else {} pull_requests_raw = run.get("pull_requests") pull_requests: list[dict[str, Any]] = [] if isinstance(pull_requests_raw, list): for item in pull_requests_raw: if isinstance(item, dict): head_raw = item.get("head") head = head_raw if isinstance(head_raw, dict) else {} pull_requests.append( { "number": item.get("number"), "url": item.get("html_url", ""), "head_branch": head.get("ref", ""), } ) return { "id": run.get("id"), "name": run.get("name", ""), "display_title": run.get("display_title", ""), "head_branch": run.get("head_branch", ""), "head_sha": run.get("head_sha", ""), "event": run.get("event", ""), "status": run.get("status", ""), "conclusion": run.get("conclusion", ""), "run_number": run.get("run_number"), "run_attempt": run.get("run_attempt"), "html_url": run.get("html_url", ""), "created_at": run.get("created_at", ""), "updated_at": run.get("updated_at", ""), "actor": actor.get("login", "") if isinstance(actor, dict) else "", "triggering_actor": triggering_actor.get("login", "") if isinstance(triggering_actor, dict) else "", "pull_requests": pull_requests, } UNGROUPED_SECTION_NAME = "ungrouped" def _append_log_section(sections: list[dict[str, str]], name: str, lines: list[str]) -> None: """Append a non-empty log section preserving original ordering.""" text = "\n".join(lines).strip() if text: sections.append({"name": name, "text": text}) def _extract_log_sections(log_text: str) -> list[dict[str, str]]: """Extract sections from GitHub Actions log output (marked by ##[group]/##[endgroup]).""" sections: list[dict[str, str]] = [] current_name: str | None = None current_lines: list[str] = [] saw_group = False for line in log_text.splitlines(): if line.startswith("##[group]"): saw_group = True _append_log_section(sections, current_name or UNGROUPED_SECTION_NAME, current_lines) current_name = line[len("##[group]") :].strip() current_lines = [] continue if line.startswith("##[endgroup]"): _append_log_section(sections, current_name or UNGROUPED_SECTION_NAME, current_lines) current_name = None current_lines = [] continue current_lines.append(line) if saw_group: _append_log_section(sections, current_name or UNGROUPED_SECTION_NAME, current_lines) if not saw_group: return [{"name": "full-log", "text": log_text.strip()}] return [section for section in sections if section.get("text")] def extract_step_log( log_text: str, *, step_name: str = "", step_number: int | None = None, ) -> dict[str, Any]: """Extract the log text for a specific step in a GitHub Actions job log, using grouping markers if available.""" sections = _extract_log_sections(log_text) selected: dict[str, str] | None = None match_strategy = "full-log" selected_idx = -1 if step_name: needle = step_name.strip().lower() for i, section in enumerate(sections): if needle and needle in section.get("name", "").lower(): selected = section match_strategy = "step_name" selected_idx = i break group_count = sum(1 for s in sections if s.get("name") != UNGROUPED_SECTION_NAME) if selected is None and step_number is not None and 1 <= step_number <= group_count: group_counter = 0 for i, section in enumerate(sections): if section.get("name") != UNGROUPED_SECTION_NAME: group_counter += 1 if group_counter == step_number: selected = section match_strategy = "step_number" selected_idx = i break if selected is None: selected = {"name": "full-log", "text": log_text.strip()} text = selected.get("text", "") # Merge trailing ungrouped annotations into the matched step log if selected_idx != -1 and selected_idx + 1 < len(sections): next_section = sections[selected_idx + 1] if next_section.get("name") == UNGROUPED_SECTION_NAME: text += "\n" + next_section.get("text", "") return { "step_name": selected.get("name", ""), "match_strategy": match_strategy, "log_text": text, } def _github_actions_is_available(sources: dict[str, dict]) -> bool: """Check if GitHub Actions tool should be available.""" github = sources.get("github", {}) return bool(github_source_available(sources) and github.get("owner") and github.get("repo")) def _github_actions_repo_params(sources: dict[str, dict]) -> dict[str, Any]: """Extract repo parameters for GitHub Actions tools.""" github = sources["github"] params: dict[str, Any] = { "owner": github["owner"], "repo": github["repo"], **github_creds(github), } return params def _github_actions_run_params(sources: dict[str, dict]) -> dict[str, Any]: """Extract run/job parameters for GitHub Actions tools.""" github = sources["github"] params = _github_actions_repo_params(sources) if github.get("run_id") is not None: params["run_id"] = github["run_id"] if github.get("job_id") is not None: params["job_id"] = github["job_id"] return params @tool( name="list_github_actions_workflow_runs", source="github", description="List recent GitHub Actions workflow runs for a repository.", use_cases=[ "Checking which deploy or test workflow failed right before an incident", "Reviewing recent workflow status, trigger, and branch context", "Finding a run that matches an outage window or rollback event", ], requires=["owner", "repo"], surfaces=("investigation", "chat"), input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "branch": {"type": "string", "default": ""}, "status": {"type": "string", "default": ""}, "event": {"type": "string", "default": ""}, "per_page": {"type": "integer", "default": 30}, "github_url": {"type": "string"}, "github_mode": {"type": "string"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo"], }, is_available=_github_actions_is_available, extract_params=_github_actions_repo_params, ) def list_github_actions_workflow_runs( owner: str, repo: str, branch: str = "", status: str = "", event: str = "", per_page: int = 30, github_url: str | None = None, github_mode: str | None = None, github_token: str | None = None, github_command: str | None = None, github_args: list[str] | None = None, **_kwargs: Any, ) -> dict[str, Any]: """List recent GitHub Actions workflow runs for a repository.""" config = resolve_github_mcp_config( github_url, github_mode, github_token, github_command, github_args ) if config is None: return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="workflow_runs", empty_value=[], ) workflow_runs_filter: dict[str, Any] = {} if branch: workflow_runs_filter["branch"] = branch if status: workflow_runs_filter["status"] = status if event: workflow_runs_filter["event"] = event arguments: dict[str, Any] = { "method": "list_workflow_runs", "owner": owner, "repo": repo, "per_page": per_page, } if workflow_runs_filter: arguments["workflow_runs_filter"] = workflow_runs_filter result = call_github_mcp_tool(config, "actions_list", arguments) payload = normalize_github_tool_result(result) if not isinstance(payload, dict): return {"error": "Unexpected payload format returned from GitHub MCP tool"} if payload.get("available"): workflow_runs_raw = _extract_list(result, "workflow_runs") workflow_runs = [_normalize_run(item) for item in workflow_runs_raw] payload["workflow_runs"] = workflow_runs payload["total"] = len(workflow_runs) else: payload["workflow_runs"] = [] payload["total"] = 0 payload["branch"] = branch payload["status"] = status payload["event"] = event return payload @tool( name="list_github_actions_active_runs", source="github", description="List GitHub Actions workflow runs that are currently queued or in progress.", use_cases=[ "Seeing what deployment jobs are still running during an incident", "Spotting queued deploys that may be waiting on a shared runner or lock", ], requires=["owner", "repo"], surfaces=("investigation", "chat"), input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "per_page": {"type": "integer", "default": 30}, "github_url": {"type": "string"}, "github_mode": {"type": "string"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo"], }, is_available=_github_actions_is_available, extract_params=_github_actions_repo_params, ) def list_github_actions_active_runs( owner: str, repo: str, per_page: int = 30, github_url: str | None = None, github_mode: str | None = None, github_token: str | None = None, github_command: str | None = None, github_args: list[str] | None = None, **_kwargs: Any, ) -> dict[str, Any]: """List GitHub Actions workflow runs that are currently queued or in progress.""" config = resolve_github_mcp_config( github_url, github_mode, github_token, github_command, github_args ) if config is None: return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="workflow_runs", empty_value=[], ) # Fetch queued runs queued_result = call_github_mcp_tool( config, "actions_list", { "method": "list_workflow_runs", "owner": owner, "repo": repo, "per_page": per_page, "workflow_runs_filter": {"status": "queued"}, }, ) # Fetch in_progress runs in_progress_result = call_github_mcp_tool( config, "actions_list", { "method": "list_workflow_runs", "owner": owner, "repo": repo, "per_page": per_page, "workflow_runs_filter": {"status": "in_progress"}, }, ) # Check for errors if queued_result.get("is_error") or in_progress_result.get("is_error"): error_texts: list[str] = [] for result in (queued_result, in_progress_result): if result.get("is_error") and result.get("text"): error_texts.append(str(result.get("text"))) error_msg = " | ".join(error_texts) if error_texts else "Failed to list active runs" return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="workflow_runs", empty_value=[], ) | {"error": error_msg} # Combine and deduplicate combined: list[dict[str, Any]] = [] seen_ids: set[Any] = set() for result in (queued_result, in_progress_result): runs_raw = _extract_list(result, "workflow_runs") for run in runs_raw: run_id = run.get("id") if run_id is not None and run_id not in seen_ids: seen_ids.add(run_id) combined.append(_normalize_run(run)) combined.sort(key=lambda item: str(item.get("created_at", "")), reverse=True) return { "source": "github", "available": True, "workflow_runs": combined, "total": len(combined), } @tool( name="list_github_actions_run_jobs", source="github", description="List jobs and step outcomes for a GitHub Actions workflow run.", use_cases=[ "Finding which job failed in a deployment workflow", "Checking step-by-step status for test, build, and deploy jobs", ], requires=["owner", "repo", "run_id"], surfaces=("investigation", "chat"), input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "run_id": {"type": "integer"}, "github_url": {"type": "string"}, "github_mode": {"type": "string"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo", "run_id"], }, is_available=_github_actions_is_available, extract_params=_github_actions_run_params, ) def list_github_actions_run_jobs( owner: str, repo: str, run_id: int, github_url: str | None = None, github_mode: str | None = None, github_token: str | None = None, github_command: str | None = None, github_args: list[str] | None = None, **_kwargs: Any, ) -> dict[str, Any]: """List jobs and step outcomes for a GitHub Actions workflow run.""" config = resolve_github_mcp_config( github_url, github_mode, github_token, github_command, github_args ) if config is None: return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="jobs", empty_value=[], ) result = call_github_mcp_tool( config, "actions_list", { "method": "list_workflow_jobs", "owner": owner, "repo": repo, "resource_id": str(run_id), }, ) payload = normalize_github_tool_result(result) if not isinstance(payload, dict): return {"error": "Unexpected payload format returned from GitHub MCP tool"} if payload.get("available"): jobs = _extract_workflow_jobs(result) payload["jobs"] = jobs payload["total"] = len(jobs) else: payload["jobs"] = [] payload["total"] = 0 payload["workflow_run_id"] = run_id return payload @tool( name="get_github_actions_step_log", source="github", description="Fetch the log output for a failed GitHub Actions job step.", use_cases=[ "Reading the error output for the step that broke a deployment", "Checking the exact log snippet for a flaky test or secret-related failure", ], requires=["owner", "repo", "run_id", "job_id"], surfaces=("investigation", "chat"), input_schema={ "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "run_id": {"type": "integer"}, "job_id": {"type": "integer"}, "step_name": {"type": "string", "default": ""}, "step_number": {"type": "integer"}, "tail_lines": {"type": "integer", "default": 500}, "github_url": {"type": "string"}, "github_mode": {"type": "string"}, "github_token": {"type": "string"}, }, "required": ["owner", "repo", "run_id", "job_id"], }, is_available=_github_actions_is_available, extract_params=_github_actions_run_params, ) def get_github_actions_step_log( owner: str, repo: str, run_id: int, job_id: int, step_name: str = "", step_number: int | None = None, tail_lines: int = 500, github_url: str | None = None, github_mode: str | None = None, github_token: str | None = None, github_command: str | None = None, github_args: list[str] | None = None, **_kwargs: Any, ) -> dict[str, Any]: """Fetch the log output for a failed GitHub Actions job step.""" config = resolve_github_mcp_config( github_url, github_mode, github_token, github_command, github_args ) if config is None: return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="log_text", empty_value="", ) # Fetch job metadata job_result = call_github_mcp_tool( config, "actions_get", { "method": "get_workflow_job", "owner": owner, "repo": repo, "resource_id": str(job_id), }, ) if job_result.get("is_error"): return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="log_text", empty_value="", ) | {"error": job_result.get("text")} job = _extract_json_text(job_result) if not isinstance(job, dict): return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="log_text", empty_value="", ) | {"error": "Unexpected job metadata format"} # Fetch job logs log_result = call_github_mcp_tool( config, "get_job_logs", { "owner": owner, "repo": repo, "job_id": job_id, "return_content": True, "tail_lines": tail_lines, }, ) if log_result.get("is_error"): return code_host_unavailable_payload( source="github", integration_name="GitHub Actions", empty_key="log_text", empty_value="", ) | {"error": log_result.get("text")} log_text = _extract_log_text(log_result) # Detect first failed step if not specified failed_step = "" steps_raw = job.get("steps") if isinstance(job, dict) else [] normalized_steps = [] if isinstance(steps_raw, list): normalized_steps = [_normalize_step(step) for step in steps_raw if isinstance(step, dict)] if not step_name: for step in steps_raw: if not isinstance(step, dict): continue if step.get("conclusion") == "failure" or step.get("status") == "failure": failed_step = str(step.get("name") or "") break # Extract and filter step log extracted = extract_step_log( str(log_text), step_name=step_name or failed_step, step_number=step_number, ) extracted.update( { "source": "github", "available": True, "workflow_run_id": run_id, "job_id": job_id, "job_name": job.get("name", ""), "job_conclusion": job.get("conclusion", ""), "job_steps": normalized_steps, } ) return extracted __all__ = [ "extract_step_log", "get_github_actions_step_log", "list_github_actions_active_runs", "list_github_actions_run_jobs", "list_github_actions_workflow_runs", ]