Pipeline Orchestrator Agent

def devops_parse_ci_config_tool(yaml_or_json_text: str) -> Dict[str, Any]:
    """
    Extract jobs/steps from a CI config(CircleCI/GitHub Actions-like).
    Returns: {"jobs":[{"name":..., "steps":[...]}]}
    """
    data = _extract_yaml_or_json(yaml_or_json_text) or {}
    jobs: List[Dict[str, Any]] = []
    for name, job in (data.get("jobs") or {}).items():
        steps: List[str] = []
        for st in job.get("steps", []):
            if isinstance(st, dict) and st:
                k = list(st.keys())[0]
                steps.append(k)
            else:
                steps.append(str(st))
        jobs.append({"name": name, "steps": steps})
    return {"jobs": jobs}

def devops_junit_summary_tool(xml_text: str) -> Dict[str, Any]:
    """
    Summarize a JUnit XML report: tests, failures, errors, skipped.
    Returns: {"tests": int, "failures": int, "errors": int, "skipped": int}
    """
    if not xml_text:
        return {"tests": 0, "failures": 0, "errors": 0, "skipped": 0}
    try:
        root = ET.fromstring(xml_text)
        if root.tag.endswith("testsuite"):
            return {
                "tests": int(root.attrib.get("tests", 0)),
                "failures": int(root.attrib.get("failures", 0)),
                "errors": int(root.attrib.get("errors", 0)),
                "skipped": int(root.attrib.get("skipped", 0)),
            }
        total = {"tests": 0, "failures": 0, "errors": 0, "skipped": 0}
        for ts in root.findall(".//testsuite"):
            total["tests"] += int(ts.attrib.get("tests", 0))
            total["failures"] += int(ts.attrib.get("failures", 0))
            total["errors"] += int(ts.attrib.get("errors", 0))
            total["skipped"] += int(ts.attrib.get("skipped", 0))
        return total
    except Exception:
        return {"tests": 0, "failures": 0, "errors": 0, "skipped": 0}