ci: integrate nemo-ci-triage with linear issues management for gitlab failures

.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

.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

.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}" \

.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

.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

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 \

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}")