MUTE: multi build mute (#38710)

7 files changed, 489 insertions, 89 deletions

diff --git a/.github/scripts/tests/junit-postprocess.py b/.github/scripts/tests/junit-postprocess.py
index df8d07b0f3e..3bd11c05511 100755
--- a/.github/scripts/tests/junit-postprocess.py
+++ b/.github/scripts/tests/junit-postprocess.py
@@ -4,7 +4,7 @@ import glob
 import os
 import re
 import xml.etree.ElementTree as ET
-from mute_utils import MuteTestCheck, mute_target, recalc_suite_info
+from mute.mute_utils import MuteTestCheck, mute_target, recalc_suite_info
 
 
 shard_suffix_re = re.compile(r"-\d+$")
diff --git a/.github/scripts/tests/mute/create_new_muted_ya.py b/.github/scripts/tests/mute/create_new_muted_ya.py
index 79b56b21cad..644fc311b37 100755
--- a/.github/scripts/tests/mute/create_new_muted_ya.py
+++ b/.github/scripts/tests/mute/create_new_muted_ya.py
@@ -17,7 +17,7 @@ for _p in (_tests_dir, _scripts_dir, os.path.join(_scripts_dir, 'analytics')):
     if _p not in sys.path:
         sys.path.insert(0, _p)
 
-from mute_check import YaMuteCheck
+from mute.mute_check import YaMuteCheck
 from mute.update_mute_issues import (
     ORG_NAME,
     PROJECT_ID,
@@ -36,6 +36,7 @@ from mute.constants import (
     get_unmute_window_days,
 )
 from mute.naming import mute_file_line_to_tests_monitor_full_name
+from mute.mute_utils import dedicated_relative
 from ydb_wrapper import YDBWrapper
 from github_issue_utils import DEFAULT_BUILD_TYPE, canonical_team_slug, make_profile_id
 
@@ -46,7 +47,6 @@ for _noisy in ('grpc', 'grpc._cython.cygrpc', 'ydb'):
 
 dir = os.path.dirname(__file__)
 repo_path = os.path.normpath(os.path.join(dir, '..', '..', '..', '..')) + os.sep
-muted_ya_path = '.github/config/muted_ya.txt'
 
 _DIGEST_NOTIFICATION_CONFIG = os.path.normpath(
     os.path.join(dir, '..', '..', '..', 'config', 'mute_issue_and_digest_config.json')
@@ -57,6 +57,12 @@ def load_manual_unmute_config():
     return get_manual_unmute_window_days(), get_manual_unmute_min_runs()
 
 
+def resolve_muted_ya_path(explicit_path: str | None, build_type: str) -> str:
+    if explicit_path and str(explicit_path).strip():
+        return str(explicit_path).strip()
+    return dedicated_relative(build_type)
+
+
 def tests_monitor_query_days_window():
     """How many calendar days of ``tests_monitor`` history we must load for mute/unmute/delete/fast-unmute."""
     return max(
@@ -769,9 +775,6 @@ def apply_and_add_mutes(
         write_file_set(os.path.join(output_path, 'to_mute.txt'), to_mute, to_mute_debug)
         write_file_set(os.path.join(output_path, 'to_unmute.txt'), to_unmute, to_unmute_debug)
 
-        if ydb_wrapper is not None and branch is not None and build_type is not None:
-            delete_fast_unmute_grace_rows(ydb_wrapper, branch, build_type, to_mute)
-
         # 3. Delete-from-mute candidates (to_delete).
         def is_delete_non_chunk(test):
             if is_chunk_test(test):
@@ -937,7 +940,36 @@ def read_tests_from_file(file_path):
     return result
 
 
-def create_mute_issues(all_tests, file_path, close_issues=True, branch='main', build_type=DEFAULT_BUILD_TYPE):
+def _format_issue_date_window(value):
+    if value is None:
+        return 'N/A'
+    if isinstance(value, datetime.datetime):
+        return value.date().strftime('%Y-%m-%d')
+    if isinstance(value, datetime.date):
+        return value.strftime('%Y-%m-%d')
+    if isinstance(value, int):
+        base_date = datetime.date(1970, 1, 1)
+        return (base_date + datetime.timedelta(days=value)).strftime('%Y-%m-%d')
+    return str(value)
+
+
+def _compute_summary_from_counts(row):
+    pass_count = int(row.get('pass_count') or 0)
+    fail_count = int(row.get('fail_count') or 0)
+    mute_count = int(row.get('mute_count') or 0)
+    skip_count = int(row.get('skip_count') or 0)
+    total_runs = pass_count + fail_count + mute_count + skip_count
+    return f"p-{pass_count}, f-{fail_count}, m-{mute_count}, s-{skip_count}, total-{total_runs}"
+
+
+def create_mute_issues(
+    all_tests,
+    file_path,
+    aggregated_tests,
+    close_issues=True,
+    branch='main',
+    build_type=DEFAULT_BUILD_TYPE,
+):
     tests_from_file = read_tests_from_file(file_path)
     issues_index = get_issues_and_tests_from_project(ORG_NAME, PROJECT_ID)
     muted_tests_in_issues = get_muted_tests_from_issues(issues_index)
@@ -959,6 +991,11 @@ def create_mute_issues(all_tests, file_path, close_issues=True, branch='main', b
         if t.get('full_name'):
             bt = t.get('build_type') or DEFAULT_BUILD_TYPE
             monitor_by_name[(t['full_name'], bt)] = t
+    aggregated_by_name = {}
+    for t in aggregated_tests:
+        if t.get('full_name'):
+            bt = t.get('build_type') or DEFAULT_BUILD_TYPE
+            aggregated_by_name[(t['full_name'], bt)] = t
 
     for test_from_file in tests_from_file:
         full_name = test_from_file['full_name']
@@ -984,23 +1021,51 @@ def create_mute_issues(all_tests, file_path, close_issues=True, branch='main', b
             continue
 
         monitor = monitor_by_name.get((full_name, build_type))
+        aggregated = aggregated_by_name.get((full_name, build_type))
         if monitor and is_chunk_test(monitor):
             logging.info(f"Skipping chunk test: {full_name}")
             continue
         if monitor:
+            if not aggregated:
+                logging.warning(
+                    "No aggregated row for %s (%s): using raw monitor fields",
+                    full_name,
+                    build_type,
+                )
+            days_in_state = monitor.get('days_in_state')
+            if days_in_state is None:
+                logging.warning(
+                    "Raw monitor row for %s (%s) has no days_in_state: using 0",
+                    full_name,
+                    build_type,
+                )
+                days_in_state = 0
+            source = aggregated or monitor
+            success_rate = source.get('success_rate')
+            if success_rate is None:
+                pass_count = int(source.get('pass_count') or 0)
+                fail_count = int(source.get('fail_count') or 0)
+                mute_count = int(source.get('mute_count') or 0)
+                skip_count = int(source.get('skip_count') or 0)
+                total_runs = pass_count + fail_count + mute_count + skip_count
+                success_rate = round((pass_count / total_runs) * 100, 1) if total_runs > 0 else 0.0
+            summary = source.get('summary') or _compute_summary_from_counts(source)
+            state = source.get('state') or monitor.get('state') or 'Muted'
+            date_window = source.get('date_window') or monitor.get('date_window')
+
             entry = {
                 'mute_string': f"{monitor.get('suite_folder')} {monitor.get('test_name')}",
                 'test_name': monitor.get('test_name'),
                 'suite_folder': monitor.get('suite_folder'),
                 'full_name': full_name,
-                'success_rate': monitor.get('success_rate'),
-                'days_in_state': monitor.get('days_in_state'),
-                'date_window': monitor.get('date_window', 'N/A'),
+                'success_rate': success_rate,
+                'days_in_state': days_in_state,
+                'date_window': _format_issue_date_window(date_window),
                 'owner': monitor.get('owner'),
-                'state': monitor.get('state'),
-                'summary': monitor.get('summary'),
-                'fail_count': monitor.get('fail_count'),
-                'pass_count': monitor.get('pass_count'),
+                'state': state,
+                'summary': summary,
+                'fail_count': source.get('fail_count'),
+                'pass_count': source.get('pass_count'),
                 'branch': monitor.get('branch'),
                 'build_type': monitor.get('build_type', DEFAULT_BUILD_TYPE),
             }
@@ -1153,11 +1218,7 @@ def create_mute_issues(all_tests, file_path, close_issues=True, branch='main', b
     return queue_items
 
 
-def load_configured_digest_profile_ids():
-    """Profile IDs listed in mute_issue_and_digest_config.json (branch:build_type).
-
-    Only these may be written to digest_queue so unsent rows always match a send_digest profile.
-    """
+def _load_issue_digest_profiles():
     try:
         with open(_DIGEST_NOTIFICATION_CONFIG, 'r', encoding='utf-8') as f:
             data = json.load(f)
@@ -1173,8 +1234,13 @@ def load_configured_digest_profile_ids():
             _DIGEST_NOTIFICATION_CONFIG,
             exc,
         )
-        return set()
-    profiles = data.get('profiles') or []
+        return []
+    return data.get('profiles') or []
+
+
+def load_configured_issue_profile_ids():
+    """Profiles allowed for issue sync (branch:build_type)."""
+    profiles = _load_issue_digest_profiles()
     ids = {
         make_profile_id(p['branch'], p['build_type'])
         for p in profiles
@@ -1182,7 +1248,30 @@ def load_configured_digest_profile_ids():
     }
     if not ids:
         logging.warning(
-            'No profiles in %s — not enqueueing to digest_queue',
+            'No profiles in %s — skipping issue creation',
+            _DIGEST_NOTIFICATION_CONFIG,
+        )
+    return ids
+
+
+def load_configured_digest_profile_ids():
+    """Profiles allowed for digest queue enqueue (branch:build_type).
+
+    Enqueue is enabled only for profiles that define digest schedule
+    via non-empty "schedule_utc_hours".
+    """
+    profiles = _load_issue_digest_profiles()
+    ids = {
+        make_profile_id(p['branch'], p['build_type'])
+        for p in profiles
+        if p.get('branch')
+        and p.get('build_type')
+        and isinstance(p.get('schedule_utc_hours'), list)
+        and len(p.get('schedule_utc_hours')) > 0
+    }
+    if not ids:
+        logging.warning(
+            'No profiles with non-empty schedule_utc_hours in %s — not enqueueing to digest_queue',
             _DIGEST_NOTIFICATION_CONFIG,
         )
     return ids
@@ -1289,14 +1378,6 @@ def mute_worker(args):
         logging.info(f"Starting mute worker with mode: {args.mode}")
         logging.info(f"Branch: {args.branch}")
         logging.info(f"build_type: {build_type}")
-        
-        # Use provided muted_ya file or fallback to default.
-        input_muted_ya_path = getattr(args, 'muted_ya_file', muted_ya_path)
-        logging.info(f"Using muted_ya file: {input_muted_ya_path}")
-        
-        mute_check = YaMuteCheck()
-        mute_check.load(input_muted_ya_path)
-        logging.info(f"Loaded muted_ya.txt with {len(mute_check.regexps)} test patterns")
 
         mute_window_days = get_mute_window_days()
         unmute_window_days = get_unmute_window_days()
@@ -1342,6 +1423,14 @@ def mute_worker(args):
             )
 
         if args.mode == 'update_muted_ya':
+            # update_muted_ya uses mute rules to build output files,
+            # so only this mode requires loading muted_ya into YaMuteCheck.
+            input_muted_ya_path = resolve_muted_ya_path(getattr(args, 'muted_ya_file', ''), build_type)
+            logging.info(f"Using muted_ya file: {input_muted_ya_path}")
+            mute_check = YaMuteCheck()
+            mute_check.load(input_muted_ya_path)
+            logging.info(f"Loaded muted_ya.txt with {len(mute_check.regexps)} test patterns")
+
             output_path = args.output_folder
             os.makedirs(output_path, exist_ok=True)
             logging.info(f"Creating mute files in: {output_path}")
@@ -1361,12 +1450,24 @@ def mute_worker(args):
                 build_type=build_type,
             )
 
+        elif args.mode == 'sync_fast_unmute_grace':
+            to_mute, _ = create_file_set(
+                aggregated_for_mute, is_mute_candidate, use_wildcards=True, resolution='to_mute'
+            )
+            delete_fast_unmute_grace_rows(ydb_wrapper, args.branch, build_type, to_mute)
+            logging.info(
+                "fast_unmute_grace cleanup completed for branch=%s build_type=%s; candidates=%d",
+                args.branch,
+                build_type,
+                len(to_mute),
+            )
+
         elif args.mode == 'create_issues':
-            file_path = args.file_path
+            file_path = resolve_muted_ya_path(getattr(args, 'file_path', ''), build_type)
             logging.info(f"Creating issues from file: {file_path}")
 
             profile_id = make_profile_id(args.branch, build_type)
-            allowed_profiles = load_configured_digest_profile_ids()
+            allowed_profiles = load_configured_issue_profile_ids()
             if profile_id not in allowed_profiles:
                 logging.info(
                     f"Profile {profile_id!r} not in mute_issue_and_digest_config.json — skipping issue creation"
@@ -1376,6 +1477,7 @@ def mute_worker(args):
             queue_items = create_mute_issues(
                 all_data,
                 file_path,
+                aggregated_tests=aggregated_for_mute,
                 close_issues=args.close_issues,
                 branch=args.branch,
                 build_type=build_type,
@@ -1397,8 +1499,24 @@ if __name__ == "__main__":
     update_muted_ya_parser = subparsers.add_parser('update_muted_ya', help='create new muted_ya')
     update_muted_ya_parser.add_argument('--output_folder', default=repo_path, required=False, help='Output folder.')
     update_muted_ya_parser.add_argument('--branch', default='main', help='Branch to get history')
-    update_muted_ya_parser.add_argument('--muted_ya_file', default=muted_ya_path, help='Path to input muted_ya.txt file')
     update_muted_ya_parser.add_argument(
+        '--muted_ya_file',
+        default='',
+        help='Path to input muted_ya.txt file (default: resolved from build-type policy)',
+    )
+    update_muted_ya_parser.add_argument(
+        '--build-type',
+        default=DEFAULT_BUILD_TYPE,
+        dest='build_type',
+        help='tests_monitor build_type slice (default: relwithdebinfo)',
+    )
+
+    sync_fast_unmute_grace_parser = subparsers.add_parser(
+        'sync_fast_unmute_grace',
+        help='remove fast_unmute_grace rows for tests that are mute candidates again',
+    )
+    sync_fast_unmute_grace_parser.add_argument('--branch', default='main', help='Branch to get history')
+    sync_fast_unmute_grace_parser.add_argument(
         '--build-type',
         default=DEFAULT_BUILD_TYPE,
         dest='build_type',
@@ -1410,7 +1528,10 @@ if __name__ == "__main__":
         help='sync issues with muted_ya file: create missing, close orphaned, enqueue to digest',
     )
     create_issues_parser.add_argument(
-        '--file_path', default=muted_ya_path, required=False, help='Path to muted_ya.txt'
+        '--file_path',
+        default='',
+        required=False,
+        help='Path to muted_ya.txt (default: resolved from build-type policy)',
     )
     create_issues_parser.add_argument('--branch', default='main', help='Branch to get history')
     create_issues_parser.add_argument('--close_issues', action=argparse.BooleanOptionalAction, default=True, help='Close issues when all tests are unmuted (default: True)')
diff --git a/.github/scripts/tests/get_muted_tests.py b/.github/scripts/tests/mute/get_muted_tests.py
index 565cd20da02..f89fe020c19 100755
--- a/.github/scripts/tests/get_muted_tests.py
+++ b/.github/scripts/tests/mute/get_muted_tests.py
@@ -7,16 +7,23 @@ import re
 import sys
 import time
 import ydb
-from get_diff_lines_of_file import get_diff_lines_of_file
-from mute_utils import pattern_to_re
-from mute_check import YaMuteCheck
 
-# Add analytics directory to path for ydb_wrapper import
-sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'analytics'))
+SCRIPT_DIR = os.path.dirname(__file__)
+TESTS_DIR = os.path.abspath(os.path.join(SCRIPT_DIR, '..'))
+ANALYTICS_DIR = os.path.abspath(os.path.join(SCRIPT_DIR, '..', '..', 'analytics'))
+
+# Keep imports working when this script is invoked directly.
+if TESTS_DIR not in sys.path:
+    sys.path.insert(0, TESTS_DIR)
+if ANALYTICS_DIR not in sys.path:
+    sys.path.insert(0, ANALYTICS_DIR)
+
+from get_diff_lines_of_file import get_diff_lines_of_file
+from mute.mute_utils import pattern_to_re
+from mute.mute_check import YaMuteCheck
 from ydb_wrapper import YDBWrapper
 
-dir = os.path.dirname(__file__)
-repo_path = f"{dir}/../../../"
+repo_path = f"{SCRIPT_DIR}/../../../../"
 muted_ya_path = '.github/config/muted_ya.txt'
 
 
@@ -301,6 +308,14 @@ def mute_applier(args):
         print(f"Unmuted tests have been written to {unmuted_tests_file}.")
 
 
+def _add_muted_ya_file_arg(p: argparse.ArgumentParser) -> None:
+    p.add_argument(
+        '--muted_ya_file',
+        type=str,
+        help='Mute list path (default: .github/config/muted_ya.txt)',
+    )
+
+
 if __name__ == "__main__":
     print(f'🚀 Starting get_muted_tests.py script')
     print(f'📅 Current time: {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
@@ -321,16 +336,12 @@ if __name__ == "__main__":
         'upload_muted_tests', help='apply mute rules for all tests in main and upload to database'
     )
     upload_muted_tests_parser.add_argument(
-        '--branch', required=True, default='main', help='branch for getting all tests'
+        '--branch', default='main', help='branch for getting all tests'
     )
     upload_muted_tests_parser.add_argument(
         '--build_type', required=True, help='build type for filtering tests'
     )
-    upload_muted_tests_parser.add_argument(
-        '--muted_ya_file',
-        type=str,
-        help='Path to muted_ya.txt file (default: .github/config/muted_ya.txt)'
-    )
+    _add_muted_ya_file_arg(upload_muted_tests_parser)
 
     get_mute_details_parser = subparsers.add_parser(
         'get_mute_diff',
@@ -353,6 +364,7 @@ if __name__ == "__main__":
         required=True,
         help='build type for filtering tests',
     )
+    _add_muted_ya_file_arg(get_mute_details_parser)
     args = parser.parse_args()
     
     print(f'📋 Parsed arguments:')
diff --git a/.github/scripts/tests/mute_check.py b/.github/scripts/tests/mute/mute_check.py
index abd6d8938f4..50f13a7df5c 100644
--- a/.github/scripts/tests/mute_check.py
+++ b/.github/scripts/tests/mute/mute_check.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 import re
-from mute_utils import pattern_to_re
+from mute.mute_utils import pattern_to_re
 
 
 class YaMuteCheck:
diff --git a/.github/scripts/tests/mute/mute_helper.py b/.github/scripts/tests/mute/mute_helper.py
new file mode 100644
index 00000000000..dba6ac16738
--- /dev/null
+++ b/.github/scripts/tests/mute/mute_helper.py
@@ -0,0 +1,248 @@
+#!/usr/bin/env python3
+"""Minimal helpers for mute workflows: matrix and resolve-path."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from typing import Dict, List, Optional, Set, Tuple
+
+def _parse_csv(raw: str) -> List[str]:
+    return [p.strip() for p in (raw or '').replace('\n', ',').split(',') if p.strip()]
+
+
+def _normalize_unique(items: List[object]) -> List[str]:
+    out: List[str] = []
+    seen: Set[str] = set()
+    for item in items:
+        t = str(item).strip().lower()
+        if not t or t in seen:
+            continue
+        seen.add(t)
+        out.append(t)
+    return out
+
+
+def load_policy(config_path: str) -> Tuple[List[str], Dict[str, List[str]]]:
+    with open(config_path, encoding='utf-8') as f:
+        data = json.load(f)
+    if not isinstance(data, dict):
+        raise ValueError(f'{config_path}: expected JSON object')
+
+    if not isinstance(data.get('default_build_types'), list):
+        raise ValueError(f'{config_path}: "default_build_types" must be a JSON array')
+    defaults = _normalize_unique(data['default_build_types'])
+    if not defaults:
+        raise ValueError(f'{config_path}: "default_build_types" must contain at least one build type')
+
+    overrides_raw = data.get('branch_overrides') or {}
+    if not isinstance(overrides_raw, dict):
+        raise ValueError(f'{config_path}: "branch_overrides" must be a JSON object')
+
+    overrides: Dict[str, List[str]] = {}
+    for branch, values in overrides_raw.items():
+        branch_name = str(branch).strip()
+        if not branch_name:
+            continue
+        if isinstance(values, dict):
+            values = values.get('build_types')
+        if not isinstance(values, list):
+            raise ValueError(
+                f'{config_path}: override for branch "{branch_name}" must be an array or '
+                '{"build_types": [...]}'
+            )
+        normalized = _normalize_unique(values)
+        if not normalized:
+            raise ValueError(f'{config_path}: override for branch "{branch_name}" has no valid build types')
+        overrides[branch_name] = normalized
+
+    return defaults, overrides
+
+
+def load_branches_from_file(branches_file: str) -> List[str]:
+    with open(branches_file, encoding='utf-8') as f:
+        data = json.load(f)
+    if not isinstance(data, list):
+        raise ValueError(f'{branches_file}: expected JSON array of branch names')
+    return [str(b).strip() for b in data if str(b).strip()]
+
+
+def _append_github_output(key: str, value: str) -> None:
+    path = os.environ.get('GITHUB_OUTPUT')
+    if path:
+        with open(path, 'a', encoding='utf-8') as f:
+            f.write(f'{key}={value}\n')
+
+
+def _emit_matrix_to_outputs(matrix: List[Dict[str, str]]) -> None:
+    compact = json.dumps(matrix, separators=(',', ':'), ensure_ascii=False)
+    _append_github_output('matrix_include', compact)
+
+    summary = os.environ.get('GITHUB_STEP_SUMMARY')
+    if not summary:
+        return
+    lines = [
+        f'### Mute update matrix ({len(matrix)} jobs)\n',
+        '```json\n',
+        json.dumps(matrix, indent=2, ensure_ascii=False) + '\n',
+        '```\n',
+    ]
+    with open(summary, 'a', encoding='utf-8') as f:
+        f.writelines(lines)
+
+
+def cmd_matrix(args: argparse.Namespace) -> int:
+    from mute_utils import dedicated_relative
+
+    branches_raw = args.branches_override or os.environ.get('INPUT_BRANCHES', '')
+    build_types_raw = args.build_types_override or os.environ.get('INPUT_BUILD_TYPES', '')
+    event_name = args.event_name or os.environ.get('GITHUB_EVENT_NAME', '')
+
+    try:
+        defaults, overrides = load_policy(args.build_types_config)
+        branch_list = _parse_csv(branches_raw) if branches_raw.strip() else load_branches_from_file(args.branches_file)
+
+        requested: Optional[Set[str]] = None
+        raw = build_types_raw.strip().lower()
+        if event_name == 'workflow_dispatch' and raw and raw != 'all':
+            requested = set(_normalize_unique(_parse_csv(build_types_raw)))
+
+        matrix: List[Dict[str, str]] = []
+        for branch in branch_list:
+            branch_types = list(overrides.get(branch, defaults))
+            if requested is not None:
+                branch_types = [p for p in branch_types if p in requested]
+            for preset in branch_types:
+                matrix.append(
+                    {
+                        'BASE_BRANCH': branch,
+                        'BUILD_TYPE': preset,
+                        'MUTED_YA_RELATIVE': dedicated_relative(preset),
+                    }
+                )
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        print(f'::error::{exc}', file=sys.stderr)
+        return 1
+
+    if not matrix:
+        print(
+            '::error::Mute update matrix is empty (no branches/build_types from config/overrides, '
+            'or workflow_dispatch build_types outside policy).',
+            file=sys.stderr,
+        )
+        return 1
+
+    _emit_matrix_to_outputs(matrix)
+
+    return 0
+
+
+def cmd_resolve_path(args: argparse.Namespace) -> int:
+    from mute_utils import dedicated_relative
+
+    print(dedicated_relative(args.preset))
+    return 0
+
+
+def cmd_issue_build_types(args: argparse.Namespace) -> int:
+    try:
+        with open(args.profiles_config, encoding='utf-8') as f:
+            data = json.load(f)
+        if not isinstance(data, dict):
+            raise ValueError(f'{args.profiles_config}: expected JSON object')
+
+        profiles = data.get('profiles') or []
+        if not isinstance(profiles, list):
+            raise ValueError(f'{args.profiles_config}: "profiles" must be a JSON array')
+
+        out: List[str] = []
+        seen: Set[str] = set()
+        for p in profiles:
+            if not isinstance(p, dict):
+                continue
+            if p.get('branch') != args.branch:
+                continue
+            bt = p.get('build_type')
+            if not isinstance(bt, str):
+                continue
+            bt = bt.strip()
+            if not bt or bt in seen:
+                continue
+            seen.add(bt)
+            out.append(bt)
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        print(f'::error::{exc}', file=sys.stderr)
+        return 1
+
+    if not out:
+        print('::error::Issue build-type matrix is empty.', file=sys.stderr)
+        return 1
+
+    print(json.dumps(out, ensure_ascii=False))
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest='command', required=True)
+
+    pm = sub.add_parser('matrix', help='Emit matrix_include JSON for update_muted_ya setup job')
+    pm.add_argument(
+        '--branches-file',
+        required=True,
+        help='Path to stable_tests_branches.json (JSON array of branch names)',
+    )
+    pm.add_argument(
+        '--build-types-config',
+        required=True,
+        help='Path to mute build-type policy JSON (default + branch overrides)',
+    )
+    pm.add_argument(
+        '--branches-override',
+        default='',
+        help='Comma-separated branches instead of branches-file (default: env INPUT_BRANCHES)',
+    )
+    pm.add_argument(
+        '--build-types-override',
+        default='',
+        help='workflow_dispatch: env INPUT_BUILD_TYPES or "all" (default: env)',
+    )
+    pm.add_argument(
+        '--event-name',
+        default='',
+        help='Override GITHUB_EVENT_NAME (default: env)',
+    )
+    pm.set_defaults(func=cmd_matrix)
+
+    rp = sub.add_parser('resolve-path', help='Print muted_ya relative path for a build preset')
+    rp.add_argument(
+        '--preset',
+        required=True,
+        help='build_preset / BUILD_TYPE, e.g. relwithdebinfo, release-asan',
+    )
+    rp.set_defaults(func=cmd_resolve_path)
+
+    ib = sub.add_parser(
+        'issue-build-types',
+        help='Print JSON array of build types for issue workflow from profiles config',
+    )
+    ib.add_argument(
+        '--profiles-config',
+        required=True,
+        help='Path to mute_issue_and_digest_config.json',
+    )
+    ib.add_argument(
+        '--branch',
+        required=True,
+        help='Branch name to select matching profiles',
+    )
+    ib.set_defaults(func=cmd_issue_build_types)
+
+    args = parser.parse_args()
+    return args.func(args)
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/.github/scripts/tests/mute_utils.py b/.github/scripts/tests/mute/mute_utils.py
index 167d8b4fd51..fd618c70812 100644
--- a/.github/scripts/tests/mute_utils.py
+++ b/.github/scripts/tests/mute/mute_utils.py
@@ -1,13 +1,50 @@
 from __future__ import annotations
-from typing import Set, Tuple
-import operator
+import os
 import re
 import xml.etree.ElementTree as ET
 import sys
-import yaml
 import json
 
-from junit_utils import add_junit_property
+CONFIG_DIR = os.path.join('.github', 'config')
+MUTE_CONFIG = os.path.join(CONFIG_DIR, 'mute_config.json')
+
+def _normalize_relative_path(path: str) -> str:
+    return path.replace('\\', '/')
+
+
+def _repo_root_from_this_file() -> str:
+    # .../<repo>/.github/scripts/tests/mute/mute_utils.py
+    return os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+
+
+def _load_muted_ya_path_policy() -> tuple[str, dict[str, str]]:
+    cfg_path = os.path.join(_repo_root_from_this_file(), MUTE_CONFIG)
+    with open(cfg_path, encoding='utf-8') as f:
+        data = json.load(f)
+    if not isinstance(data, dict):
+        raise ValueError(f'{cfg_path}: expected JSON object')
+
+    raw_default = data.get('default_muted_ya_path')
+    if not isinstance(raw_default, str) or not raw_default.strip():
+        raise ValueError(f'{cfg_path}: "default_muted_ya_path" must be a non-empty string')
+    default_path = _normalize_relative_path(raw_default.strip())
+
+    per_preset: dict[str, str] = {}
+    raw_paths = data.get('muted_ya_paths', {})
+    if not isinstance(raw_paths, dict):
+        raise ValueError(f'{cfg_path}: "muted_ya_paths" must be a JSON object')
+    for preset, rel_path in raw_paths.items():
+        if not isinstance(preset, str) or not preset:
+            raise ValueError(f'{cfg_path}: "muted_ya_paths" keys must be non-empty strings')
+        if not isinstance(rel_path, str) or not rel_path.strip():
+            raise ValueError(f'{cfg_path}: muted_ya_paths["{preset}"] must be a non-empty string')
+        per_preset[preset] = _normalize_relative_path(rel_path.strip())
+    return default_path, per_preset
+
+
+def dedicated_relative(preset: str) -> str:
+    default_path, per_preset = _load_muted_ya_path_policy()
+    return per_preset.get(preset, default_path)
 
 
 def pattern_to_re(pattern):
@@ -44,6 +81,12 @@ class MuteTestCheck:
 
 
 def mute_target(testcase):
+    try:
+        from junit_utils import add_junit_property
+    except ModuleNotFoundError:
+        sys.path.append(os.path.dirname(os.path.dirname(__file__)))
+        from junit_utils import add_junit_property
+
     err_text = []
     err_msg = None
     found = False
@@ -84,38 +127,6 @@ def mute_target(testcase):
     return True
 
 
-def remove_failure(node):
-    while 1:
-        failure = node.find("failure")
-        if failure is None:
-            break
-        node.remove(failure)
-
-
-def op_attr(node, attr, op, value):
-    v = int(node.get(attr, 0))
-    node.set(attr, str(op(v, value)))
-
-
-def inc_attr(node, attr, value):
-    return op_attr(node, attr, operator.add, value)
-
-
-def dec_attr(node, attr, value):
-    return op_attr(node, attr, operator.sub, value)
-
-
-def update_suite_info(root, n_remove_failures=None, n_remove_errors=None, n_skipped=None):
-    if n_remove_failures:
-        dec_attr(root, "failures", n_remove_failures)
-
-    if n_remove_errors:
-        dec_attr(root, "errors", n_remove_errors)
-
-    if n_skipped:
-        inc_attr(root, "skipped", n_skipped)
-
-
 def recalc_suite_info(suite):
     tests = failures = skipped = 0
     elapsed = 0.0
@@ -141,8 +152,8 @@ def _split(s: str, sep: str) -> tuple[str, str]:
     else:
         return s[:p], s[p + 1 :]
 
-def get_previously_skipped_tests(report_json_path: str) -> Set[Tuple[str, str]]:
-    result = set()
+def get_previously_skipped_tests(report_json_path: str) -> set[tuple[str, str]]:
+    result: set[tuple[str, str]] = set()
     if report_json_path:
         with open(report_json_path, 'r') as f:
             report = json.load(f)
@@ -161,6 +172,8 @@ def get_previously_skipped_tests(report_json_path: str) -> Set[Tuple[str, str]]:
     return result
 
 def convert_muted_txt_to_yaml(muted_txt_path: str, report_json_path: str) -> None:
+    import yaml
+
     with open(muted_txt_path) as file:
         muted_tests = file.readlines()
     previously_skipped = get_previously_skipped_tests(report_json_path)
@@ -195,5 +208,11 @@ def convert_muted_txt_to_yaml(muted_txt_path: str, report_json_path: str) -> Non
 
 
 if __name__ == "__main__":
-    args = sys.argv
-    globals()[args[1]](*args[2:])
+    if len(sys.argv) == 4 and sys.argv[1] == "convert_muted_txt_to_yaml":
+        convert_muted_txt_to_yaml(sys.argv[2], sys.argv[3])
+    else:
+        print(
+            "Usage: mute_utils.py convert_muted_txt_to_yaml <muted_txt_path> <report_json_path>",
+            file=sys.stderr,
+        )
+        raise SystemExit(2)
diff --git a/.github/scripts/tests/transform_build_results.py b/.github/scripts/tests/transform_build_results.py
index d6d16c2c3a2..abc92874b56 100755
--- a/.github/scripts/tests/transform_build_results.py
+++ b/.github/scripts/tests/transform_build_results.py
@@ -14,7 +14,7 @@ import time
 import urllib.parse
 import zipfile
 from typing import Set
-from mute_check import YaMuteCheck
+from mute.mute_check import YaMuteCheck
 
 
 def log_print(*args, **kwargs):