diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 2eb1b43be0c..ae27dc6d2f4 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -154,6 +154,7 @@ stages: - integration_tests - functional_tests - publish + - triage default: interruptible: true @@ -268,7 +269,21 @@ variables: - "upgrade-dependencies" description: Type of publish (freeze or final release) + RUN_LINEAR_STATUS: + value: "True" + options: + - "True" + - "False" + description: Reconcile functional-test failures against Linear + RUN_LINEAR_WRITE: + value: "True" + options: + - "True" + - "False" + description: Apply proposed Linear issue opens, updates, and closes + # CI wide variables + NEMO_CI_TRIAGE_CONFIG: .gitlab/nemo-ci-triage.yml CI_MCORE_LTS_IMAGE: ${GITLAB_ENDPOINT}:5005/adlr/megatron-lm/mcore_ci_lts CI_MCORE_DEV_IMAGE: ${GITLAB_ENDPOINT}:5005/adlr/megatron-lm/mcore_ci_dev CI_NEMO_IMAGE: ${GITLAB_ENDPOINT}:5005/adlr/megatron-lm/nemo_ci @@ -282,3 +297,4 @@ include: - .gitlab/stages/03.integration-tests.yml - .gitlab/stages/04.functional-tests.yml - .gitlab/stages/05.publish.yml + - .gitlab/stages/06.triage.yml diff --git a/.gitlab/nemo-ci-triage.yml b/.gitlab/nemo-ci-triage.yml new file mode 100644 index 00000000000..a32617b2ef0 --- /dev/null +++ b/.gitlab/nemo-ci-triage.yml @@ -0,0 +1,14 @@ +# Megatron-LM configuration for nemo-ci-triage. + +gitlab: + project_id: 19378 + repo_name: ADLR/megatron-lm + +modules: + megatron_lm: + build_module: megatron-lm + team_key: MCORE + project_template: "MCore CI Testing" + enable_linear_open: true + enable_linear_modify: true + enable_linear_close: true diff --git a/.gitlab/stages/02.test.yml b/.gitlab/stages/02.test.yml index d81e1be4857..93385bd8e1b 100644 --- a/.gitlab/stages/02.test.yml +++ b/.gitlab/stages/02.test.yml @@ -195,8 +195,9 @@ test:unit_tests_notify: fi - export RO_API_TOKEN=${PROJECT_ACCESS_TOKEN_MCORE} - export GITLAB_ENDPOINT - - export TAG_TEAM=$([[ "$CI_COMMIT_BRANCH" == "main" ]] && echo "1" || "0") + - export TAG_TEAM=$([[ "$CI_COMMIT_BRANCH" == "main" ]] && echo "1" || echo "0") - export TEAM_SLUG=$SLACK_ADMIN + - export PYTHONPATH=$(pwd) - | python tests/test_utils/python_scripts/notify.py \ --pipeline-id "${CI_PIPELINE_ID}" \ diff --git a/.gitlab/stages/04.functional-tests.yml b/.gitlab/stages/04.functional-tests.yml index f83a2de4563..b7eb169acd4 100644 --- a/.gitlab/stages/04.functional-tests.yml +++ b/.gitlab/stages/04.functional-tests.yml @@ -450,6 +450,7 @@ functional:smoke_notify: fi - export RO_API_TOKEN=${PROJECT_ACCESS_TOKEN_MCORE} - export GITLAB_ENDPOINT + - export PYTHONPATH=$(pwd) - | python tests/test_utils/python_scripts/notify.py \ --pipeline-id "${CI_PIPELINE_ID}" \ @@ -494,19 +495,28 @@ functional:x_notify: - export RO_API_TOKEN=${PROJECT_ACCESS_TOKEN_MCORE} - export GITLAB_ENDPOINT - export CONTEXT=$FUNCTIONAL_TEST_SCOPE - - export TAG_TEAM=$([[ "$CI_COMMIT_BRANCH" == "main" ]] && echo "1" || "0") + - export TAG_TEAM=$([[ "$CI_COMMIT_BRANCH" == "main" ]] && echo "1" || echo "0") - export TEAM_SLUG=$SLACK_ADMIN + - export PYTHONPATH=$(pwd) - | python tests/test_utils/python_scripts/notify.py \ --pipeline-id "${CI_PIPELINE_ID}" \ --check-for functional-tests \ --pipeline-context $CONTEXT \ - --pipeline-created-at "${CI_PIPELINE_CREATED_AT}" + --pipeline-created-at "${CI_PIPELINE_CREATED_AT}" \ + --summary-output pipeline_summaries.json \ + --failure-buckets-output failure_buckets.json \ + --slack-output slack_notification.json artifacts: when: always paths: - scripts + - pipeline_summaries.json + - failure_buckets.json + - slack_notification.json + - inference_metrics.json + - agent_formatter_debug.txt rules: - if: ($CI_PIPELINE_SOURCE == "schedule" || $CI_COMMIT_BRANCH == "main") && $FUNCTIONAL_TEST == "yes" when: always diff --git a/.gitlab/stages/06.triage.yml b/.gitlab/stages/06.triage.yml new file mode 100644 index 00000000000..84996e99114 --- /dev/null +++ b/.gitlab/stages/06.triage.yml @@ -0,0 +1,95 @@ +.linear_reconcile_rules: + rules: + - if: >- + ($CI_PIPELINE_SOURCE == "schedule" || $CI_COMMIT_BRANCH == "main") && + $FUNCTIONAL_TEST == "yes" && + $RUN_LINEAR_STATUS == "True" + when: always + - when: never + +.linear_triage_job: + stage: triage + image: ${UTILITY_IMAGE}:${CI_PIPELINE_ID} + tags: + - arch/amd64 + - env/prod + - origin/jet-fleet + - owner/jet-core + - purpose/utility + - team/megatron + +triage:linear_reconcile: + extends: [.linear_triage_job, .linear_reconcile_rules] + needs: + - job: functional:x_notify + artifacts: true + script: + - >- + nemo-ci-linear status + --config "${NEMO_CI_TRIAGE_CONFIG}" + --build-module-regex '^megatron-lm$' + --output linear_status_report.json + - >- + nemo-ci-linear reconcile + --failure-buckets failure_buckets.json + --linear-report linear_status_report.json + --pipeline-summaries pipeline_summaries.json + --output linear_action_plan.json + artifacts: + when: always + paths: + - linear_status_report.json + - linear_action_plan.json + - inference_metrics.json + +triage:linear_write: + extends: [.linear_triage_job] + needs: + - job: triage:linear_reconcile + artifacts: true + allow_failure: true + script: + - >- + nemo-ci-linear write + --config "${NEMO_CI_TRIAGE_CONFIG}" + --plan linear_action_plan.json + --output linear_action_plan_post.json + artifacts: + when: always + paths: + - linear_action_plan.json + - linear_action_plan_post.json + - linear_status_report.json + rules: + - if: >- + ($CI_PIPELINE_SOURCE == "schedule" || $CI_COMMIT_BRANCH == "main") && + $FUNCTIONAL_TEST == "yes" && + $RUN_LINEAR_STATUS == "True" && + $RUN_LINEAR_WRITE == "True" + when: always + - when: never + +triage:slack_linear_followup: + extends: [.linear_triage_job] + needs: + - job: functional:x_notify + artifacts: true + - job: triage:linear_write + artifacts: true + allow_failure: true + script: + - >- + nemo-ci-notify + --pipeline-summary slack_notification.json + --linear-plan linear_action_plan_post.json + --slack-bot-token "${MCORE_SLACK_BOT_TOKEN:-${ALERTMANAGER_TOKEN}}" + --slack-channel-id "${MCORE_SLACK_CHANNEL_ID}" + rules: + # Post the applied Linear actions under the functional-test notification. + - if: >- + ($CI_PIPELINE_SOURCE == "schedule" || $CI_COMMIT_BRANCH == "main") && + $FUNCTIONAL_TEST == "yes" && + $RUN_LINEAR_STATUS == "True" && + $RUN_LINEAR_WRITE == "True" + when: always + - when: never diff --git a/docker/Dockerfile.linting b/docker/Dockerfile.linting index 15cea7b6b57..afd9260aa95 100644 --- a/docker/Dockerfile.linting +++ b/docker/Dockerfile.linting @@ -24,7 +24,7 @@ RUN --mount=type=secret,id=JET_INDEX_URLS \ # Keep this in the internal-only stage so public CI has no internal service dependency. ARG CI_SERVER_URL -ARG NEMO_CI_TRIAGE_COMMIT=8e65fa4ae20b58578d0e0f20ebea37ee7d92c8ea +ARG NEMO_CI_TRIAGE_COMMIT=5474f95417758c76c75523ae5319727a1e703437 RUN --mount=type=secret,id=NEMO_CI_TRIAGE_TOKEN \ GIT_CONFIG_COUNT=1 \ GIT_CONFIG_KEY_0=http.extraHeader \ diff --git a/tests/test_utils/python_scripts/linear_ci.py b/tests/test_utils/python_scripts/linear_ci.py new file mode 100644 index 00000000000..04f96913f5f --- /dev/null +++ b/tests/test_utils/python_scripts/linear_ci.py @@ -0,0 +1,192 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + +"""Megatron-LM adapters for nemo-ci-triage's failure-reporting workflow. + +The triage package owns LLM summarization, Linear reconciliation, and Slack +follow-up logic. This module only converts Megatron-LM's direct child-pipeline +jobs into the generic failure records consumed by the package summarizer. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any, Callable + +from nemo_ci_triage.agent import summarize_pipeline_failures as summarizer + +LINEAR_MODULE = "megatron_lm" +_FUNCTIONAL_PREFIX = "functional:run_" + + +def _variant_name(pipeline_name: str) -> str: + """Return the stable environment/platform suffix of a functional bridge.""" + return pipeline_name.removeprefix(_FUNCTIONAL_PREFIX).replace("_", "-") + + +def _recipe_name(pipeline_name: str, config_name: str) -> str: + """Disambiguate the same recipe across dev/LTS and GPU child pipelines.""" + return f"{config_name}@{_variant_name(pipeline_name)}" + + +def _job_url(project_url: str, job: dict) -> str: + return job.get("web_url") or f"{project_url}/-/jobs/{job['id']}" + + +def _failure_record(pipeline_name: str, job: dict, report: dict | None, project_url: str) -> dict: + """Return the raw failure shape accepted by the upstream LLM summarizer.""" + return { + "test_name": _recipe_name(pipeline_name, job["config_name"]), + "module": LINEAR_MODULE, + "report": report, + "job_url": _job_url(project_url, job), + "job_error_type": job.get("error_type"), + } + + +def _fallback_summary(failure: dict) -> dict: + """Preserve a failed test when its per-test LLM summary is unavailable.""" + report = failure.get("report") or {} + category = ( + report.get("error_type") + or report.get("category") + or failure.get("job_error_type") + or "Unknown" + ) + subtype = report.get("error_subtype") or failure.get("job_error_type") + subtype = subtype or (f"No structured error report was available for {failure['test_name']}") + summary = subtype if subtype == category else f"{category}: {subtype}" + return { + "test_name": failure["test_name"], + "module": failure["module"], + "category": category, + "summary": summary, + "excerpt": report.get("excerpt"), + "job_url": failure["job_url"], + } + + +def _summarize_failures(raw_failures: list[dict]) -> list[dict]: + """Use upstream LLM summaries, falling back without dropping failures.""" + with_reports = [failure for failure in raw_failures if failure.get("report")] + summarized = summarizer._summarize_failures( + with_reports, summarizer._SUMMARIZER_PROMPT.read_text(encoding="utf-8").strip() + ) + by_job = {(failure["test_name"], failure["job_url"]): failure for failure in summarized} + return [ + by_job.get((failure["test_name"], failure["job_url"]), _fallback_summary(failure)) + for failure in raw_failures + ] + + +def build_pipeline_reports( + pipeline_id: int, + scope: str, + pipeline_jobs: list[tuple[str, int, list[dict]]], + load_error_report: Callable[[int], dict | None], + project_url: str, +) -> tuple[dict, dict]: + """Build the two JSON contracts consumed by nemo-ci-triage reconciliation. + + Each recipe is qualified by its child-pipeline variant. A recipe is only + included in ``passed_tests`` when that exact variant completed successfully; + failed, canceled, and ambiguous allow-failure jobs can therefore never close + a live Linear issue accidentally. + """ + passed: set[str] = set() + unknown: set[str] = set() + raw_failures: list[dict] = [] + failed_jobs = 0 + + for pipeline_name, _, jobs in sorted(pipeline_jobs, key=lambda item: item[0]): + for job in sorted(jobs, key=lambda item: (item["config_name"], item["id"])): + recipe = _recipe_name(pipeline_name, job["config_name"]) + status = job.get("status") + report = None + + if status == "failed" or (status == "success" and job.get("allow_failure")): + report = load_error_report(job["id"]) + + suppressed_failure = bool( + status == "success" and report and report.get("exit_code_training") not in (None, 0) + ) + if status == "failed" or suppressed_failure: + failed_jobs += 1 + raw_failures.append(_failure_record(pipeline_name, job, report, project_url)) + elif status == "success" and (not job.get("allow_failure") or report is not None): + passed.add(recipe) + else: + unknown.add(recipe) + + failed_recipes = {failure["test_name"] for failure in raw_failures} + passed_tests = sorted(passed - failed_recipes - unknown) + + failures = _summarize_failures(raw_failures) + buckets, failed_stage = summarizer._subcategorize(failures) + bucketing_failed = buckets is None + if bucketing_failed: + print( + f"WARNING: LLM categorizer failed at {failed_stage}; " + "Linear reconciliation will skip this report", + file=sys.stderr, + ) + buckets = [] + else: + summarizer._attach_categories(buckets, failures) + + digest = summarizer._digest( + failures, {LINEAR_MODULE: {"passed": len(passed_tests), "failed": failed_jobs}} + ) + + module_stats = { + "passed": len(passed_tests), + "failed": failed_jobs, + "passed_tests": passed_tests, + } + summaries = { + "pipeline_id": pipeline_id, + "scope": scope, + "modules": {LINEAR_MODULE: module_stats}, + "digest": digest, + "failures": failures, + } + failure_buckets = { + "pipeline_id": pipeline_id, + "bucketing_failed": bucketing_failed, + "buckets": summarizer._denormalize_buckets(buckets, failures), + } + return summaries, failure_buckets + + +def fetch_error_report(project: Any, job_id: int) -> dict | None: + """Fetch one child job's structured report, degrading safely if absent.""" + try: + raw = project.jobs.get(job_id, lazy=True).artifact("error_report.json") + if isinstance(raw, bytes): + raw = raw.decode("utf-8") + return json.loads(raw) + except Exception as exc: + print(f"WARNING: job {job_id}: could not read error_report.json: {exc}", file=sys.stderr) + return None + + +def write_pipeline_reports( + pipeline_id: int, + scope: str, + pipeline_jobs: list[tuple[str, int, list[dict]]], + project: Any, + project_url: str, + summaries_path: Path, + buckets_path: Path, +) -> None: + summaries, buckets = build_pipeline_reports( + pipeline_id, + scope, + pipeline_jobs, + lambda job_id: fetch_error_report(project, job_id), + project_url, + ) + summaries_path.write_text(json.dumps(summaries, indent=2) + "\n", encoding="utf-8") + buckets_path.write_text(json.dumps(buckets, indent=2) + "\n", encoding="utf-8") + print(f"Wrote {summaries_path} and {buckets_path}") diff --git a/tests/test_utils/python_scripts/notify.py b/tests/test_utils/python_scripts/notify.py index db875271d89..cbb9bf527f7 100644 --- a/tests/test_utils/python_scripts/notify.py +++ b/tests/test_utils/python_scripts/notify.py @@ -1,19 +1,28 @@ # Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +import json import logging import os +from pathlib import Path +from typing import Any import click import gitlab from nemo_ci_triage.slack_notification import notification +from nemo_ci_triage.slack_notification.utils import repository_settings -PROJECT_ID = int(os.getenv("CI_PROJECT_ID", 19378)) +from tests.test_utils.python_scripts import linear_ci + +TRIAGE_CONFIG = Path(os.getenv("NEMO_CI_TRIAGE_CONFIG", ".gitlab/nemo-ci-triage.yml")) +PROJECT_ID, REPO_NAME = repository_settings(TRIAGE_CONFIG) WEBHOOK_URL = os.getenv("WEBHOOK_URL", "") +SLACK_BOT_TOKEN = os.getenv("MCORE_SLACK_BOT_TOKEN") or os.getenv("ALERTMANAGER_TOKEN", "") +SLACK_CHANNEL_ID = os.getenv("MCORE_SLACK_CHANNEL_ID", "") GITLAB_ENDPOINT = os.getenv("GITLAB_ENDPOINT") if not GITLAB_ENDPOINT: raise ValueError("GITLAB_ENDPOINT is required") SERVER_URL = f"https://{GITLAB_ENDPOINT}" -PROJECT_URL = os.getenv("CI_PROJECT_URL", f"{SERVER_URL}/ADLR/megatron-lm") +PROJECT_URL = os.getenv("CI_PROJECT_URL", f"{SERVER_URL}/{REPO_NAME}") TAG_TEAM = os.getenv("TAG_TEAM", "0") == "1" TEAM_SLUG = os.getenv("TEAM_SLUG", "") @@ -33,6 +42,11 @@ def get_gitlab_handle() -> gitlab.Gitlab: return gitlab.Gitlab(SERVER_URL, private_token=os.getenv("RO_API_TOKEN")) +def get_project() -> Any: + """Return the configured Megatron-LM GitLab project.""" + return get_gitlab_handle().projects.get(PROJECT_ID) + + def _bridge_gpu(bridge_name: str) -> str: for gpu in ("GB200", "H100", "A100"): if gpu.lower() in bridge_name.lower(): @@ -40,9 +54,11 @@ def _bridge_gpu(bridge_name: str) -> str: return "Unknown" -def get_pipeline_jobs(pipeline_id: int, job_prefix: str) -> list[tuple[str, int, list[dict]]]: +def get_pipeline_jobs( + pipeline_id: int, job_prefix: str, project: Any | None = None +) -> list[tuple[str, int, list[dict]]]: """Collect Megatron-LM's direct child pipelines using nemo-ci-triage-2.""" - project = get_gitlab_handle().projects.get(PROJECT_ID) + project = project or get_project() root_pipeline = project.pipelines.get(pipeline_id) pipeline_jobs = [] @@ -62,10 +78,13 @@ def get_pipeline_jobs(pipeline_id: int, job_prefix: str) -> list[tuple[str, int, return pipeline_jobs -def configure_notification_urls() -> None: - """Point nemo-ci-triage-2's notification links at Megatron-LM.""" - notification.JOB_URL_TEMPLATE = f"{PROJECT_URL}/-/jobs/{{}}" - notification.PIPELINE_URL_TEMPLATE = f"{PROJECT_URL}/-/pipelines/{{}}" +def write_slack_context(output: Path | None, thread_timestamp: str | None) -> None: + """Persist the non-secret Slack coordinates needed by a follow-up job.""" + if output is None: + return + context = {"channel_id": SLACK_CHANNEL_ID or None, "thread_timestamp": thread_timestamp} + output.write_text(json.dumps(context, indent=2) + "\n", encoding="utf-8") + logger.info("Wrote Slack thread context to %s", output) @click.command() @@ -77,23 +96,66 @@ def configure_notification_urls() -> None: ) @click.option("--pipeline-context", required=True, type=str) @click.option("--pipeline-created-at", required=True, type=str, expose_value=False) -def main(pipeline_id: int, check_for: str, pipeline_context: str) -> None: - pipeline_jobs = get_pipeline_jobs(pipeline_id, JOB_PREFIXES[check_for]) +@click.option("--summary-output", type=click.Path(path_type=Path), default=None) +@click.option("--failure-buckets-output", type=click.Path(path_type=Path), default=None) +@click.option("--slack-output", type=click.Path(path_type=Path), default=None) +def main( + pipeline_id: int, + check_for: str, + pipeline_context: str, + summary_output: Path | None, + failure_buckets_output: Path | None, + slack_output: Path | None, +) -> None: + if bool(summary_output) != bool(failure_buckets_output): + raise click.UsageError( + "--summary-output and --failure-buckets-output must be provided together" + ) + + project = get_project() + pipeline_jobs = get_pipeline_jobs(pipeline_id, JOB_PREFIXES[check_for], project=project) + + if summary_output: + linear_ci.write_pipeline_reports( + pipeline_id, + pipeline_context, + pipeline_jobs, + project, + PROJECT_URL, + summary_output, + failure_buckets_output, + ) if check_for == "smoke-tests": if all(job["status"] == "success" for _, _, jobs in pipeline_jobs for job in jobs): logger.info("All smoke tests passed, skipping Slack notification") + write_slack_context(slack_output, None) return - if not WEBHOOK_URL: - logger.info("No webhook URL configured, skipping Slack notification") + use_bot = bool(SLACK_BOT_TOKEN and SLACK_CHANNEL_ID) + if bool(SLACK_BOT_TOKEN) != bool(SLACK_CHANNEL_ID): + logger.warning( + "Both MCORE_SLACK_BOT_TOKEN (or ALERTMANAGER_TOKEN) and " + "MCORE_SLACK_CHANNEL_ID are required for threaded Slack replies" + ) + + if not WEBHOOK_URL and not use_bot: + logger.info("No Slack bot or webhook configured, skipping Slack notification") + write_slack_context(slack_output, None) return - configure_notification_urls() slack_mentions = f"{TEAM_SLUG} <@U09TX0DHZ97>" if TAG_TEAM else None - notification.send_slack_notification( - "megatron-lm", pipeline_context, pipeline_jobs, slack_mentions, webhook_url=WEBHOOK_URL + thread_timestamp = notification.send_slack_notification( + "megatron-lm", + pipeline_context, + pipeline_jobs, + slack_mentions, + webhook_url=WEBHOOK_URL or None, + slack_bot_token=SLACK_BOT_TOKEN if use_bot else None, + slack_channel_id=SLACK_CHANNEL_ID if use_bot else None, + config=TRIAGE_CONFIG, ) + write_slack_context(slack_output, thread_timestamp) if __name__ == "__main__": diff --git a/tests/test_utils/test_ci_triage.py b/tests/test_utils/test_ci_triage.py index 74ca1bc354a..2c8ec07b720 100644 --- a/tests/test_utils/test_ci_triage.py +++ b/tests/test_utils/test_ci_triage.py @@ -1,5 +1,7 @@ # Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +import json +from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock @@ -7,12 +9,42 @@ import yaml from click.testing import CliRunner -from tests.test_utils.python_scripts import generate_jet_trigger_job, recipe_parser +from tests.test_utils.python_scripts import generate_jet_trigger_job, linear_ci, recipe_parser + + +def _mock_llm_reporting(monkeypatch): + summarize = Mock( + side_effect=lambda failures, _prompt: [ + linear_ci._fallback_summary(failure) for failure in failures + ] + ) + + def group_failures(failures): + grouped = {} + for failure in failures: + grouped.setdefault((failure["category"], failure["summary"]), []).append( + failure["test_name"] + ) + return ( + [ + {"label": f"test-bucket-{index}", "rationale": summary, "tests": tests} + for index, ((_, summary), tests) in enumerate(grouped.items(), 1) + ], + None, + ) + + subcategorize = Mock(side_effect=group_failures) + digest = Mock(return_value="LLM pipeline digest") + monkeypatch.setattr(linear_ci.summarizer, "_summarize_failures", summarize) + monkeypatch.setattr(linear_ci.summarizer, "_subcategorize", subcategorize) + monkeypatch.setattr(linear_ci.summarizer, "_digest", digest) + return summarize, subcategorize, digest @pytest.fixture -def notify_module(): +def notify_module(monkeypatch): pytest.importorskip("nemo_ci_triage.slack_notification") + monkeypatch.setenv("GITLAB_ENDPOINT", "ci.example.com") from tests.test_utils.python_scripts import notify return notify @@ -87,6 +119,49 @@ def test_error_extraction_is_opt_in_for_generated_jobs( assert job["artifacts"]["paths"] == ["results/"] +def test_notification_rules_use_expected_pipeline_sources(): + unit = yaml.safe_load(Path(".gitlab/stages/02.test.yml").read_text()) + functional = yaml.safe_load(Path(".gitlab/stages/04.functional-tests.yml").read_text()) + triage = yaml.safe_load(Path(".gitlab/stages/06.triage.yml").read_text()) + + unit_conditions = [ + rule["if"] for rule in unit["test:unit_tests_notify"]["rules"] if "if" in rule + ] + assert unit_conditions == [ + '$CI_PIPELINE_SOURCE == "schedule" && ' + '($CI_COMMIT_BRANCH == "ci-unit-test-extended" || ' + '$CI_COMMIT_BRANCH == "ci-dev-unit-test-extended")' + ] + + smoke_condition = functional["functional:smoke_notify"]["rules"][1]["if"] + assert smoke_condition == ( + '$FUNCTIONAL_TEST == "yes" && $FUNCTIONAL_TEST_SCOPE =~ /^(mr|nightly)$/ && ' + '($CI_PIPELINE_SOURCE == "schedule" || $CI_COMMIT_BRANCH == "main" || ' + '$CI_MERGE_REQUEST_EVENT_TYPE == "merged_result")' + ) + assert functional["functional:x_notify"]["rules"][0]["if"] == ( + '($CI_PIPELINE_SOURCE == "schedule" || $CI_COMMIT_BRANCH == "main") && ' + '$FUNCTIONAL_TEST == "yes"' + ) + + triage_jobs = (".linear_reconcile_rules", "triage:linear_write", "triage:slack_linear_followup") + for job_name in triage_jobs: + condition = triage[job_name]["rules"][0]["if"] + assert '$FUNCTIONAL_TEST == "yes"' in condition + assert '$CI_PIPELINE_SOURCE == "schedule"' in condition + assert '$CI_COMMIT_BRANCH == "main"' in condition + + +def test_all_generated_test_types_enable_error_extraction(): + unit = Path(".gitlab/stages/02.test.yml").read_text() + integration = Path(".gitlab/stages/03.integration-tests.yml").read_text() + functional = Path(".gitlab/stages/04.functional-tests.yml").read_text() + + assert unit.count('"--enable-error-extraction"') >= 1 + assert integration.count('"--enable-error-extraction"') >= 1 + assert functional.count('"--enable-error-extraction"') >= 2 + + def test_get_pipeline_jobs_uses_triage_collector(monkeypatch, notify_module): notify = notify_module bridge = SimpleNamespace( @@ -110,14 +185,180 @@ def test_get_pipeline_jobs_uses_triage_collector(monkeypatch, notify_module): collector.assert_called_once_with(project, 101) +def test_build_linear_reports_groups_matching_failures(monkeypatch): + summarize, subcategorize, digest = _mock_llm_reporting(monkeypatch) + pipeline_jobs = [ + ( + "functional:run_dev_dgx_h100", + 101, + [ + { + "config_name": "gpt_pass", + "id": 1, + "status": "success", + "allow_failure": False, + "error_type": None, + }, + { + "config_name": "gpt_fail_a", + "id": 2, + "status": "failed", + "allow_failure": False, + "error_type": "CUDA OOM", + }, + ], + ), + ( + "functional:run_lts_dgx_h100", + 102, + [ + { + "config_name": "gpt_fail_b", + "id": 3, + "status": "failed", + "allow_failure": True, + "error_type": None, + } + ], + ), + ] + reports = { + 2: { + "exit_code_training": 1, + "category": "CUDA OOM", + "error_subtype": "torch.OutOfMemoryError", + "excerpt": "CUDA out of memory", + }, + 3: { + "exit_code_training": 1, + "category": "CUDA OOM", + "error_subtype": "torch.OutOfMemoryError", + "excerpt": "CUDA out of memory", + }, + } + + summaries, buckets = linear_ci.build_pipeline_reports( + 123, "nightly", pipeline_jobs, reports.get, "https://ci.example.com/ADLR/megatron-lm" + ) + + stats = summaries["modules"][linear_ci.LINEAR_MODULE] + assert stats == {"passed": 1, "failed": 2, "passed_tests": ["gpt_pass@dev-dgx-h100"]} + assert len(buckets["buckets"]) == 1 + bucket = buckets["buckets"][0] + assert bucket["category"] == "CUDA OOM" + assert bucket["rationale"] == "CUDA OOM: torch.OutOfMemoryError" + assert bucket["tests"] == [ + { + "name": "gpt_fail_a@dev-dgx-h100", + "job_url": "https://ci.example.com/ADLR/megatron-lm/-/jobs/2", + }, + { + "name": "gpt_fail_b@lts-dgx-h100", + "job_url": "https://ci.example.com/ADLR/megatron-lm/-/jobs/3", + }, + ] + summarize.assert_called_once() + subcategorize.assert_called_once() + digest.assert_called_once() + + +def test_allow_failure_without_report_is_not_counted_as_passed(monkeypatch): + _mock_llm_reporting(monkeypatch) + pipeline_jobs = [ + ( + "functional:run_dev_dgx_h100", + 101, + [ + { + "config_name": "ambiguous", + "id": 4, + "status": "success", + "allow_failure": True, + "error_type": None, + } + ], + ) + ] + + summaries, buckets = linear_ci.build_pipeline_reports( + 123, + "nightly", + pipeline_jobs, + lambda _job_id: None, + "https://ci.example.com/ADLR/megatron-lm", + ) + + stats = summaries["modules"][linear_ci.LINEAR_MODULE] + assert stats["passed_tests"] == [] + assert stats["failed"] == 0 + assert buckets["buckets"] == [] + + +def test_failed_job_without_report_still_creates_a_safe_bucket(monkeypatch): + _mock_llm_reporting(monkeypatch) + pipeline_jobs = [ + ( + "functional:run_dev_dgx_h100", + 101, + [ + { + "config_name": "missing_report", + "id": 5, + "status": "failed", + "allow_failure": False, + "error_type": None, + } + ], + ) + ] + + summaries, buckets = linear_ci.build_pipeline_reports( + 123, + "nightly", + pipeline_jobs, + lambda _job_id: None, + "https://ci.example.com/ADLR/megatron-lm", + ) + + assert summaries["modules"][linear_ci.LINEAR_MODULE]["failed"] == 1 + assert buckets["buckets"][0]["tests"][0]["name"] == "missing_report@dev-dgx-h100" + assert "No structured error report" in buckets["buckets"][0]["rationale"] + + +def test_triage_config_selects_megatron_and_enables_write_actions(): + linear_status = pytest.importorskip("nemo_ci_triage.linear.linear_status") + linear_write = pytest.importorskip("nemo_ci_triage.linear.linear_write") + config = Path(".gitlab/nemo-ci-triage.yml") + + assert linear_status.modules_for_regex("^megatron-lm$", config) == [ + ( + linear_ci.LINEAR_MODULE, + { + "build_module": "megatron-lm", + "team_key": "MCORE", + "project_template": "MCore CI Testing", + "enable_linear_open": True, + "enable_linear_modify": True, + "enable_linear_close": True, + }, + ) + ] + assert linear_write.write_gates(config) == { + linear_ci.LINEAR_MODULE: {"open": True, "modify": True, "close": True} + } + + def test_notification_delegates_to_triage_package(monkeypatch, notify_module): notify = notify_module pipeline_jobs = [("functional:run_dev_dgx_h100", 101, [{"status": "failed"}])] sender = Mock() monkeypatch.setattr(notify, "WEBHOOK_URL", "https://slack.invalid/webhook") + monkeypatch.setattr(notify, "SLACK_BOT_TOKEN", "") + monkeypatch.setattr(notify, "SLACK_CHANNEL_ID", "") monkeypatch.setattr(notify, "PROJECT_URL", "https://ci.example.com/ADLR/megatron-lm") - monkeypatch.setattr(notify, "get_pipeline_jobs", lambda *_args: pipeline_jobs) + monkeypatch.setattr(notify, "get_project", Mock()) + monkeypatch.setattr(notify, "get_pipeline_jobs", lambda *_args, **_kwargs: pipeline_jobs) monkeypatch.setattr(notify.notification, "send_slack_notification", sender) result = CliRunner().invoke( @@ -135,13 +376,98 @@ def test_notification_delegates_to_triage_package(monkeypatch, notify_module): ) assert result.exit_code == 0, result.output - assert ( - notify.notification.JOB_URL_TEMPLATE == "https://ci.example.com/ADLR/megatron-lm/-/jobs/{}" + sender.assert_called_once_with( + "megatron-lm", + "mr", + pipeline_jobs, + None, + webhook_url="https://slack.invalid/webhook", + slack_bot_token=None, + slack_channel_id=None, + config=notify.TRIAGE_CONFIG, ) - assert ( - notify.notification.PIPELINE_URL_TEMPLATE - == "https://ci.example.com/ADLR/megatron-lm/-/pipelines/{}" + + +def test_notification_records_bot_thread_context(monkeypatch, tmp_path, notify_module): + notify = notify_module + pipeline_jobs = [("functional:run_dev_dgx_h100", 101, [{"status": "failed"}])] + sender = Mock(return_value="1712345678.000100") + slack_output = tmp_path / "slack_notification.json" + + monkeypatch.setattr(notify, "WEBHOOK_URL", "") + monkeypatch.setattr(notify, "SLACK_BOT_TOKEN", "xoxb-test") + monkeypatch.setattr(notify, "SLACK_CHANNEL_ID", "C0123456789") + monkeypatch.setattr(notify, "get_project", Mock()) + monkeypatch.setattr(notify, "get_pipeline_jobs", lambda *_args, **_kwargs: pipeline_jobs) + monkeypatch.setattr(notify.notification, "send_slack_notification", sender) + + result = CliRunner().invoke( + notify.main, + [ + "--pipeline-id", + "123", + "--check-for", + "functional-tests", + "--pipeline-context", + "mr", + "--pipeline-created-at", + "2026-07-12T00:00:00Z", + "--slack-output", + str(slack_output), + ], ) + + assert result.exit_code == 0, result.output sender.assert_called_once_with( - "megatron-lm", "mr", pipeline_jobs, None, webhook_url="https://slack.invalid/webhook" + "megatron-lm", + "mr", + pipeline_jobs, + None, + webhook_url=None, + slack_bot_token="xoxb-test", + slack_channel_id="C0123456789", + config=notify.TRIAGE_CONFIG, + ) + assert json.loads(slack_output.read_text()) == { + "channel_id": "C0123456789", + "thread_timestamp": "1712345678.000100", + } + + +def test_notification_writes_linear_inputs_without_webhook(monkeypatch, tmp_path, notify_module): + notify = notify_module + project = Mock() + pipeline_jobs = [("functional:run_dev_dgx_h100", 101, [])] + writer = Mock() + + monkeypatch.setattr(notify, "WEBHOOK_URL", "") + monkeypatch.setattr(notify, "SLACK_BOT_TOKEN", "") + monkeypatch.setattr(notify, "SLACK_CHANNEL_ID", "") + monkeypatch.setattr(notify, "get_project", lambda: project) + monkeypatch.setattr(notify, "get_pipeline_jobs", lambda *_args, **_kwargs: pipeline_jobs) + monkeypatch.setattr(notify.linear_ci, "write_pipeline_reports", writer) + summaries = tmp_path / "pipeline_summaries.json" + buckets = tmp_path / "failure_buckets.json" + + result = CliRunner().invoke( + notify.main, + [ + "--pipeline-id", + "123", + "--check-for", + "functional-tests", + "--pipeline-context", + "nightly", + "--pipeline-created-at", + "2026-07-12T00:00:00Z", + "--summary-output", + str(summaries), + "--failure-buckets-output", + str(buckets), + ], + ) + + assert result.exit_code == 0, result.output + writer.assert_called_once_with( + 123, "nightly", pipeline_jobs, project, notify.PROJECT_URL, summaries, buckets )