#!/usr/bin/env python3
import argparse
import base64
import csv
import json
import mimetypes
import os
import re
import requests
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import parse_qs, quote, urlencode, urljoin, urlparse, urlunparse

from get_pingkai_ticket import API_BASE_URL, TicketAPIClient
from lib.contracts.evaluate_tikv_slow import (
    ALLOWED_REFERENCE_FILES as TIKV_SLOW_EVALUATE_REFERENCE_FILES,
    DEFAULT_ASSIGNEE as TIKV_SLOW_EVALUATE_DEFAULT_ASSIGNEE,
    DEFAULT_CLINIC_ONLY as TIKV_SLOW_EVALUATE_DEFAULT_CLINIC_ONLY,
    DEFAULT_IGNORE_GRAFANA_ONLY as TIKV_SLOW_EVALUATE_DEFAULT_IGNORE_GRAFANA_ONLY,
    DEFAULT_PAGE_SIZE as TIKV_SLOW_EVALUATE_DEFAULT_PAGE_SIZE,
    DEFAULT_SAMPLE_SIZE as TIKV_SLOW_EVALUATE_DEFAULT_SAMPLE_SIZE,
    DEFAULT_SOURCE as TIKV_SLOW_EVALUATE_DEFAULT_SOURCE,
    DEFAULT_STATUS as TIKV_SLOW_EVALUATE_DEFAULT_STATUS,
    DEFAULT_TIME_RANGE as TIKV_SLOW_EVALUATE_DEFAULT_TIME_RANGE,
    FINAL_HUMAN_CONCLUSION_STATUS_VALUES as TIKV_SLOW_EVALUATE_HUMAN_CONCLUSION_STATUS_VALUES,
    FOCUS_KEYWORDS as TIKV_SLOW_EVALUATE_KEYWORDS,
    MATCH_LEVEL_VALUES as TIKV_SLOW_EVALUATE_MATCH_LEVEL_VALUES,
    MERGE_READINESS_VALUES as TIKV_SLOW_EVALUATE_MERGE_READINESS_VALUES,
    MISMATCH_TYPES as TIKV_SLOW_EVALUATE_MISMATCH_TYPES,
    NEGATIVE_KEYWORDS as TIKV_SLOW_EVALUATE_NEGATIVE_KEYWORDS,
    PROFILE_NAME as TIKV_SLOW_EVALUATE_PROFILE,
    REPLAY_ENTRY_MODES as TIKV_SLOW_EVALUATE_ENTRY_MODES,
    SKILL_CONCLUSION_FIELDS as TIKV_SLOW_EVALUATE_SKILL_CONCLUSION_FIELDS,
)
from lib.contracts.missing_evidence import analyze_missing_evidence_reasons
from lib.models.incident import EvaluateCase
from lib.sources.jira import JiraSourceAdapter
from lib.sources.pingkai import PingkaiSourceAdapter
from lib.workflows.discovery_list import (
    build_jira_list_entries as workflow_build_jira_list_entries,
    collect_list_entries,
)
from lib.workflows.diagnosis_analyze import (
    execute_analyze_context,
    resolve_analyze_context,
)
from lib.workflows.diagnosis_result import build_diagnosis_result
from lib.workflows.evaluate_tikv_slow import (
    build_eval_case_from_jira_raw,
    build_eval_case_from_pingkai_ticket,
    discover_jira_cases,
    discover_pingkai_cases,
    evaluate_exclusion_reason as workflow_evaluate_exclusion_reason,
    is_tikv_slow_focus_text as workflow_is_tikv_slow_focus_text,
)
from lib.workflows.evaluate_contract import (
    build_replay_contract,
    build_replay_ready_case,
)
from lib.workflows.evaluate_compare import (
    build_case_review_stub,
    build_comparison_contract,
)
from lib.workflows.evaluate_review import (
    apply_maintainer_evidence_hints_to_case_review,
    apply_live_diagnosis_result_to_case_review,
    build_case_review_from_jira_case,
    build_case_review_from_pingkai_case,
)
from lib.workflows.evaluate_improvements import aggregate_case_review_improvements
from lib.workflows.reporting_report import (
    build_explicit_report_blocks,
    build_jira_report_entries as workflow_build_jira_report_entries,
    build_pingkai_report_entries as workflow_build_pingkai_report_entries,
    collect_report_entries,
)
from lib.workflows.reporting_storage_weekly import (
    build_storage_weekly_jira_entries as workflow_build_storage_weekly_jira_entries,
    build_storage_weekly_pingkai_entries as workflow_build_storage_weekly_pingkai_entries,
    collect_storage_weekly_sections,
    render_storage_weekly_report as workflow_render_storage_weekly_report,
)

BASE_DIR = Path(__file__).resolve().parent
SKILLS_ROOT = BASE_DIR.parents[2]
CLINIC_ACCESS_DISPATCH_SCRIPT = SKILLS_ROOT / 'platform' / 'op-clinic-grafana-access' / 'scripts' / 'dispatch_access.py'
CLINIC_CHECK_LOGIN_SCRIPT = SKILLS_ROOT / 'platform' / 'op-clinic-grafana-access' / 'scripts' / 'check_login.py'
CLINIC_API_BASE_URL = os.environ.get('CLINIC_API_BASE_URL', 'https://clinic.pingcap.com/clinic/api/v1').rstrip('/')
JIRA_API_BASE_URL = os.environ.get('JIRA_API_BASE_URL', 'https://tidb.atlassian.net').rstrip('/')
TEMPO_API_BASE_URL = os.environ.get('TEMPO_API_BASE_URL', 'https://api.tempo.io').rstrip('/')
DEFAULT_RCA_LINK_DIR = os.environ.get(
    'RCA_LINK_DIR',
    'https://pingcap.feishu.cn/wiki/RO1lwXzk6iNMmZk9A4McSML1nec',
).strip()
DEFAULT_CLOSE_COMMENT = (
    'This issue has been open for a long time and we have not received any further feedback. '
    'We will close this ticket for now. If there are any further questions or the issue persists, '
    'please submit a new ticket.'
)
CRITICAL_EVIDENCE_MISSING_PREFIX = 'Critical evidence missing:'
CLINIC_LINK_RE = re.compile(
    r'https://clinic\.pingcap\.com(?:\.cn)?/portal/#/orgs/\d+/clusters/\d+[^\s<>()"\']*',
    flags=re.I,
)
CLINIC_DOWNLOAD_LINK_RE = re.compile(
    r'https://clinic\.pingcap\.com(?:\.cn)?/clinic/api/v1/orgs/\d+/clusters/\d+/(?:download|downloadConf)/[^\s<>()"\']+',
    flags=re.I,
)
PINGKAI_FILE_LINK_RE = re.compile(
    r'(?:https://support\.pingkai\.cn)?/api/v1/files/download/[A-Za-z0-9]+',
    flags=re.I,
)
HTTP_LINK_RE = re.compile(r'https?://[^\s<>()"\']+', flags=re.I)
CLINIC_CLUSTER_VERSION_CACHE: dict[str, str] = {}
CLINIC_PROFILE_CACHE: dict[str, str] = {}
STORAGE_WEEKLY_COMPONENTS = (
    ('PD', 'PD'),
    ('TXN', 'TiDB Transaction'),
    ('TiKV', 'TiKV'),
)
STORAGE_WEEKLY_DEFAULT_L2_ASSIGNEE = '江红梅'
TEXT_ATTACHMENT_SUFFIXES = {
    '.log', '.txt', '.out', '.err', '.trace', '.json', '.yaml', '.yml', '.toml', '.conf', '.cfg', '.sql', '.md',
}
SKIPPED_IMAGE_ATTACHMENT_SUFFIXES = {
    '.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.tif', '.tiff', '.svg',
}
SKIPPED_IMAGE_ATTACHMENT_MIME_PREFIXES = (
    'image/',
)
PROFILE_ATTACHMENT_HINTS = (
    'profile', 'pprof', 'heap', 'cpu', 'trace', 'flamegraph', 'mutex', 'goroutine',
)
ATTACHMENT_ANALYSIS_HINTS = (
    'log', '日志', 'profile', 'pprof', 'trace', '栈', 'stack', 'panic', 'exception', 'dump', 'flamegraph', '配置', 'config',
)


def run(cmd, check=True):
    p = subprocess.run(cmd, text=True, capture_output=True)
    if check and p.returncode != 0:
        raise RuntimeError(p.stderr.strip() or p.stdout.strip() or f'command failed: {cmd}')
    return p


def normalized_base_url(raw: str) -> str:
    text = str(raw or '').strip()
    if not text:
        return ''
    parsed = urlparse(text)
    if parsed.scheme and parsed.netloc:
        return f'{parsed.scheme}://{parsed.netloc}'.rstrip('/')
    return text.rstrip('/')


def detect_clinic_profile(base_url: str) -> str:
    normalized = normalized_base_url(base_url)
    if not normalized or not CLINIC_CHECK_LOGIN_SCRIPT.exists():
        return ''
    if normalized in CLINIC_PROFILE_CACHE:
        return CLINIC_PROFILE_CACHE[normalized]

    proc = run([sys.executable, str(CLINIC_CHECK_LOGIN_SCRIPT), '--base-url', normalized], check=False)
    if proc.returncode != 0:
        return ''
    try:
        payload = json.loads((proc.stdout or '').strip())
    except Exception:
        return ''

    profile_name = str((payload or {}).get('profile_name') or '').strip()
    if profile_name:
        CLINIC_PROFILE_CACHE[normalized] = profile_name
    return profile_name


def effective_clinic_profile(args, *, candidate_base_url: str = '') -> str:
    explicit = str(getattr(args, 'clinic_profile', '') or '').strip()
    if explicit:
        return explicit
    return detect_clinic_profile(candidate_base_url)


def validated_url(raw_url: str, *, allowed_schemes: tuple[str, ...] = ('https',), allowed_hosts: tuple[str, ...], path_prefixes: tuple[str, ...] = ()) -> str:
    url = str(raw_url or '').strip()
    parsed = urlparse(url)
    if parsed.scheme not in allowed_schemes:
        raise ValueError(f'unsupported url scheme: {parsed.scheme or "empty"}')
    host = str(parsed.hostname or '').strip().lower()
    if not host or host not in {item.lower() for item in allowed_hosts}:
        raise ValueError(f'unsupported url host: {host or "empty"}')
    normalized_path = str(parsed.path or '/')
    if path_prefixes and not any(normalized_path.startswith(prefix) for prefix in path_prefixes):
        raise ValueError(f'unsupported url path: {normalized_path}')
    return url


def build_validated_api_url(base_url: str, path: str, *, query: dict | None = None, allowed_hosts: tuple[str, ...]) -> str:
    normalized_base = normalized_base_url(base_url)
    parsed_base = urlparse(normalized_base)
    base_path = parsed_base.path.rstrip('/')
    normalized_path = '/' + str(path or '').lstrip('/')
    final_url = f'{parsed_base.scheme}://{parsed_base.netloc}{base_path}{normalized_path}'
    if query:
        query_string = urlencode({k: v for k, v in query.items() if v not in (None, '')}, doseq=True)
        if query_string:
            final_url = f'{final_url}?{query_string}'
    return validated_url(
        final_url,
        allowed_hosts=allowed_hosts,
        path_prefixes=(f'{base_path}/' if base_path else '/',),
    )


def print_json(data):
    print(json.dumps(data, ensure_ascii=False, indent=2))


def tempo_api_token():
    for name in ('TEMPO_API_TOKEN', 'TEMPO_TOKEN'):
        value = str(os.environ.get(name) or '').strip()
        if value:
            return value
    return ''


def clinic_api_token():
    value = str(os.environ.get('CLINIC_API_KEY') or '').strip()
    if value:
        return value
    env_file = os.environ.get('ENV_FILE', '.env')
    try:
        with open(env_file, 'r', encoding='utf-8') as fh:
            for line in fh:
                m = re.match(r'\s*CLINIC_API_KEY\s*=\s*("?)(.+?)\1\s*$', line.strip())
                if m:
                    return m.group(2).strip()
    except OSError:
        pass
    return ''


def clinic_api_available() -> bool:
    return bool(clinic_api_token())


def clinic_api_json(path: str, query: dict | None = None):
    token = clinic_api_token()
    if not token:
        return None

    url = build_validated_api_url(
        CLINIC_API_BASE_URL,
        path,
        query=query,
        allowed_hosts=('clinic.pingcap.com', 'clinic.pingcap.com.cn'),
    )

    try:
        resp = requests.get(
            url,
            headers={
                'Authorization': f'Bearer {token}',
                'Accept': 'application/json',
                'User-Agent': 'oncall-toolkit/1.0',
            },
            timeout=20,
        )
        resp.raise_for_status()
        payload = resp.text
    except (requests.RequestException, TimeoutError):
        return None

    try:
        return json.loads(payload)
    except Exception:
        return None


def clinic_cluster_version(cluster_id: str) -> str:
    cid = str(cluster_id or '').strip()
    if not cid:
        return ''
    if cid in CLINIC_CLUSTER_VERSION_CACHE:
        return CLINIC_CLUSTER_VERSION_CACHE[cid]
    if not clinic_api_available():
        CLINIC_CLUSTER_VERSION_CACHE[cid] = ''
        return ''

    search = clinic_api_json(
        '/dashboard/clusters',
        query={
            'cluster_id': cid,
            'show_deleted': 'true',
            'limit': 10,
            'page': 1,
        },
    )
    items = (search or {}).get('items') or []
    org_id = ''
    for item in items:
        if str((item or {}).get('clusterID') or '').strip() != cid:
            continue
        version = str((item or {}).get('clusterVersion') or '').strip()
        if version:
            CLINIC_CLUSTER_VERSION_CACHE[cid] = version
            return version
        org_id = str((item or {}).get('orgID') or '').strip()
        break

    if org_id:
        detail = clinic_api_json(f'/orgs/{quote(org_id)}/clusters/{quote(cid)}')
        version = str((detail or {}).get('version') or '').strip()
        if version:
            CLINIC_CLUSTER_VERSION_CACHE[cid] = version
            return version

    CLINIC_CLUSTER_VERSION_CACHE[cid] = ''
    return ''


def tempo_api_available() -> bool:
    return bool(tempo_api_token())


def tempo_api_json(path: str, query: dict | None = None):
    token = tempo_api_token()
    if not token:
        return None

    url = build_validated_api_url(
        TEMPO_API_BASE_URL,
        path,
        query=query,
        allowed_hosts=('api.tempo.io',),
    )

    try:
        resp = requests.get(
            url,
            headers={
                'Authorization': f'Bearer {token}',
                'Accept': 'application/json',
                'User-Agent': 'oncall-toolkit/1.0',
            },
            timeout=20,
        )
        resp.raise_for_status()
        payload = resp.text
    except (requests.RequestException, TimeoutError):
        return None

    try:
        return json.loads(payload)
    except Exception:
        return None


def tempo_get_worklog_by_jira_id(jira_worklog_id: str):
    worklog_id = str(jira_worklog_id or '').strip()
    if not worklog_id:
        return None
    return tempo_api_json(f'/4/worklogs/jira/{quote(worklog_id)}')


def jira_api_email():
    for name in ('ATLASSIAN_EMAIL', 'JIRA_EMAIL'):
        value = str(os.environ.get(name) or '').strip()
        if value:
            return value
    return ''


def jira_api_token():
    for name in ('JIRA_API_TOKEN', 'ATLASSIAN_API_TOKEN', 'JIRA_TOKEN', 'ATLASSIAN_TOKEN'):
        value = str(os.environ.get(name) or '').strip()
        if value:
            return value
    return ''


def jira_api_available() -> bool:
    return bool(jira_api_email() and jira_api_token())


def jira_api_json(path: str, query: dict | None = None):
    if not jira_api_available():
        return None

    url = build_validated_api_url(
        JIRA_API_BASE_URL,
        path,
        query=query,
        allowed_hosts=('tidb.atlassian.net',),
    )

    auth = base64.b64encode(f'{jira_api_email()}:{jira_api_token()}'.encode('utf-8')).decode('ascii')
    try:
        resp = requests.get(
            url,
            headers={
                'Authorization': f'Basic {auth}',
                'Accept': 'application/json',
                'User-Agent': 'oncall-toolkit/1.0',
            },
            timeout=20,
        )
        resp.raise_for_status()
        payload = resp.text
    except (requests.RequestException, TimeoutError):
        return None

    try:
        return json.loads(payload)
    except Exception:
        return None


def jira_api_request(method: str, path: str, payload: dict | None = None, query: dict | None = None):
    if not jira_api_available():
        return None

    url = build_validated_api_url(
        JIRA_API_BASE_URL,
        path,
        query=query,
        allowed_hosts=('tidb.atlassian.net',),
    )

    auth = base64.b64encode(f'{jira_api_email()}:{jira_api_token()}'.encode('utf-8')).decode('ascii')
    headers = {
        'Authorization': f'Basic {auth}',
        'Accept': 'application/json',
        'User-Agent': 'oncall-toolkit/1.0',
    }
    if payload is not None:
        headers['Content-Type'] = 'application/json'
    try:
        resp = requests.request(
            method.upper(),
            url,
            headers=headers,
            json=payload,
            timeout=20,
        )
        resp.raise_for_status()
        body = resp.text
    except (requests.RequestException, TimeoutError):
        return None

    if not body.strip():
        return {}
    try:
        return json.loads(body)
    except Exception:
        return {}


def current_user():
    try:
        p = run(['jira', 'me'])
        return p.stdout.strip()
    except Exception:
        pass
    return 'currentUser()'


def normalize_jql_value(value: str) -> str:
    return str(value or '').replace('"', '\\"')


def strip_jql_order_by(jql: str) -> str:
    text = str(jql or '').strip()
    if not text:
        return ''
    return re.sub(r'\s+ORDER\s+BY\s+.+$', '', text, flags=re.I | re.S).strip()


def resolve_current_user_aliases():
    raw = current_user().strip()
    aliases = []
    seen = set()

    def add(v):
        s = str(v or '').strip()
        if not s:
            return
        key = s.lower()
        if key in seen:
            return
        seen.add(key)
        aliases.append(s)

    add(raw)
    if '@' in raw:
        try:
            sample = subprocess.run(
                ['jira', 'issue', 'list', '-q', f'project in (TCOC, GTOC) AND created >= -30d AND "L2 assignee" = "{normalize_jql_value(raw)}"', '--plain'],
                text=True,
                capture_output=True,
            )
            key_match = re.search(r'\b([A-Z]+-\d+)\b', sample.stdout or '')
            if key_match:
                issue_raw = get_issue_raw(key_match.group(1))
                fields = (issue_raw or {}).get('fields', {})
                l2 = fields.get('customfield_10053') or {}
                add(l2.get('displayName'))
                add(l2.get('emailAddress'))
                add(l2.get('accountId'))
        except Exception:
            pass

        local = raw.split('@', 1)[0]
        parts = [p for p in re.split(r'[._-]+', local) if p]
        if parts:
            add(' '.join(p.capitalize() for p in parts))
            if len(parts) >= 2:
                add(f'{parts[-1].capitalize()} {parts[0].capitalize()}')
    return aliases or [raw or 'currentUser()']


def expand_assignee_aliases(who: str):
    raw = str(who or '').strip()
    if not raw:
        return []
    if raw == 'currentUser()':
        return resolve_current_user_aliases()

    aliases = []
    seen = set()

    def add(value):
        s = str(value or '').strip()
        if not s:
            return
        key = s.lower()
        if key in seen:
            return
        seen.add(key)
        aliases.append(s)

    has_non_ascii = any(ord(ch) > 127 for ch in raw)
    if not has_non_ascii:
        add(raw)

    # For Jira filtering, do not emit Chinese display names into JQL.
    # If the user passed a Chinese name like "江红梅", treat it only as a
    # hint to expand into the current Jira login's stable aliases.
    if has_non_ascii:
        for alias in resolve_current_user_aliases():
            if any(ord(ch) > 127 for ch in alias):
                continue
            add(alias)

    return aliases


def build_time_jql(time_range: str) -> str:
    mapping = {
        '1w': 'created >= -7d',
        '7d': 'created >= -7d',
        '2w': 'created >= -14d',
        '30d': 'created >= -30d',
        'this_week': 'created >= startOfWeek()',
    }
    return mapping.get(time_range, 'created >= -7d')


def parse_time_window_value(value: str | None):
    text = str(value or '').strip()
    if not text:
        return None
    dt = parse_datetime_loose(text)
    if not dt:
        raise ValueError(
            f'invalid time value: {value}. Use formats like 2026-04-02, 2026-04-02 18:00, or 2026-04-02T18:00:00+08:00.'
        )
    return dt.astimezone(timezone.utc)


def resolve_time_window(start_time: str | None, end_time: str | None, default_days: int = 7):
    start_dt = parse_time_window_value(start_time)
    end_dt = parse_time_window_value(end_time)

    if not start_dt and not end_dt:
        end_dt = datetime.now(timezone.utc)
        start_dt = end_dt - timedelta(days=default_days)
    elif start_dt and not end_dt:
        end_dt = start_dt + timedelta(days=default_days)
    elif end_dt and not start_dt:
        start_dt = end_dt - timedelta(days=default_days)

    if start_dt >= end_dt:
        raise ValueError('--start-time must be earlier than --end-time')
    return start_dt, end_dt


def format_jira_jql_datetime(value: datetime) -> str:
    return value.astimezone().strftime('%Y-%m-%d %H:%M')


def build_absolute_time_jql(start_time: datetime, end_time: datetime) -> str:
    start_text = format_jira_jql_datetime(start_time)
    end_text = format_jira_jql_datetime(end_time)
    return f'created >= "{start_text}" AND created < "{end_text}"'


def build_status_jql(status: str) -> str:
    mapping = {
        'all': '',
        'unresolved': 'statusCategory != Done',
        'done': 'statusCategory = Done',
        'resolved': 'statusCategory = Done',
        'open': 'statusCategory != Done',
    }
    return mapping.get(status, '')


def build_components_jql(components: str) -> str:
    if not components or components == 'all':
        return 'component in (TiKV, TiDB, PD)'
    parts = [p.strip() for p in components.split(',') if p.strip()]
    if not parts:
        return 'component in (TiKV, TiDB, PD)'
    quoted = ', '.join(f'"{p}"' for p in parts)
    return f'component in ({quoted})'


def parse_assignee_filters(assignee: str):
    parts = [part.strip() for part in str(assignee or '').split(',') if part.strip()]
    return parts or ['currentUser()']


def build_owner_jql(assignee: str) -> str:
    filters = parse_assignee_filters(assignee)
    if any(part.lower() == 'all' for part in filters):
        return ''

    values = []
    for who in filters:
        values.extend(expand_assignee_aliases(who))

    parts = []
    for v in values:
        if v == 'currentUser()':
            parts.append('(assignee = currentUser())')
            parts.append('("L2 assignee" = currentUser())')
            continue
        q = normalize_jql_value(v)
        parts.append(f'(assignee = "{q}")')
        parts.append(f'("L2 assignee" = "{q}")')
    return ' OR '.join(parts)


def jira_search_issues(jql: str, fields: list[str] | None = None, max_results: int = 100):
    if not jira_api_available():
        return None

    issues = []
    start_at = 0
    page_size = max(1, min(int(max_results or 100), 100))

    while True:
        data = jira_api_json(
            '/rest/api/3/search',
            query={
                'jql': jql,
                'fields': fields or ['summary'],
                'startAt': start_at,
                'maxResults': page_size,
            },
        )
        if not isinstance(data, dict):
            return None

        batch = data.get('issues') or []
        if not isinstance(batch, list):
            return None
        issues.extend(batch)

        total = int(data.get('total') or 0)
        start_at += len(batch)
        if not batch or start_at >= total:
            break

    return issues


def search_jira_issue_keys(jql: str, mode: str):
    cli_jql = strip_jql_order_by(jql)
    try:
        p = run(['jira', 'issue', 'list', '-p', 'GTOC', '-q', cli_jql, '--raw'], check=False)
    except FileNotFoundError:
        p = None
    if p and p.returncode == 0 and p.stdout.strip():
        try:
            data = json.loads(p.stdout)
            if isinstance(data, list):
                return {
                    'keys': [
                        str((issue or {}).get('key') or '').strip()
                        for issue in data
                        if str((issue or {}).get('key') or '').strip()
                    ]
                }
        except Exception:
            pass

    issues = jira_search_issues(jql, fields=['summary'], max_results=100)
    if issues is not None:
        return {
            'keys': [
                str((issue or {}).get('key') or '').strip()
                for issue in issues
                if str((issue or {}).get('key') or '').strip()
            ]
        }

    return {
        'error_json': {
            'mode': mode,
            'jql': jql,
            'note': 'jira cli and Jira API are both unavailable; returning computed JQL only',
        }
    }


def build_list_jql(args) -> str:
    clauses = [
        'project in (TCOC, GTOC)',
        build_components_jql(args.components),
        build_time_jql(args.time_range),
        build_owner_jql(args.assignee),
    ]
    status_clause = build_status_jql(args.status)
    if status_clause:
        clauses.append(status_clause)
    return ' AND '.join(f'({c})' for c in clauses if c)


def parse_issue_keys(list_output: str):
    keys = []
    for line in list_output.splitlines():
        m = re.search(r'\b([A-Z]+-\d+)\b', line)
        if m:
            keys.append(m.group(1))
    return keys


def get_issue_raw(key: str):
    try:
        p = run(['jira', 'issue', 'view', key, '--raw'], check=False)
    except FileNotFoundError:
        p = None
    if p and p.returncode == 0 and p.stdout.strip():
        try:
            return json.loads(p.stdout)
        except Exception:
            pass
    raw = jira_api_json(f'/rest/api/3/issue/{quote(key)}')
    if isinstance(raw, dict) and raw.get('fields'):
        return raw
    return None


def get_issue_view_plain(key: str):
    return run(['jira', 'issue', 'view', key, '--plain'], check=False)


def parse_issue_view_plain(text: str, key: str):
    lines = text.splitlines()
    title = 'N/A'
    created = 'N/A'
    reporter = 'N/A'
    assignee = 'N/A'

    for i, line in enumerate(lines):
        s = line.strip()
        if s.startswith('# '):
            title = s[2:].strip()
            break

    if len(lines) > 1:
        top = lines[1]
        m = re.search(r'⌛\s+([^👷]+?)\s+👷\s+([^🔑]+?)\s+🔑️\s+' + re.escape(key), top)
        if m:
            created = m.group(1).strip()
            assignee = m.group(2).strip()

    meta_line = ''
    for line in lines:
        if '🔎 ' in line and '🚀' in line:
            meta_line = line
            break
    if meta_line:
        m = re.search(r'🔎\s+([^🚀]+?)\s+🚀', meta_line)
        if m:
            reporter = m.group(1).strip()

    return {
        'key': key,
        'title': title,
        'created': created,
        'reporter': reporter or 'N/A',
        'assignee': assignee or 'N/A',
    }


def enrich_from_raw(item: dict, raw: dict):
    fields = (raw or {}).get('fields', {})
    labels = fields.get('labels') or []
    assignee = (fields.get('assignee') or {}).get('displayName') or item.get('assignee') or 'N/A'
    reporter = (fields.get('reporter') or {}).get('displayName') or item.get('reporter') or 'N/A'
    created = fields.get('created') or item.get('created') or 'N/A'
    if created != 'N/A' and 'T' in created:
        created = created.split('T', 1)[0]
    l2 = 'N/A'
    for k, v in fields.items():
        if isinstance(v, dict) and v.get('displayName'):
            if k == 'customfield_10053' or 'l2' in k.lower():
                l2 = v.get('displayName') or 'N/A'
                break
    item.update({
        'reporter': reporter,
        'assignee': assignee,
        'created': created,
        'l2_assignee': l2,
        'labels': ','.join(labels) if labels else 'N/A',
    })
    return item


def parse_sources(value: str | None, default: str = 'jira') -> list[str]:
    raw = str(value or default).strip()
    if not raw:
        raw = default
    parts = []
    seen = set()
    for part in raw.split(','):
        token = part.strip().lower()
        if not token:
            continue
        if token not in {'jira', 'pingkai'}:
            raise ValueError(f'unsupported source: {part.strip()}')
        if token in seen:
            continue
        seen.add(token)
        parts.append(token)
    return parts or [default]


def parse_sort_datetime(value):
    s = str(value or '').strip()
    if not s or s == 'N/A':
        return None
    if s.endswith('Z'):
        s = s[:-1] + '+00:00'
    try:
        dt = datetime.fromisoformat(s)
    except ValueError:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def make_jira_source_adapter() -> JiraSourceAdapter:
    return JiraSourceAdapter(
        build_components_jql=build_components_jql,
        build_time_jql=build_time_jql,
        build_absolute_time_jql=build_absolute_time_jql,
        build_owner_jql=build_owner_jql,
        build_status_jql=build_status_jql,
        search_issue_keys=search_jira_issue_keys,
        get_issue_raw=get_issue_raw,
        get_issue_view_plain=get_issue_view_plain,
    )


def make_pingkai_source_adapter() -> PingkaiSourceAdapter:
    return PingkaiSourceAdapter(
        list_entries=list_pingkai_entries,
        list_entries_in_window=list_pingkai_entries_in_window,
        load_ticket=load_pingkai_ticket,
        ticket_ref=pingkai_ticket_ref,
    )


def list_cmd(args):
    try:
        sources = parse_sources(getattr(args, 'source', 'jira'))
    except ValueError as e:
        print(str(e))
        return 1

    result = collect_list_entries(
        args,
        sources,
        jira_source=make_jira_source_adapter(),
        build_jira_entries=lambda current_args, jira_source: workflow_build_jira_list_entries(
            current_args,
            jira_source,
            parse_issue_view_plain=parse_issue_view_plain,
            enrich_from_raw=enrich_from_raw,
            parse_sort_datetime=parse_sort_datetime,
            issue_url=issue_url,
        ),
        list_pingkai_entries=list_pingkai_entries,
    )
    if result.get('error_json'):
        print(json.dumps(result['error_json'], ensure_ascii=False, indent=2))
        return 0
    if result.get('returncode'):
        sys.stdout.write(result.get('stdout') or '')
        if result.get('stderr'):
            sys.stderr.write(result['stderr'])
        return result['returncode']

    all_entries = result.get('entries') or []

    if not all_entries:
        if len(sources) == 1 and sources[0] == 'pingkai':
            print('No Pingkai tickets found for the requested filters.')
        elif len(sources) == 1 and sources[0] == 'jira':
            print('No Jira issues found for the requested filters.')
        else:
            print('No Jira or Pingkai tickets found for the requested filters.')
        return 0

    for item in all_entries:
        print(item['line'])
    return 0


def parse_pingkai_time_range(time_range: str):
    raw = (time_range or '').strip().lower()
    if not raw:
        raw = '1w'
    if raw == 'all':
        return None
    if raw == 'this_week':
        local_now = datetime.now().astimezone()
        start_local = (local_now - timedelta(days=local_now.weekday())).replace(hour=0, minute=0, second=0, microsecond=0)
        return start_local.astimezone(timezone.utc)

    match = re.fullmatch(r'(\d+)([hdw])', raw)
    if not match:
        raise ValueError(f'unsupported Pingkai time range: {time_range}')

    amount = int(match.group(1))
    unit = match.group(2)
    if amount <= 0:
        raise ValueError(f'unsupported Pingkai time range: {time_range}')

    if unit == 'h':
        delta = timedelta(hours=amount)
    elif unit == 'd':
        delta = timedelta(days=amount)
    else:
        delta = timedelta(weeks=amount)
    return datetime.now(timezone.utc) - delta


def parse_pingkai_datetime(value: str):
    s = str(value or '').strip()
    if not s:
        return None
    if s.endswith('Z'):
        s = s[:-1] + '+00:00'
    try:
        dt = datetime.fromisoformat(s)
    except ValueError:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def format_pingkai_datetime(value: str) -> str:
    dt = parse_pingkai_datetime(value)
    if not dt:
        return 'N/A'
    return dt.astimezone().strftime('%Y-%m-%d %H:%M:%S %z')


def parse_filter_tokens(value: str):
    return [part.strip().lower() for part in str(value or '').split(',') if part.strip()]


def match_filter_value(actual: str, expected: str) -> bool:
    tokens = parse_filter_tokens(expected)
    if not tokens:
        return True
    actual_text = str(actual or '').strip().lower()
    return bool(actual_text) and any(token in actual_text for token in tokens)


def match_any_filter_value(values, expected: str) -> bool:
    tokens = parse_filter_tokens(expected)
    if not tokens:
        return True
    normalized = [str(value or '').strip().lower() for value in values if str(value or '').strip()]
    return any(token in actual for token in tokens for actual in normalized)


def pingkai_ticket_ref(ticket: dict) -> str:
    alias = ticket.get('alias')
    if alias not in (None, ''):
        return f'TICKET-{alias}'
    ticket_id = str(ticket.get('id') or '').strip()
    return ticket_id or 'TICKET-N/A'


def normalize_pingkai_module_filter(args) -> str:
    module = str(getattr(args, 'module', '') or '').strip()
    if module:
        return module
    components = (args.components or '').strip()
    if components and components != 'tikv,tidb,pd':
        return components
    return ''


def pingkai_ticket_matches(ticket: dict, args, client: TicketAPIClient) -> bool:
    module_filter = normalize_pingkai_module_filter(args)
    if module_filter and not match_filter_value(ticket.get('module_name', ''), module_filter):
        return False

    assignee_filters = parse_assignee_filters(args.assignee)
    has_assignee_filter = bool(str(args.assignee or '').strip())
    if has_assignee_filter:
        if any(part.lower() == 'all' for part in assignee_filters):
            return True
        if assignee_filters == ['currentUser()']:
            aliases = current_pingkai_user_aliases(client)
            if not match_any_filter_value(
                [ticket.get('l2_assignee_name', ''), ticket.get('l3_assignee_name', '')],
                ','.join(aliases),
            ):
                return False
        elif not match_any_filter_value(
            [ticket.get('l2_assignee_name', ''), ticket.get('l3_assignee_name', '')],
            ','.join(assignee_filters),
        ):
            return False

    return True


def render_pingkai_list_line(ticket: dict) -> str:
    ticket_ref = pingkai_ticket_ref(ticket)
    title = ticket.get('title') or 'N/A'
    created = format_pingkai_datetime(ticket.get('created_at'))
    creator = ticket.get('creator_name') or ((ticket.get('creator') or {}).get('full_name')) or 'N/A'
    assignee = ticket.get('assignee_name') or 'N/A'
    l2_assignee = ticket.get('l2_assignee_name') or 'N/A'
    l3_assignee = ticket.get('l3_assignee_name') or 'N/A'
    module_name = ticket.get('module_name') or 'N/A'
    return (
        f'- {ticket_ref} | {title} | {created} | creator={creator} | '
        f'assignee={assignee} | l2_assignee={l2_assignee} | l3_assignee={l3_assignee} | module={module_name}'
    )


def list_pingkai_entries(args):
    if args.status and args.status != 'all':
        print('Pingkai list currently supports time/module/assignee filters only; --status is not implemented.')
        return {'returncode': 1}

    try:
        created_since = parse_pingkai_time_range(args.time_range)
    except ValueError as e:
        print(str(e))
        return {'returncode': 1}
    return list_pingkai_entries_in_window(args, created_since=created_since)


def list_pingkai_entries_in_window(args, created_since: datetime | None = None, created_before: datetime | None = None):
    try:
        client = TicketAPIClient(verbose=False)
        if not client.login():
            return {'returncode': 1}
    except RuntimeError as e:
        print(str(e))
        return {'returncode': 1}

    page = 1
    matched = []
    page_size = max(1, min(int(args.page_size or 100), 100))

    while True:
        result = client.list_tickets(
            page=page,
            page_size=page_size,
            order_by='created_at',
            order_direction='desc',
        )
        if not result:
            return {'returncode': 1}

        tickets = result.get('tickets') or []
        if not tickets:
            break

        reached_cutoff = False
        for ticket in tickets:
            created_at = parse_pingkai_datetime(ticket.get('created_at'))
            if (created_since or created_before) and not created_at:
                continue
            if created_before and created_at and created_at >= created_before:
                continue
            if created_since and created_at and created_at < created_since:
                reached_cutoff = True
                break
            if pingkai_ticket_matches(ticket, args, client):
                matched.append(ticket)

        if reached_cutoff:
            break

        pagination = result.get('pagination') or {}
        total_pages = int(pagination.get('total_pages') or 0)
        if total_pages and page >= total_pages:
            break
        if len(tickets) < page_size:
            break
        page += 1

    matched.sort(
        key=lambda item: parse_pingkai_datetime(item.get('created_at')) or datetime.min.replace(tzinfo=timezone.utc),
        reverse=True,
    )

    return {
        'client': client,
        'entries': [
            {
                'source': 'pingkai',
                'created_at': parse_pingkai_datetime(ticket.get('created_at')),
                'line': render_pingkai_list_line(ticket),
                'ticket': ticket,
            }
            for ticket in matched
        ],
    }


def list_pingkai_cmd(args):
    result = list_pingkai_entries(args)
    if result.get('returncode'):
        return result['returncode']
    entries = result.get('entries') or []
    if not entries:
        print('No Pingkai tickets found for the requested filters.')
        return 0
    for item in entries:
        print(item['line'])
    return 0


def issue_url(key: str) -> str:
    return f'https://tidb.atlassian.net/browse/{key}'


def split_issue_inputs(values):
    issue_ids = []
    for value in values or []:
        for part in re.split(r'[\s,]+', str(value or '').strip()):
            if part:
                issue_ids.append(part)
    return issue_ids


def extract_text(value):
    if value is None:
        return ''
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        return '\n'.join(filter(None, (extract_text(v) for v in value)))
    if isinstance(value, dict):
        t = value.get('type')
        if t == 'text':
            return value.get('text', '')
        parts = []
        for key in ('content', 'paragraphs', 'items'):
            if key in value:
                parts.append(extract_text(value[key]))
        return '\n'.join(filter(None, parts))
    return str(value)


def extract_urls_from_rich_value(value) -> list[str]:
    urls: list[str] = []
    seen: set[str] = set()

    def add(candidate) -> None:
        text = normalize_link_candidate(candidate)
        key = text.lower()
        if not text or not text.startswith(('http://', 'https://')) or key in seen:
            return
        seen.add(key)
        urls.append(text)

    def walk(node) -> None:
        if node is None:
            return
        if isinstance(node, str):
            for match in HTTP_LINK_RE.finditer(node):
                add(match.group(0))
            return
        if isinstance(node, list):
            for item in node:
                walk(item)
            return
        if not isinstance(node, dict):
            return

        attrs = node.get('attrs')
        if isinstance(attrs, dict):
            add(attrs.get('url'))
            add(attrs.get('href'))

        for mark in node.get('marks') or []:
            if isinstance(mark, dict):
                mark_attrs = mark.get('attrs')
                if isinstance(mark_attrs, dict):
                    add(mark_attrs.get('href'))
                    add(mark_attrs.get('url'))

        for key in ('content', 'paragraphs', 'items'):
            if key in node:
                walk(node.get(key))

    walk(value)
    return urls


def summarize_text(s: str, limit: int = 220) -> str:
    s = re.sub(r'\s+', ' ', (s or '').strip())
    if not s:
        return ''
    if len(s) <= limit:
        return s
    return s[: limit - 3].rstrip() + '...'


def clean_line(s: str) -> str:
    s = re.sub(r'\s+', ' ', (s or '').strip())
    s = re.sub(r'^(dear experts|hi team|hello team|hello|dear team)[:,\s-]*', '', s, flags=re.I)
    s = re.sub(r'^(could you please|please)\s+', '', s, flags=re.I)
    return s.strip(' -:')


def is_noise_line(s: str) -> bool:
    raw = (s or '').strip()
    if not raw:
        return True
    lowered = raw.lower()
    if raw.startswith('```'):
        return True
    if re.match(r'^#+\s*time:', lowered):
        return True
    if re.match(r'^\d{4}-\d{2}-\d{2}[ t]\d{2}:\d{2}:\d{2}', raw):
        return True
    if re.match(r'^(select|update|insert|delete|explain|trace)\b', lowered):
        return True
    if any(token in lowered for token in ['stack trace', 'stacktrace', 'txn_start_ts', 'conn_', 'backtrace', 'panic:', 'exception:', 'trace_id', 'scan_detail']):
        return True
    if re.search(r'\b(user@host|cop_proc_avg|max_process_time|parse_time|compile_time|process_time|request_count)\b', lowered):
        return True
    if re.search(r'\b[0-9a-f]{16,}\b', lowered):
        return True
    if len(raw) > 180 and raw.count(' ') < 8:
        return True
    return False


def normalize_sentence(s: str) -> str:
    s = clean_line(s)
    s = re.sub(r'https?://\S+', ' ', s)
    s = re.sub(r'`[^`]*`', ' ', s)
    s = re.sub(r'\s+', ' ', s)
    s = re.sub(r'\b(question \d+|q\d+)\b[:：-]*', '', s, flags=re.I)
    s = re.sub(r'\b(answer|a)\b[:：-]*', '', s, flags=re.I)
    return s.strip(' -:')


def split_sentences(text: str):
    raw = (text or '').replace('\r', '\n')
    parts = []
    for block in raw.split('\n'):
        block = block.strip()
        if not block:
            continue
        segs = re.split(r'(?<=[.!?。；;])\s+|\s*\n\s*', block)
        for seg in segs:
            seg = normalize_sentence(seg)
            if not seg:
                continue
            if is_noise_line(seg):
                continue
            if len(seg) < 12:
                continue
            parts.append(seg)
    return parts


def dedupe_keep_order(items):
    out = []
    seen = set()
    for item in items:
        key = re.sub(r'\W+', '', item or '').lower()
        if not key or key in seen:
            continue
        seen.add(key)
        out.append(item)
    return out


def summarize_description_for_phenomenon(text: str) -> str:
    sentences = dedupe_keep_order(split_sentences(text))
    if not sentences:
        return 'TBD'

    preferred = []
    fallback = []
    for s in sentences:
        ll = s.lower()
        if any(x in ll for x in ['customer', 'user', 'application', 'request', 'error', 'timeout', 'upgrade', 'restart', 'observed', 'detected', 'failed']):
            preferred.append(s)
        else:
            fallback.append(s)

    chosen = preferred[:2] or fallback[:2] or sentences[:2]
    joined = ' '.join(chosen)
    return summarize_text(joined, 260) or 'TBD'


def summarize_owner_comments(text: str):
    raw = text or ''
    lowered = raw.lower()

    root = ''
    workaround = ''
    todo = ''

    if 're-election of new leader reaches ~9s' in lowered or 'split vote' in lowered or 'network abnormalities' in lowered:
        root = (
            'Temporary leaderless regions during TiKV restart caused excessive backoff and request timeout; '
            'the leader re-election took about 9 seconds after split votes, and the owner concluded the trigger was likely network abnormalities such as packet loss.'
        )

    has_reduce_lease = 'reduce the max-leader-lease' in lowered
    has_timeout_15 = (
        'timeout value with “15s” if possible' in raw
        or 'timeout value with "15s" if possible' in raw
        or 'timeout value with “15s”' in raw
        or 'timeout value with "15s"' in raw
    )
    has_timeout_12 = (
        'timeout value should be >= “12s”' in raw
        or 'timeout value should be >= "12s"' in raw
        or '>= “12s”' in raw
        or '>= "12s"' in raw
    )
    if has_timeout_12 or has_timeout_15 or has_reduce_lease:
        workaround = 'Increase the application timeout to at least 12s, preferably 15s.'
        if has_reduce_lease:
            workaround += ' Reducing max-leader-lease may help but is not generally recommended.'

    if 'we need some time to further investigate' in lowered:
        todo = 'Continue investigating why tikv-17 lost leadership and why the new leader election took around 10 seconds.'

    sentences = dedupe_keep_order(split_sentences(raw))
    merged_sentences = []
    i = 0
    while i < len(sentences):
        sentence = sentences[i]
        ll = sentence.lower()
        if (
            i + 1 < len(sentences)
            and ll.startswith(('please try', 'try ', 'run ', 'truncate '))
            and sentences[i + 1].lower().startswith(('and ', 'then '))
        ):
            sentence = f'{sentence} {sentences[i + 1]}'
            i += 1
        merged_sentences.append(sentence)
        i += 1
    sentences = dedupe_keep_order(merged_sentences)
    root_labels = ['Root cause', 'Cause', 'Reason', 'Conclusion', '结论', '根因', '原因']
    workaround_labels = ['Workaround', 'Mitigation', 'Suggestion', 'Solution', 'Action', '解决方案', '建议', '处理建议']
    todo_labels = ['TODO', 'Follow-up', 'Next step', 'Action item', '下一步', '后续', '待办']
    root_keywords = (
        'root cause',
        'caused by',
        'must be caused by',
        'likely cause',
        'most likely cause',
        'the cause is',
        'reason is',
        'suspect',
        'likely',
        'due to',
    )
    workaround_keywords = (
        'workaround',
        'mitigation',
        'please try',
        'try ',
        'run ',
        'truncate ',
        'evict ',
        'disable ',
        'enable ',
        'upgrade ',
        'downgrade ',
        'increase ',
        'reduce ',
    )
    todo_keywords = (
        'follow up',
        'follow-up',
        'next step',
        'need to',
        'needs to',
        'continue investigating',
        'please provide',
        'please share',
        'collect',
        'confirm',
        'check again',
        'observe',
    )

    for sentence in reversed(sentences):
        ll = sentence.lower()

        if not root:
            for label in root_labels:
                value = extract_field_value(sentence, label)
                if value:
                    root = compact_summary(value, 220)
                    break
            if (
                not root
                and any(token in ll for token in root_keywords)
                and 'please try' not in ll
            ):
                root = compact_summary(sentence, 220)

        if not workaround:
            for label in workaround_labels:
                value = extract_field_value(sentence, label)
                if value:
                    workaround = compact_summary(value, 220)
                    break
            if not workaround and any(token in ll for token in workaround_keywords):
                workaround = compact_summary(sentence, 220)

        if not todo:
            for label in todo_labels:
                value = extract_field_value(sentence, label)
                if value:
                    todo = compact_summary(value, 220)
                    break
            if not todo and any(token in ll for token in todo_keywords):
                todo = compact_summary(sentence, 220)

        if root and workaround and todo:
            break

    return (root or 'TBD', workaround or 'None yet', todo or 'TBD')


def extract_comment_entries(raw: dict):
    comments = (((raw or {}).get('fields') or {}).get('comment') or {}).get('comments') or []
    entries = []
    for c in comments:
        body = extract_text(c.get('body'))
        author = ((c.get('author') or {}).get('displayName') or '').strip()
        if body.strip():
            entries.append({
                'author': author,
                'body': body.strip(),
            })
    return entries


def comments_text(raw: dict) -> str:
    return '\n'.join(e['body'] for e in extract_comment_entries(raw))


def author_set(fields: dict):
    names = set()
    for who in [fields.get('assignee'), fields.get('customfield_10053')]:
        if isinstance(who, dict):
            for k in ('displayName', 'emailAddress', 'accountId'):
                v = (who.get(k) or '').strip()
                if v:
                    names.add(v.lower())
    return names


def is_owner_comment(author: str, owners: set) -> bool:
    a = (author or '').strip().lower()
    return bool(a and a in owners)


def split_candidate_lines(text: str):
    out = []
    for line in (text or '').splitlines():
        s = clean_line(line)
        if not s:
            continue
        if len(s) < 8:
            continue
        out.append(s)
    return out


def extract_field_value(line: str, label: str) -> str:
    m = re.search(rf'\b{re.escape(label)}\b\s*[:：-]\s*(.+)$', line, flags=re.I)
    if not m:
        return ''
    v = clean_line(m.group(1))
    if not v or v.lower() == label.lower():
        return ''
    return v


def first_sentence(text: str) -> str:
    s = clean_line(text)
    if not s:
        return ''
    m = re.split(r'(?<=[.!?。；;])\s+', s, maxsplit=1)
    return m[0].strip()


def compact_summary(text: str, limit: int = 220) -> str:
    s = clean_line(text)
    if not s:
        return ''
    s = re.sub(r'\b(root cause|workaround|todo|phenomenon)\b\s*[:：-]\s*', '', s, flags=re.I)
    return summarize_text(s, limit)


def collect_owner_comment_text(raw: dict, fields: dict) -> str:
    owners = author_set(fields)
    chunks = []
    for entry in extract_comment_entries(raw):
        if is_owner_comment(entry.get('author', ''), owners):
            body = clean_line(entry.get('body', ''))
            if body:
                chunks.append(body)
    return '\n'.join(chunks)


def derive_version(fields: dict) -> str:
    cluster_id = str(fields.get('customfield_10149') or '').strip()
    version = clinic_cluster_version(cluster_id)
    if version:
        return version

    jira_cluster_version = str(fields.get('customfield_10147') or '').strip()
    if jira_cluster_version and jira_cluster_version.upper() != 'N/A':
        return jira_cluster_version

    versions = fields.get('versions') or []
    if versions:
        names = [v.get('name') for v in versions if isinstance(v, dict) and v.get('name')]
        if names:
            return ', '.join(names)
    fix_versions = fields.get('fixVersions') or []
    if fix_versions:
        names = [v.get('name') for v in fix_versions if isinstance(v, dict) and v.get('name')]
        if names:
            return ', '.join(names)
    return 'N/A'


def derive_priority(fields: dict) -> str:
    priority = fields.get('priority') or {}
    return priority.get('name') or 'JIRA'


def report_placeholder_blank(value: str) -> str:
    text = str(value or '').strip()
    if text in {'TBD', 'TBD.'}:
        return ''
    return text


def render_summary_block_from_raw(key: str, raw: dict) -> str:
    fields = (raw or {}).get('fields', {})
    title = fields.get('summary') or key
    priority = derive_priority(fields)
    labels = fields.get('labels') or []
    labels_text = ', '.join(labels) if labels else 'N/A'
    description_text = extract_text(fields.get('description'))
    owner_comment_text = collect_owner_comment_text(raw, fields)

    phenomenon = summarize_description_for_phenomenon(description_text)
    root_cause, workaround, todo = summarize_owner_comments(owner_comment_text)

    assignee = ((fields.get('assignee') or {}).get('displayName') or '').strip()
    l2_assignee = ((fields.get('customfield_10053') or {}).get('displayName') or '').strip()
    mentions = ' '.join([m for m in [f'@{assignee}' if assignee else '', f'@{l2_assignee}' if l2_assignee else ''] if m])

    heading = f'### {priority} {title} [{key}]({issue_url(key)})'
    if mentions:
        heading = f'{heading} {mentions}'

    lines = [
        heading,
        f'- Labels: {labels_text}',
        f'- Version: {derive_version(fields)}',
        f'- Phenomenon: {report_placeholder_blank(phenomenon)}',
        f'- Root cause: {report_placeholder_blank(root_cause)}',
        f'- Workaround: {workaround}',
        f'- TODO: {report_placeholder_blank(todo)}',
    ]
    return '\n'.join(lines)


def pingkai_owner_aliases(ticket: dict):
    aliases = set()

    def add(value):
        s = str(value or '').strip()
        if s:
            aliases.add(s.lower())

    add(ticket.get('assignee_id'))
    add(ticket.get('assignee_name'))
    add(ticket.get('l2_assignee_id'))
    add(ticket.get('l2_assignee_name'))
    return aliases


def pingkai_comment_entries(ticket: dict):
    comments = (ticket or {}).get('comments') or []
    entries = []
    for comment in comments:
        created = comment.get('created_at') or ''
        author = (comment.get('author') or {}) if isinstance(comment.get('author'), dict) else {}
        entries.append({
            'created': created,
            'author_name': (author.get('full_name') or comment.get('author_name') or '').strip(),
            'author_username': (author.get('username') or '').strip(),
            'author_email': (author.get('email') or '').strip(),
            'author_id': (author.get('id') or comment.get('author_id') or '').strip(),
            'content': comment.get('content', '') or '',
            'body': strip_text_for_pingkai_comment(comment.get('content', '')),
            'is_internal': bool(comment.get('is_internal')),
            'is_system_generated': bool(comment.get('is_system_generated')),
        })
    entries.sort(key=lambda item: item.get('created') or '')
    return entries


def pingkai_owner_comment_entries(ticket: dict):
    aliases = pingkai_owner_aliases(ticket)
    entries = []
    for item in pingkai_comment_entries(ticket):
        if item.get('is_system_generated'):
            continue
        author_candidates = {
            x.lower()
            for x in [
                item.get('author_name', ''),
                item.get('author_username', ''),
                item.get('author_email', ''),
                item.get('author_id', ''),
            ]
            if str(x or '').strip()
        }
        if aliases.intersection(author_candidates):
            entries.append(item)
    return entries


def pingkai_priority_text(ticket: dict) -> str:
    urgency = str(ticket.get('urgency_name') or '').strip()
    if urgency:
        return urgency
    priority = ticket.get('priority')
    if priority not in (None, ''):
        return f'P{priority}'
    return 'PINGKAI'


def pingkai_version_text(ticket: dict) -> str:
    version = str(ticket.get('version_name') or '').strip()
    if version:
        return version

    search_space = '\n'.join(
        [strip_text_for_pingkai_comment(ticket.get('description', ''))]
        + [item.get('body', '') for item in pingkai_comment_entries(ticket)]
    )
    match = re.search(r'\bv\d+(?:\.\d+){1,3}(?:[-._][0-9A-Za-z.]+)?\b', search_space, flags=re.I)
    if match:
        return match.group(0)
    return 'N/A'


def pingkai_labels_text(ticket: dict) -> str:
    tags = ticket.get('tags') or []
    labels = []
    for tag in tags:
        if isinstance(tag, dict):
            value = tag.get('name') or tag.get('label') or tag.get('value')
        else:
            value = tag
        value = str(value or '').strip()
        if value:
            labels.append(value)
    return ', '.join(labels) if labels else 'N/A'


def pingkai_support_base_url(client: TicketAPIClient) -> str:
    raw = str(getattr(client, 'base_url', '') or API_BASE_URL).rstrip('/')
    base = re.sub(r'/api(?:/v\d+)?$', '', raw, flags=re.I)
    return base or raw


def pingkai_ticket_link(ticket: dict, client: TicketAPIClient) -> str:
    support_base = pingkai_support_base_url(client)
    ticket_id = str(ticket.get('id') or '').strip()
    if ticket_id:
        return f'{support_base}/tickets/{ticket_id}'
    alias = ticket.get('alias')
    if alias not in (None, ''):
        return f'{support_base}/tickets/TICKET-{alias}'
    return support_base


def extract_pingkai_owner_conclusions(entries):
    root = ''
    workaround = ''
    todo = ''
    root_keywords = ('root cause', 'cause', 'reason', 'because', 'due to', 'likely', 'suspect', '根因', '原因', '问题在于', '导致', '未支持', '不支持')
    workaround_keywords = ('workaround', 'mitigation', 'temporary', 'restart', 'retry', 'upgrade', 'downgrade', 'config', 'timeout', '建议', '临时', '规避', '升级', '重启', '调整')
    todo_keywords = ('todo', 'follow up', 'follow-up', 'next step', 'need', 'please', 'check', 'collect', 'confirm', 'investigate', '排查', '确认', '收集', '继续', '跟进', '需要')

    for item in reversed(entries):
        body = clean_line(item.get('body', ''))
        if not body:
            continue
        lowered = body.lower()
        if not root:
            for label in ['Root cause', 'Cause', 'Reason', '根因', '原因', '结论']:
                value = extract_field_value(body, label)
                if value:
                    root = compact_summary(value, 220)
                    break
            if not root and any(token in lowered for token in root_keywords) and '?' not in body:
                root = compact_summary(first_sentence(body), 220)

        if not workaround:
            for label in ['Workaround', 'Mitigation', '建议', '规避', '处理建议', '解决办法']:
                value = extract_field_value(body, label)
                if value:
                    workaround = compact_summary(value, 220)
                    break
            if not workaround and any(token in lowered for token in workaround_keywords):
                workaround = compact_summary(first_sentence(body), 220)

        if not todo:
            for label in ['TODO', 'Follow-up', 'Next step', '下一步', '后续', '待办']:
                value = extract_field_value(body, label)
                if value:
                    todo = compact_summary(value, 220)
                    break
            if not todo and any(token in lowered for token in todo_keywords):
                todo = compact_summary(first_sentence(body), 220)

        if root and workaround and todo:
            break

    return (
        root or 'TBD',
        workaround or 'None yet',
        todo or 'TBD',
    )


def render_pingkai_summary_block(ticket: dict, client: TicketAPIClient) -> str:
    ticket_ref = pingkai_ticket_ref(ticket)
    title = ticket.get('title') or ticket_ref
    description_text = strip_text_for_pingkai_comment(ticket.get('description', ''))
    owner_entries = pingkai_owner_comment_entries(ticket)
    phenomenon = summarize_description_for_phenomenon(description_text)
    root_cause, workaround, todo = extract_pingkai_owner_conclusions(owner_entries)
    assignee = str(ticket.get('assignee_name') or '').strip()
    l2_assignee = str(ticket.get('l2_assignee_name') or '').strip()
    mentions = ' '.join([m for m in [f'@{assignee}' if assignee else '', f'@{l2_assignee}' if l2_assignee else ''] if m])

    heading = f'### {pingkai_priority_text(ticket)} {title} [{ticket_ref}]({pingkai_ticket_link(ticket, client)})'
    if mentions:
        heading = f'{heading} {mentions}'

    lines = [
        heading,
        f'- Labels: {pingkai_labels_text(ticket)}',
        f'- Version: {pingkai_version_text(ticket)}',
        f'- Phenomenon: {report_placeholder_blank(phenomenon)}',
        f'- Root cause: {report_placeholder_blank(root_cause)}',
        f'- Workaround: {workaround}',
        f'- TODO: {report_placeholder_blank(todo)}',
    ]
    return '\n'.join(lines)


def storage_weekly_mentions(*names: str) -> str:
    mentions = []
    seen = set()
    for name in names:
        text = str(name or '').strip()
        if not text or text == 'N/A':
            continue
        key = text.casefold()
        if key in seen:
            continue
        seen.add(key)
        mentions.append(f'@{text}')
    return ' '.join(mentions)


def storage_pingkai_ticket_ref(ticket: dict) -> str:
    alias = ticket.get('alias')
    if alias not in (None, ''):
        return f'TICKET-{alias}'
    ticket_id = str(ticket.get('id') or '').strip()
    return ticket_id or 'TICKET-N/A'


def render_storage_weekly_jira_block(key: str, raw: dict) -> str:
    fields = (raw or {}).get('fields', {})
    title = fields.get('summary') or key
    description_text = extract_text(fields.get('description'))
    owner_comment_text = collect_owner_comment_text(raw, fields)
    phenomenon = summarize_description_for_phenomenon(description_text)
    root_cause, workaround, todo = summarize_owner_comments(owner_comment_text)

    assignee = ((fields.get('assignee') or {}).get('displayName') or '').strip()
    l2_assignee = ((fields.get('customfield_10053') or {}).get('displayName') or '').strip()
    mentions = storage_weekly_mentions(l2_assignee, assignee)
    heading = f'### {title} [{key}]({issue_url(key)})'
    if mentions:
        heading = f'{heading} {mentions}'

    lines = [
        heading,
        f'- Version: {derive_version(fields)}',
        f'- Phenomenon: {report_placeholder_blank(phenomenon)}',
        f'- Root cause: {report_placeholder_blank(root_cause)}',
        f'- Workaround: {workaround}',
        f'- TODO: {report_placeholder_blank(todo)}',
    ]
    return '\n'.join(lines)


def render_storage_weekly_pingkai_block(ticket: dict, client: TicketAPIClient) -> str:
    ticket_ref = storage_pingkai_ticket_ref(ticket)
    title = ticket.get('title') or ticket_ref
    description_text = strip_text_for_pingkai_comment(ticket.get('description', ''))
    owner_entries = pingkai_owner_comment_entries(ticket)
    phenomenon = summarize_description_for_phenomenon(description_text)
    root_cause, workaround, todo = extract_pingkai_owner_conclusions(owner_entries)

    l2_assignee = str(ticket.get('l2_assignee_name') or '').strip()
    l3_assignee = str(ticket.get('l3_assignee_name') or '').strip()
    mentions = storage_weekly_mentions(l2_assignee, l3_assignee)
    heading = f'### {title} [{ticket_ref}]({pingkai_ticket_link(ticket, client)})'
    if mentions:
        heading = f'{heading} {mentions}'

    lines = [
        heading,
        f'- Version: {pingkai_version_text(ticket)}',
        f'- Phenomenon: {report_placeholder_blank(phenomenon)}',
        f'- Root cause: {report_placeholder_blank(root_cause)}',
        f'- Workaround: {workaround}',
        f'- TODO: {report_placeholder_blank(todo)}',
    ]
    return '\n'.join(lines)


def storage_weekly_pingkai_ticket_matches(ticket: dict) -> bool:
    l3_assignee = str(ticket.get('l3_assignee_name') or '').strip()
    l2_assignee = str(ticket.get('l2_assignee_name') or '').strip()
    return bool(l3_assignee) or l2_assignee == STORAGE_WEEKLY_DEFAULT_L2_ASSIGNEE


def make_storage_weekly_pingkai_args(component: str):
    return argparse.Namespace(
        status='all',
        time_range='all',
        page_size=100,
        components=component,
        module='',
        assignee='',
        l2_assignee='',
        l3_assignee='',
    )


def render_storage_weekly_report_from_args(args):
    try:
        start_time, end_time = resolve_time_window(args.start_time, args.end_time, default_days=7)
    except ValueError as e:
        print(str(e))
        return 1

    result = collect_storage_weekly_sections(
        STORAGE_WEEKLY_COMPONENTS,
        start_time,
        end_time,
        jira_source=make_jira_source_adapter(),
        pingkai_source=make_pingkai_source_adapter(),
        build_jira_entries=lambda component, current_start, current_end, jira_source: workflow_build_storage_weekly_jira_entries(
            component,
            current_start,
            current_end,
            jira_source,
            parse_sort_datetime=parse_sort_datetime,
            render_storage_weekly_jira_block=render_storage_weekly_jira_block,
        ),
        build_pingkai_entries=lambda component, current_start, current_end, pingkai_source: workflow_build_storage_weekly_pingkai_entries(
            component,
            current_start,
            current_end,
            pingkai_source,
            make_args=make_storage_weekly_pingkai_args,
            ticket_matches=storage_weekly_pingkai_ticket_matches,
            render_storage_weekly_pingkai_block=render_storage_weekly_pingkai_block,
        ),
    )
    if result.get('error_json'):
        print(json.dumps(result['error_json'], ensure_ascii=False, indent=2))
        return 1
    if result.get('returncode'):
        sys.stdout.write(result.get('stdout') or '')
        if result.get('stderr'):
            sys.stderr.write(result['stderr'])
        return result['returncode']

    sys.stdout.write(workflow_render_storage_weekly_report(result.get('sections') or []))
    return 0


def normalize_link_candidate(value: str) -> str:
    text = str(value or '').strip()
    text = text.replace('&amp;', '&')
    return text.rstrip('.,);]>\'"')


def extract_clinic_links_from_text(text: str) -> list[str]:
    links = []
    seen = set()
    for match in CLINIC_LINK_RE.finditer(str(text or '')):
        link = normalize_link_candidate(match.group(0))
        key = link.lower()
        if not link or key in seen:
            continue
        seen.add(key)
        links.append(link)
    return links


def extract_clinic_download_links_from_text(text: str) -> list[str]:
    links = []
    seen = set()
    for match in CLINIC_DOWNLOAD_LINK_RE.finditer(str(text or '')):
        link = normalize_link_candidate(match.group(0))
        key = link.lower()
        if not link or key in seen:
            continue
        seen.add(key)
        links.append(link)
    return links


def extract_pingkai_file_links_from_text(text: str) -> list[str]:
    links = []
    seen = set()
    for match in PINGKAI_FILE_LINK_RE.finditer(str(text or '')):
        link = normalize_link_candidate(match.group(0))
        if link.startswith('/'):
            link = urljoin(API_BASE_URL, link)
        key = link.lower()
        if not link or key in seen:
            continue
        seen.add(key)
        links.append(link)
    return links


def extract_grafana_links_from_text(text: str) -> list[str]:
    links = []
    seen = set()
    for match in HTTP_LINK_RE.finditer(str(text or '')):
        link = normalize_link_candidate(match.group(0))
        if not is_grafana_link(link):
            continue
        key = link.lower()
        if not link or key in seen:
            continue
        seen.add(key)
        links.append(link)
    return links


def compact_comment_preview(text: str, limit: int = 160) -> str:
    return compact_summary(text, limit) or 'N/A'


def extract_jira_clinic_references(raw: dict) -> list[dict]:
    refs = []
    fields = (raw or {}).get('fields') or {}

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_clinic_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                }
            )

    def add_rich_links(value, source: str, author: str = '', created: str = ''):
        for link in extract_urls_from_rich_value(value):
            for normalized in extract_clinic_links_from_text(link):
                refs.append(
                    {
                        'link': normalized,
                        'source': source,
                        'author': author or '',
                        'created': created or '',
                    }
                )

    add_links(fields.get('summary') or '', 'summary')
    add_links(extract_text(fields.get('description')), 'description')
    add_rich_links(fields.get('description'), 'description_rich')
    add_links(json.dumps(fields.get('description') or {}, ensure_ascii=False), 'description_raw')
    comments = (((raw or {}).get('fields') or {}).get('comment') or {}).get('comments') or []
    for entry in extract_comment_entries_with_time(raw):
        add_links(entry.get('body', ''), 'comment', author=entry.get('author', ''), created=entry.get('created', ''))
    for comment in comments:
        author = (((comment or {}).get('author') or {}).get('displayName') or '').strip()
        created = str((comment or {}).get('created') or '').strip()
        add_rich_links((comment or {}).get('body'), 'comment_rich', author=author, created=created)

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)

    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_jira_grafana_references(raw: dict) -> list[dict]:
    refs = []
    fields = (raw or {}).get('fields') or {}

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_grafana_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                }
            )

    def add_rich_links(value, source: str, author: str = '', created: str = ''):
        for link in extract_urls_from_rich_value(value):
            for normalized in extract_grafana_links_from_text(link):
                refs.append(
                    {
                        'link': normalized,
                        'source': source,
                        'author': author or '',
                        'created': created or '',
                    }
                )

    add_links(fields.get('summary') or '', 'summary')
    add_links(extract_text(fields.get('description')), 'description')
    add_rich_links(fields.get('description'), 'description_rich')
    add_links(json.dumps(fields.get('description') or {}, ensure_ascii=False), 'description_raw')
    comments = (((raw or {}).get('fields') or {}).get('comment') or {}).get('comments') or []
    for entry in extract_comment_entries_with_time(raw):
        add_links(entry.get('body', ''), 'comment', author=entry.get('author', ''), created=entry.get('created', ''))
    for comment in comments:
        author = (((comment or {}).get('author') or {}).get('displayName') or '').strip()
        created = str((comment or {}).get('created') or '').strip()
        add_rich_links((comment or {}).get('body'), 'comment_rich', author=author, created=created)

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)

    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_jira_clinic_download_references(raw: dict) -> list[dict]:
    refs = []
    fields = (raw or {}).get('fields') or {}

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_clinic_download_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                }
            )

    def add_rich_links(value, source: str, author: str = '', created: str = ''):
        for link in extract_urls_from_rich_value(value):
            for normalized in extract_clinic_download_links_from_text(link):
                refs.append(
                    {
                        'link': normalized,
                        'source': source,
                        'author': author or '',
                        'created': created or '',
                    }
                )

    add_links(fields.get('summary') or '', 'summary')
    add_links(extract_text(fields.get('description')), 'description')
    add_rich_links(fields.get('description'), 'description_rich')
    add_links(json.dumps(fields.get('description') or {}, ensure_ascii=False), 'description_raw')
    comments = (((raw or {}).get('fields') or {}).get('comment') or {}).get('comments') or []
    for entry in extract_comment_entries_with_time(raw):
        add_links(entry.get('body', ''), 'comment', author=entry.get('author', ''), created=entry.get('created', ''))
    for comment in comments:
        author = (((comment or {}).get('author') or {}).get('displayName') or '').strip()
        created = str((comment or {}).get('created') or '').strip()
        add_rich_links((comment or {}).get('body'), 'comment_rich', author=author, created=created)

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)

    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_pingkai_clinic_references(ticket: dict) -> list[dict]:
    refs = []

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_clinic_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                }
            )

    add_links(ticket.get('title') or '', 'summary')
    add_links(strip_text_for_pingkai_comment(ticket.get('description', '')), 'description')
    for entry in pingkai_comment_entries(ticket):
        add_links(entry.get('body', ''), 'comment', author=entry.get('author_name', ''), created=entry.get('created', ''))

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)

    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_pingkai_grafana_references(ticket: dict) -> list[dict]:
    refs = []

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_grafana_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                }
            )

    add_links(ticket.get('title') or '', 'summary')
    add_links(strip_text_for_pingkai_comment(ticket.get('description', '')), 'description')
    for entry in pingkai_comment_entries(ticket):
        add_links(entry.get('body', ''), 'comment', author=entry.get('author_name', ''), created=entry.get('created', ''))

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)

    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_pingkai_clinic_download_references(ticket: dict) -> list[dict]:
    refs = []

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_clinic_download_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                }
            )

    add_links(ticket.get('title') or '', 'summary')
    add_links(strip_text_for_pingkai_comment(ticket.get('description', '')), 'description')
    for entry in pingkai_comment_entries(ticket):
        add_links(entry.get('body', ''), 'comment', author=entry.get('author_name', ''), created=entry.get('created', ''))

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)

    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_jira_initial_clinic_links(raw: dict) -> list[str]:
    fields = (raw or {}).get('fields') or {}
    text_parts = [
        str(fields.get('summary') or ''),
        extract_text(fields.get('description')),
        json.dumps(fields.get('description') or {}, ensure_ascii=False),
    ]
    links = []
    seen = set()
    for text in text_parts:
        for link in extract_clinic_links_from_text(text):
            key = link.lower()
            if key in seen:
                continue
            seen.add(key)
            links.append(link)
    return links


def extract_jira_initial_grafana_links(raw: dict) -> list[str]:
    fields = (raw or {}).get('fields') or {}
    text_parts = [
        str(fields.get('summary') or ''),
        extract_text(fields.get('description')),
        json.dumps(fields.get('description') or {}, ensure_ascii=False),
    ]
    links = []
    seen = set()
    for text in text_parts:
        for link in extract_grafana_links_from_text(text):
            key = link.lower()
            if key in seen:
                continue
            seen.add(key)
            links.append(link)
    return links


def extract_pingkai_initial_clinic_links(ticket: dict) -> list[str]:
    text_parts = [
        str(ticket.get('title') or ''),
        strip_text_for_pingkai_comment(ticket.get('description', '')),
        str(ticket.get('description') or ''),
    ]
    links = []
    seen = set()
    for text in text_parts:
        for link in extract_clinic_links_from_text(text):
            key = link.lower()
            if key in seen:
                continue
            seen.add(key)
            links.append(link)
    return links


def extract_pingkai_initial_grafana_links(ticket: dict) -> list[str]:
    text_parts = [
        str(ticket.get('title') or ''),
        strip_text_for_pingkai_comment(ticket.get('description', '')),
        str(ticket.get('description') or ''),
    ]
    links = []
    seen = set()
    for text in text_parts:
        for link in extract_grafana_links_from_text(text):
            key = link.lower()
            if key in seen:
                continue
            seen.add(key)
            links.append(link)
    return links


def extract_pingkai_attachment_references(ticket: dict) -> list[dict]:
    refs = []

    def add_links(text: str, source: str, author: str = '', created: str = ''):
        for link in extract_pingkai_file_links_from_text(text):
            refs.append(
                {
                    'link': link,
                    'source': source,
                    'author': author or '',
                    'created': created or '',
                    'provider': 'pingkai',
                }
            )

    add_links(ticket.get('description') or '', 'description')
    for entry in pingkai_comment_entries(ticket):
        add_links(entry.get('content', '') or entry.get('body', ''), 'comment', author=entry.get('author_name', ''), created=entry.get('created', ''))

    refs.sort(key=lambda item: (item.get('created') or '', item.get('source') == 'comment'), reverse=True)
    deduped = []
    seen = set()
    for item in refs:
        key = item['link'].lower()
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def extract_jira_attachment_references(raw: dict) -> list[dict]:
    fields = (raw or {}).get('fields') or {}
    refs = []
    for item in (fields.get('attachment') or []):
        refs.append(
            {
                'provider': 'jira',
                'id': str(item.get('id') or '').strip(),
                'link': str(item.get('content') or '').strip(),
                'filename': str(item.get('filename') or '').strip(),
                'mime_type': str(item.get('mimeType') or '').strip(),
                'size': item.get('size'),
                'created': str(item.get('created') or '').strip(),
                'author': str(((item.get('author') or {}).get('displayName') or '')).strip(),
                'source': 'attachment',
            }
        )
    return [item for item in refs if item.get('link')]


def recent_jira_comment_context(raw: dict, limit: int = 5) -> list[dict]:
    entries = extract_comment_entries_with_time(raw)
    result = []
    for item in entries[-limit:]:
        result.append(
            {
                'created': item.get('created') or '',
                'author': item.get('author') or 'N/A',
                'summary': compact_comment_preview(item.get('body', '')),
            }
        )
    return result


def jira_comment_corpus_context(raw: dict, limit: int = 40) -> list[dict]:
    entries = extract_comment_entries_with_time(raw)
    result = []
    for item in entries[:limit]:
        result.append(
            {
                'created': item.get('created') or '',
                'author': item.get('author') or 'N/A',
                'summary': compact_comment_preview(item.get('body', ''), limit=220),
            }
        )
    return result


def recent_pingkai_comment_context(ticket: dict, limit: int = 5) -> list[dict]:
    entries = pingkai_comment_entries(ticket)
    result = []
    for item in entries[-limit:]:
        result.append(
            {
                'created': item.get('created') or '',
                'author': item.get('author_name') or item.get('author_username') or 'N/A',
                'summary': compact_comment_preview(item.get('body', '')),
                'is_internal': bool(item.get('is_internal')),
            }
        )
    return result


def pingkai_comment_corpus_context(ticket: dict, limit: int = 60) -> list[dict]:
    entries = pingkai_comment_entries(ticket)
    result = []
    for item in entries[:limit]:
        result.append(
            {
                'created': item.get('created') or '',
                'author': item.get('author_name') or item.get('author_username') or 'N/A',
                'summary': compact_comment_preview(item.get('body', ''), limit=220),
                'is_internal': bool(item.get('is_internal')),
            }
        )
    return result


def sanitize_issue_dir_name(value: str) -> str:
    text = str(value or '').strip()
    text = text.replace('/', '_').replace('\\', '_').strip(' .')
    return text or 'issue'


def issue_download_dir(issue_ref: str, root: str = '') -> Path:
    base = Path(root).expanduser() if str(root or '').strip() else Path.home() / 'Downloads'
    return base / sanitize_issue_dir_name(issue_ref)


def issue_attachment_dir(issue_ref: str, root: str = '') -> Path:
    return issue_download_dir(issue_ref, root=root) / 'attachments'


def response_filename_from_headers(headers) -> str:
    value = str(headers.get('Content-Disposition') or headers.get('content-disposition') or '').strip()
    if 'filename=' not in value:
        return ''
    filename = value.split('filename=', 1)[1].strip().strip('"')
    return filename


def sanitize_attachment_name(value: str, fallback: str = 'attachment') -> str:
    raw_name = str(value or '').strip().replace('/', '_').replace('\\', '_')
    if not raw_name:
        raw_name = fallback
    return raw_name


def should_skip_attachment_download(ref: dict) -> bool:
    filename = str(ref.get('filename') or '').strip().lower()
    mime_type = str(ref.get('mime_type') or '').strip().lower()
    link = str(ref.get('link') or '').strip().lower()
    guessed_type = ''
    if filename:
        guessed_type = str(mimetypes.guess_type(filename)[0] or '').strip().lower()
    elif link:
        guessed_type = str(mimetypes.guess_type(link)[0] or '').strip().lower()

    if any(filename.endswith(ext) for ext in SKIPPED_IMAGE_ATTACHMENT_SUFFIXES):
        return True
    if any(mime_type.startswith(prefix) for prefix in SKIPPED_IMAGE_ATTACHMENT_MIME_PREFIXES):
        return True
    if any(guessed_type.startswith(prefix) for prefix in SKIPPED_IMAGE_ATTACHMENT_MIME_PREFIXES):
        return True
    if any(link.endswith(ext) for ext in SKIPPED_IMAGE_ATTACHMENT_SUFFIXES):
        return True
    return False


def context_requires_attachment_analysis(context_text: str, attachment_refs: list[dict]) -> bool:
    lowered = str(context_text or '').lower()
    if any(token in lowered for token in ATTACHMENT_ANALYSIS_HINTS):
        return True
    for item in attachment_refs or []:
        filename = str(item.get('filename') or '').lower()
        mime_type = str(item.get('mime_type') or '').lower()
        if any(token in filename for token in PROFILE_ATTACHMENT_HINTS):
            return True
        if any(filename.endswith(ext) for ext in TEXT_ATTACHMENT_SUFFIXES):
            return True
        if any(token in mime_type for token in ['text/', 'json', 'yaml']):
            return True
    return False


def attachment_kind(filename: str, mime_type: str) -> str:
    low_name = str(filename or '').lower()
    low_type = str(mime_type or '').lower()
    if any(token in low_name for token in PROFILE_ATTACHMENT_HINTS):
        return 'profile'
    if any(low_name.endswith(ext) for ext in TEXT_ATTACHMENT_SUFFIXES):
        return 'text'
    if low_type.startswith('text/') or 'json' in low_type or 'yaml' in low_type:
        return 'text'
    if low_type.startswith('image/'):
        return 'image'
    if low_name.endswith(('.svg', '.html')):
        return 'profile'
    return 'binary'


def read_attachment_preview(path: Path, limit: int = 200000) -> str:
    data = path.read_bytes()[:limit]
    return data.decode('utf-8', errors='replace')


SLOW_LOG_STAGE_PATTERNS = {
    'prewrite': re.compile(r'prewrite:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'store_batch_wait': re.compile(r'store_batch_wait:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'persist_log_total': re.compile(r'persist_log:\s*\{[^}]*?total:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'sync_log': re.compile(r'sync_log:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'commit_log': re.compile(r'commit_log:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'apply_batch_wait': re.compile(r'apply_batch_wait:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'apply_total': re.compile(r'apply:\s*\{[^}]*?total:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'write_wal': re.compile(r'write_wal:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
    'tikv_wall_time': re.compile(r'tikv_wall_time:\s*([0-9.]+(?:µs|us|ms|s))', re.I),
}
SLOW_LOG_STORE_RE = re.compile(r'store:\s*([^,\s}]+)', re.I)
SLOW_LOG_INSTANCE_STORE_RE = re.compile(r'([A-Za-z0-9._-]*tikv-\d[^\s,:}]*)', re.I)
SLOW_LOG_DURATION_RE = re.compile(r'^([0-9.]+)(µs|us|ms|s)$', re.I)
SLOW_LOG_DIRECT_COLUMNS = {
    'query_time': 'query_time',
    'prewrite_time': 'prewrite_time',
    'commit_time': 'commit_time',
    'process_time': 'process_time',
    'wait_time': 'wait_time',
    'backoff_time': 'backoff_time',
    'resolve_lock_time': 'resolve_lock_time',
    'cop_time': 'cop_time',
    'wait_ts': 'wait_ts',
}


def _safe_float(value) -> float | None:
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def parse_duration_seconds(raw: str) -> float | None:
    text = str(raw or '').strip()
    if not text:
        return None
    match = SLOW_LOG_DURATION_RE.match(text)
    if not match:
        return None
    value = _safe_float(match.group(1))
    if value is None:
        return None
    unit = match.group(2).lower()
    if unit in {'µs', 'us'}:
        return value / 1_000_000.0
    if unit == 'ms':
        return value / 1000.0
    return value


def summarize_numeric_samples(samples: list[float]) -> dict:
    if not samples:
        return {}
    ordered = sorted(float(item) for item in samples if item is not None)
    if not ordered:
        return {}

    def percentile(ratio: float) -> float:
        if len(ordered) == 1:
            return ordered[0]
        index = max(0, min(len(ordered) - 1, int(round((len(ordered) - 1) * ratio))))
        return ordered[index]

    return {
        'count': len(ordered),
        'median_seconds': ordered[len(ordered) // 2],
        'p95_seconds': percentile(0.95),
        'max_seconds': ordered[-1],
    }


def summarize_high_ts_windows(timestamps: list[float], *, gap_seconds: int = 4 * 3600) -> list[dict]:
    ordered = sorted(int(ts) for ts in timestamps if ts is not None)
    if not ordered:
        return []
    groups: list[list[int]] = [[ordered[0]]]
    for value in ordered[1:]:
        if value - groups[-1][-1] > gap_seconds:
            groups.append([value])
        else:
            groups[-1].append(value)
    windows = []
    for group in groups:
        windows.append(
            {
                'from_ts': group[0],
                'to_ts': group[-1],
                'from': format_unix_ts_local(group[0]),
                'to': format_unix_ts_local(group[-1]),
                'sample_count': len(group),
            }
        )
    return windows


def slow_log_stage_priority(name: str) -> tuple[int, int]:
    stage = str(name or '').strip()
    priority_map = {
        'sync_log': 0,
        'persist_log_total': 1,
        'write_wal': 2,
        'commit_log': 3,
        'store_batch_wait': 4,
        'apply_batch_wait': 5,
        'prewrite': 6,
        'prewrite_time': 7,
        'commit_time': 8,
        'wait_ts': 9,
        'cop_time': 10,
        'process_time': 11,
        'query_time': 20,
    }
    return (priority_map.get(stage, 15), len(stage))


def slow_log_focus_hints(top_stages: list[dict]) -> list[str]:
    hints = []
    names = {str(item.get('name') or '') for item in top_stages}
    if {'sync_log', 'persist_log_total', 'write_wal'} & names:
        hints.append('Clinic follow-up should prioritize store-local WAL / disk tail metrics on p99.999, p99.9999, or max instead of stopping at p99.')
    if {'commit_log', 'store_batch_wait', 'apply_batch_wait'} & names:
        hints.append('Clinic follow-up should split queueing before persist from persist itself and check whether the slowest store is localized.')
    if 'prewrite' in names:
        hints.append('Slow-log prewrite tails should be treated as the entry anchor for the TiKV write path, then narrowed with store-level Clinic metrics.')
    return hints


def analyze_slow_log_csv(path: Path) -> dict:
    stage_samples: dict[str, list[float]] = {}
    direct_samples: dict[str, list[float]] = {}
    stores: dict[str, int] = {}
    instances: dict[str, int] = {}
    suspicious_lines: list[str] = []
    suspicious_timestamps: list[float] = []
    timestamp_values: list[float] = []
    row_count = 0
    high_query_rows = 0

    with path.open('r', encoding='utf-8', errors='replace', newline='') as fh:
        reader = csv.DictReader(fh)
        if not reader.fieldnames:
            return {}
        for row in reader:
            row_count += 1
            timestamp = _safe_float(row.get('timestamp'))
            if timestamp is not None:
                timestamp_values.append(timestamp)
            query_time = _safe_float(row.get('query_time')) or 0.0
            if query_time >= 1.0:
                high_query_rows += 1
                if timestamp is not None:
                    suspicious_timestamps.append(timestamp)

            # Some exported slow-query CSVs encode stage details inside textual plan cells.
            row_text = ' '.join(str(value or '') for value in row.values())
            if len(suspicious_lines) < 8 and any(token in row_text.lower() for token in ('prewrite:', 'sync_log', 'persist_log', 'store_batch_wait', 'commit_log')):
                suspicious_lines.append(compact_summary(row_text, 260))

            for column_name, alias in SLOW_LOG_DIRECT_COLUMNS.items():
                value = _safe_float(row.get(column_name))
                if value is not None and value > 0:
                    direct_samples.setdefault(alias, []).append(value)
                    if value >= 0.03 and timestamp is not None:
                        suspicious_timestamps.append(timestamp)

            for stage_name, pattern in SLOW_LOG_STAGE_PATTERNS.items():
                for match in pattern.finditer(row_text):
                    seconds = parse_duration_seconds(match.group(1))
                    if seconds is None:
                        continue
                    stage_samples.setdefault(stage_name, []).append(seconds)
                    if seconds >= 0.01 and timestamp is not None:
                        suspicious_timestamps.append(timestamp)

            for match in SLOW_LOG_STORE_RE.finditer(row_text):
                store = str(match.group(1) or '').strip()
                if store:
                    stores[store] = stores.get(store, 0) + 1
            instance = str(row.get('instance') or '').strip()
            if instance:
                instances[instance] = instances.get(instance, 0) + 1
            for match in SLOW_LOG_INSTANCE_STORE_RE.finditer(row_text):
                store_hint = str(match.group(1) or '').strip()
                if store_hint:
                    stores[store_hint] = stores.get(store_hint, 0) + 1

    if row_count == 0:
        return {}

    combined_stage_stats = []
    for name, samples in {**direct_samples, **stage_samples}.items():
        stats = summarize_numeric_samples(samples)
        if not stats:
            continue
        combined_stage_stats.append({'name': name, **stats})
    combined_stage_stats.sort(
        key=lambda item: (
            slow_log_stage_priority(item.get('name')),
            -(float(item.get('max_seconds') or 0.0)),
            -(float(item.get('p95_seconds') or 0.0)),
            -(int(item.get('count') or 0)),
        ),
    )
    top_stages = combined_stage_stats[:5]
    windows = summarize_high_ts_windows(suspicious_timestamps or timestamp_values)
    dominant_store = ''
    if stores:
        dominant_store = max(stores.items(), key=lambda item: item[1])[0]

    summary_lines = []
    if windows:
        first_window = windows[0]
        summary_lines.append(
            f'Slow-log window spans {first_window["from"]} to {windows[-1]["to"]} with {len(windows)} distinct high-latency cluster(s).'
        )
    if top_stages:
        stage = top_stages[0]
        summary_lines.append(
            f'Dominant slow-log stage is `{stage["name"]}` with max {format_seconds(stage.get("max_seconds"))} '
            f'and p95 {format_seconds(stage.get("p95_seconds"))}.'
        )
    if dominant_store:
        summary_lines.append(f'Most repeated TiKV/store hint in slow logs is `{dominant_store}`.')

    return {
        'detected': True,
        'row_count': row_count,
        'high_query_rows': high_query_rows,
        'time_range': {
            'from': format_unix_ts_local(min(timestamp_values)) if timestamp_values else 'N/A',
            'to': format_unix_ts_local(max(timestamp_values)) if timestamp_values else 'N/A',
            'from_ts': int(min(timestamp_values)) if timestamp_values else 0,
            'to_ts': int(max(timestamp_values)) if timestamp_values else 0,
        },
        'instance_count': len(instances),
        'store_count': len(stores),
        'dominant_store': dominant_store,
        'top_stages': top_stages,
        'windows': windows,
        'focus_hints': slow_log_focus_hints(top_stages),
        'summary_lines': summary_lines,
        'sample_lines': suspicious_lines[:5],
    }


def analyze_downloaded_attachment(path: Path, meta: dict) -> dict:
    filename = str(path.name)
    mime_type = str(meta.get('mime_type') or '')
    kind = attachment_kind(filename, mime_type)
    result = {
        'path': str(path),
        'filename': filename,
        'mime_type': mime_type,
        'size': path.stat().st_size if path.exists() else 0,
        'kind': kind,
    }
    if kind == 'text':
        try:
            preview = read_attachment_preview(path)
        except Exception as exc:
            result['analysis_error'] = str(exc)
            return result
        lines = [line.strip() for line in preview.splitlines() if line.strip()]
        suspicious = []
        pattern = re.compile(r'error|warn|panic|exception|timeout|slow|leader drop|write too slow|stack|trace|fail', re.I)
        for line in lines:
            if pattern.search(line):
                suspicious.append(line[:300])
            if len(suspicious) >= 8:
                break
        result['preview_excerpt'] = preview[:1000]
        result['suspicious_lines'] = suspicious
        low_name = filename.lower()
        if low_name.endswith('.csv') and 'slowquery' in low_name:
            try:
                slow_log_summary = analyze_slow_log_csv(path)
            except Exception as exc:
                result['slow_log_analysis_error'] = str(exc)
            else:
                if slow_log_summary:
                    result['slow_log_summary'] = slow_log_summary
                    for line in slow_log_summary.get('sample_lines') or []:
                        if line not in result['suspicious_lines']:
                            result['suspicious_lines'].append(line)
                        if len(result['suspicious_lines']) >= 8:
                            break
    elif kind == 'profile':
        result['profile_hint'] = 'profile-like attachment downloaded; manual or specialized profile analysis may be required'
    return result


def add_skill(skills: list[str], seen: set[str], value: str) -> None:
    text = str(value or '').strip()
    if not text or text in seen:
        return
    seen.add(text)
    skills.append(text)


def recommend_diagnosis_skills(context_text: str, clinic_refs: list[dict]) -> tuple[list[str], list[str]]:
    text = str(context_text or '')
    lowered = text.lower()
    skills = []
    seen: set[str] = set()
    reasons = []

    add_skill(skills, seen, 'tidb-perf-diagnosis')
    reasons.append('Use tidb-perf-diagnosis as the default performance triage skill for ticket-driven diagnosis.')

    if clinic_refs:
        add_skill(skills, seen, 'op-clinic-grafana-access')
        reasons.append('A Clinic or Grafana monitoring reference was found in the ticket context, so op-clinic-grafana-access can provide normalized monitoring evidence.')

    if any(token in lowered for token in ['leader drop', 'leader missing', 're-election', '掉leader', '掉主', '选主']):
        add_skill(skills, seen, 'tikv-slow-diagnosis')
        reasons.append('The primary symptom mentions leader drop or re-election, so tikv-slow-diagnosis should use the dedicated leader-drop workflow.')

    if any(token in lowered for token in ['io jitter', 'disk jitter', 'ebs', 'volume io exceeded', 'disk wait latency', 'io抖动', '盘抖动', '磁盘抖动', 'store-local tail latency']):
        add_skill(skills, seen, 'tikv-slow-diagnosis')
        reasons.append('The ticket mentions IO-jitter-style keywords, so tikv-slow-diagnosis should use the dedicated IO-jitter workflow.')

    if any(token in lowered for token in ['tikvrpc', 'batch receive', 'wait connection establish', 'rpc layer latency', 'unreachable', 'network jitter', 'network latency', 'network interruption', 'network saturation', '带宽打满', '网络抖动', '网络抖动', '网络异常']):
        add_skill(skills, seen, 'tikv-slow-diagnosis')
        reasons.append('The ticket mentions TiKV-network-style signals, so tikv-slow-diagnosis should use the dedicated network troubleshooting workflow.')

    if any(token in lowered for token in ['serverisbusy', 'server is busy', 'deadline exceeded', 'flow control', 'throttle duration', 'discard ratio', 'scheduler pending write', 'pending compaction', 'l0 file', 'memtable', 'memory_quota_exceeded', 'max_pending_tasks_exceeded', 'kv backoff', 'channelfull', 'raftstore is busy']):
        add_skill(skills, seen, 'tikv-slow-diagnosis')
        reasons.append('The ticket mentions ServerIsBusy or throttling signals, so tikv-slow-diagnosis should use the dedicated flow-control workflow.')

    if any(token in lowered for token in ['changefeed', 'ticdc', 'cdc lag', 'replication lag', 'sink uri']):
        add_skill(skills, seen, 'ticdc-health-inspection')
        reasons.append('The ticket mentions TiCDC / changefeed signals, so ticdc-health-inspection is relevant.')

    if any(token in lowered for token in ['oom', 'out of memory', 'memory leak', 'swap', 'allocstall', 'memory', '内存']):
        add_skill(skills, seen, 'tikv-memory')
        reasons.append('The ticket mentions memory pressure or OOM symptoms, so tikv-memory is relevant.')

    if any(token in lowered for token in ['unsafe recover', 'recovery', 'replica loss', 'tombstone', 'data loss']):
        add_skill(skills, seen, 'tikv-recovery')
        reasons.append('The ticket mentions recovery / replica-loss keywords, so tikv-recovery is relevant.')

    if any(token in lowered for token in ['deadlock', 'pessimistic lock', 'resolve lock', 'async commit', 'stale read', 'transaction']):
        add_skill(skills, seen, 'transaction-oncall-handbook')
        reasons.append('The ticket mentions transaction or lock-related keywords, so transaction-oncall-handbook is relevant.')

    if any(token in lowered for token in ['commit log', 'raftstore', 'grpc', 'tikv latency', 'write slow', 'read slow', 'raft']):
        add_skill(skills, seen, 'tikv-fast-tune')
        reasons.append('The ticket mentions TiKV / raft / gRPC latency signals, so tikv-fast-tune is relevant.')

    return skills, reasons


def is_http_link(text: str) -> bool:
    value = str(text or '').strip().lower()
    return value.startswith('http://') or value.startswith('https://')


def is_grafana_link(text: str) -> bool:
    value = str(text or '').strip().lower()
    if not is_http_link(value):
        return False
    return '/d/' in value or ':3000' in value or '/grafana' in value


def clinic_link_has_window(link: str) -> bool:
    parsed = urlparse(str(link or '').strip())
    fragment = parsed.fragment or ''
    _, _, fragment_query = fragment.partition('?')
    query = parse_qs(fragment_query, keep_blank_values=True)
    return bool((query.get('from') or [''])[0] and (query.get('to') or [''])[0])


def apply_window_to_clinic_link(link: str, from_ts: str, to_ts: str) -> str:
    raw_link = str(link or '').strip()
    if not raw_link or clinic_link_has_window(raw_link):
        return raw_link
    if not str(from_ts or '').strip() or not str(to_ts or '').strip():
        return raw_link
    parsed = urlparse(raw_link)
    fragment = parsed.fragment or ''
    fragment_path, _, fragment_query = fragment.partition('?')
    query = parse_qs(fragment_query, keep_blank_values=True)
    query['from'] = [str(from_ts).strip()]
    query['to'] = [str(to_ts).strip()]
    new_fragment = f'{fragment_path}?{urlencode(query, doseq=True)}'
    return urlunparse(parsed._replace(fragment=new_fragment))


def prepend_reference(refs: list[dict], link: str, source: str) -> list[dict]:
    normalized = str(link or '').strip()
    if not normalized:
        return refs
    out = [{'link': normalized, 'source': source, 'author': '', 'created': ''}]
    seen = {normalized.lower()}
    for item in refs or []:
        item_link = str((item or {}).get('link') or '').strip()
        if not item_link or item_link.lower() in seen:
            continue
        seen.add(item_link.lower())
        out.append(item)
    return out


def run_json_script(script_path: Path, argv: list[str], *, missing_label: str, failure_label: str) -> tuple[dict, str]:
    if not script_path.exists():
        return {}, f'{missing_label} script not found: {script_path}'

    proc = run([sys.executable, str(script_path), *argv], check=False)
    stdout = (proc.stdout or '').strip()
    stderr = (proc.stderr or '').strip()
    if proc.returncode != 0:
        return {}, stderr or stdout or f'{failure_label} failed with exit code {proc.returncode}'

    try:
        return json.loads(stdout), ''
    except Exception:
        return {}, f'{failure_label} returned non-JSON output'


def run_clinic_access_dispatch(route: str, target: str, args, *, extra_argv: list[str] | None = None) -> tuple[dict, str]:
    argv = [target, '--route', route]
    if extra_argv:
        argv.extend(extra_argv)
    return run_json_script(
        CLINIC_ACCESS_DISPATCH_SCRIPT,
        argv,
        missing_label='clinic/grafana access dispatcher',
        failure_label=f'{route} access dispatcher',
    )


def run_clinic_analyze(link: str, args) -> dict:
    clinic_base_url = str(getattr(args, 'clinic_base_url', '') or '').strip()
    clinic_profile = effective_clinic_profile(
        args,
        candidate_base_url=normalized_base_url(clinic_base_url or link),
    )
    clinic_dashboard_name = str(getattr(args, 'clinic_dashboard_name', '') or '').strip()
    dispatch_argv: list[str] = []
    if clinic_base_url:
        dispatch_argv.extend(['--base-url', clinic_base_url])
    if clinic_profile:
        dispatch_argv.extend(['--profile', clinic_profile])
    if clinic_dashboard_name:
        dispatch_argv.extend(['--dashboard-name', clinic_dashboard_name])
    dispatch_argv.extend([
        '--include-metrics-evidence',
        '--include-tikv-write-path-evidence',
    ])

    access_payload, access_error = run_clinic_access_dispatch(
        'clinic',
        link,
        args,
        extra_argv=dispatch_argv,
    )
    if access_error:
        return {
            'link': link,
            'error': access_error,
        }

    result = {
        'link': link,
        'source': access_payload.get('source') or {},
        'clinic_context': access_payload.get('clinic_context') or {},
        'data_status': access_payload.get('data_status') or {},
        'access_warnings': list(access_payload.get('access_warnings') or []),
        'dashboard_discovery': access_payload.get('dashboard_discovery') or {},
        'cluster_meta': access_payload.get('cluster_meta') or {},
        'cluster_summary': access_payload.get('cluster_summary') or {},
        'metrics_evidence': access_payload.get('metrics_evidence') or {},
        'tikv_write_path_evidence': access_payload.get('tikv_write_path_evidence') or {},
    }

    if not result['data_status'] or not bool(result['data_status'].get('ready')):
        if bool(getattr(args, 'full_clinic_json', False)):
            result['payload'] = {'access': access_payload}
        return result

    if bool(getattr(args, 'full_clinic_json', False)):
        result['payload'] = access_payload
    return result


def run_grafana_analyze(link: str, args) -> dict:
    grafana_profile = str(getattr(args, 'grafana_profile', '') or '').strip()
    if not grafana_profile:
        grafana_profile = effective_clinic_profile(
            args,
            candidate_base_url=normalized_base_url(link),
        )
    grafana_datasource_id = str(getattr(args, 'grafana_datasource_id', '') or '1').strip()
    grafana_dashboard_uid = str(getattr(args, 'grafana_dashboard_uid', '') or '').strip()
    from_ts = str(getattr(args, 'from_ts', '') or '').strip()
    to_ts = str(getattr(args, 'to_ts', '') or '').strip()
    dispatch_argv: list[str] = []
    if grafana_profile:
        dispatch_argv.extend(['--profile', grafana_profile])
    if grafana_datasource_id:
        dispatch_argv.extend(['--datasource-id', grafana_datasource_id])
    if grafana_dashboard_uid:
        dispatch_argv.extend(['--dashboard-uid', grafana_dashboard_uid])
    if from_ts:
        dispatch_argv.extend(['--from', from_ts])
    if to_ts:
        dispatch_argv.extend(['--to', to_ts])

    payload, error = run_clinic_access_dispatch(
        'grafana',
        link,
        args,
        extra_argv=dispatch_argv,
    )
    if error:
        return {
            'link': link,
            'error': error,
        }

    result = {
        'link': link,
        'source': payload.get('source') or {},
        'grafana_context': payload.get('grafana_context') or {},
        'access_warnings': payload.get('access_warnings') or [],
        'cluster_summary': payload.get('cluster_summary') or {},
        'dashboard_discovery': payload.get('dashboard_discovery') or {},
    }
    if bool(getattr(args, 'full_clinic_json', False)):
        result['payload'] = payload
    return result


def run_clinic_bundle_download(download_url: str, issue_ref: str, args, status_link: str = '') -> dict:
    status_link = str(status_link or '').strip()
    if not status_link:
        return {
            'download_url': download_url,
            'error': 'clinic bundle download now requires a matching Clinic portal link; direct Clinic download URLs are no longer supported as the bundle target',
        }

    dispatch_argv = ['--output-subdir', str(issue_ref or 'clinic-bundle')]
    clinic_profile = effective_clinic_profile(
        args,
        candidate_base_url=normalized_base_url(status_link),
    )
    clinic_download_root = str(getattr(args, 'clinic_download_root', '') or '').strip()
    if clinic_profile:
        dispatch_argv.extend(['--profile', clinic_profile])
    if clinic_download_root:
        dispatch_argv.extend(['--output-root', clinic_download_root])

    payload, error = run_clinic_access_dispatch(
        'bundle',
        status_link,
        args,
        extra_argv=dispatch_argv,
    )
    if error:
        return {
            'download_url': download_url,
            'error': error,
        }

    logs = payload.get('logs') or {}
    components = logs.get('components') or {}

    def hosts_with_logs(items: list[dict]) -> list[str]:
        return [str(item.get('host') or '').strip() for item in items if item.get('log_files')]

    result = {
        'download_url': download_url,
        'archive_path': ((payload.get('download') or {}).get('archive_path')) or '',
        'extract_dir': ((payload.get('extraction') or {}).get('extract_dir')) or '',
        'meta_yaml_path': ((payload.get('meta') or {}).get('meta_yaml_path')) or '',
        'tidb_version': ((payload.get('meta') or {}).get('tidb_version')) or '',
        'component_counts': ((payload.get('meta') or {}).get('component_counts')) or {},
        'warnings': logs.get('warnings') or [],
        'pd_hosts_with_logs': hosts_with_logs(components.get('pd') or []),
        'tikv_hosts_with_logs': hosts_with_logs(components.get('tikv') or []),
        'tidb_hosts_with_logs': hosts_with_logs(components.get('tidb') or []),
        'slow_log_files': components.get('slow') or [],
        'slow_metric_count': len(components.get('slow_metrics') or []),
    }
    if bool(getattr(args, 'full_clinic_json', False)):
        result['payload'] = payload
    return result


def download_pingkai_attachment(client: TicketAPIClient, ref: dict, issue_ref: str, root: str = '') -> dict:
    directory = issue_attachment_dir(issue_ref, root=root)
    directory.mkdir(parents=True, exist_ok=True)
    url = str(ref.get('link') or '').strip()
    resp = client.session.get(url, stream=True, timeout=60)
    resp.raise_for_status()
    response_name = response_filename_from_headers(resp.headers)
    file_id = url.rstrip('/').rsplit('/', 1)[-1]
    preferred = sanitize_attachment_name(f'{file_id}_{response_name}' if response_name else file_id, fallback=file_id)
    path = directory / preferred
    with resp:
        with path.open('wb') as fh:
            for chunk in resp.iter_content(chunk_size=1024 * 1024):
                if chunk:
                    fh.write(chunk)
    mime_type = str(resp.headers.get('Content-Type') or mimetypes.guess_type(path.name)[0] or '')
    meta = {
        'provider': 'pingkai',
        'source': ref.get('source') or '',
        'author': ref.get('author') or '',
        'created': ref.get('created') or '',
        'download_url': url,
        'path': str(path),
        'filename': path.name,
        'mime_type': mime_type,
        'size': path.stat().st_size,
    }
    meta['analysis'] = analyze_downloaded_attachment(path, meta)
    return meta


def download_jira_attachment(ref: dict, issue_ref: str, root: str = '') -> dict:
    if not jira_api_available():
        return {
            'provider': 'jira',
            'download_url': ref.get('link') or '',
            'error': 'Jira API credentials are unavailable, so attachments cannot be downloaded',
        }
    directory = issue_attachment_dir(issue_ref, root=root)
    directory.mkdir(parents=True, exist_ok=True)
    auth = base64.b64encode(f'{jira_api_email()}:{jira_api_token()}'.encode('utf-8')).decode('ascii')
    attachment_url = validated_url(
        str(ref.get('link') or '').strip(),
        allowed_hosts=('tidb.atlassian.net',),
        path_prefixes=('/rest/api/3/attachment/content/', '/rest/api/2/attachment/content/'),
    )
    with requests.get(
        attachment_url,
        headers={
            'Authorization': f'Basic {auth}',
            'Accept': '*/*',
            'User-Agent': 'oncall-toolkit/1.0',
        },
        timeout=60,
        stream=True,
    ) as resp:
        resp.raise_for_status()
        response_name = response_filename_from_headers(resp.headers)
        attachment_id = str(ref.get('id') or '').strip()
        preferred = sanitize_attachment_name(
            f'{attachment_id}_{ref.get("filename") or response_name or "jira-attachment"}',
            fallback='jira-attachment',
        )
        path = directory / preferred
        with path.open('wb') as fh:
            for chunk in resp.iter_content(chunk_size=1024 * 1024):
                if not chunk:
                    continue
                fh.write(chunk)
        mime_type = str(resp.headers.get('Content-Type') or ref.get('mime_type') or mimetypes.guess_type(path.name)[0] or '')
    meta = {
        'provider': 'jira',
        'source': ref.get('source') or '',
        'author': ref.get('author') or '',
        'created': ref.get('created') or '',
        'download_url': ref.get('link') or '',
        'path': str(path),
        'filename': path.name,
        'mime_type': mime_type,
        'size': path.stat().st_size,
    }
    meta['analysis'] = analyze_downloaded_attachment(path, meta)
    return meta


def run_attachment_downloads(context: dict, args) -> list[dict]:
    attachment_refs = context.get('attachment_references') or []
    if not attachment_refs:
        return []
    root = str(getattr(args, 'attachment_download_root', '') or '').strip()
    issue_ref = str(context.get('issue_ref') or 'issue').strip()
    results = []
    pingkai_client = None
    for ref in attachment_refs:
        if should_skip_attachment_download(ref):
            continue
        provider = str(ref.get('provider') or '').strip()
        try:
            if provider == 'pingkai':
                if pingkai_client is None:
                    pingkai_client, _ = load_pingkai_ticket(issue_ref, include_internal=True)
                    if not pingkai_client:
                        raise RuntimeError('failed to initialize Pingkai client for attachment downloads')
                result = download_pingkai_attachment(pingkai_client, ref, issue_ref, root=root)
            elif provider == 'jira':
                result = download_jira_attachment(ref, issue_ref, root=root)
            else:
                result = {
                    'provider': provider or 'unknown',
                    'download_url': ref.get('link') or '',
                    'error': f'unsupported attachment provider: {provider}',
                }
        except Exception as exc:
            result = {
                'provider': provider or 'unknown',
                'download_url': ref.get('link') or '',
                'error': str(exc),
            }
        result['source'] = ref.get('source') or result.get('source') or ''
        result['author'] = ref.get('author') or result.get('author') or ''
        result['created'] = ref.get('created') or result.get('created') or ''
        results.append(result)
    return results


def build_preliminary_diagnosis(
    context: dict,
    clinic_results: list[dict],
    grafana_results: list[dict],
    bundle_results: list[dict],
    attachment_results: list[dict],
) -> str:
    parts = []
    blocking_error = first_blocking_evidence_error(clinic_results, bundle_results)
    if blocking_error:
        return blocking_error
    pending = [item for item in clinic_results if not item.get('error') and clinic_data_pending(item)]
    successful = [item for item in clinic_results if not item.get('error')]
    if pending:
        first_pending = pending[0]
        data_status = first_pending.get('data_status') or {}
        statuses = data_status.get('status_values') or []
        ready_status = data_status.get('ready_status')
        status_text = ", ".join(str(value) for value in statuses) if statuses else "no status item yet"
        parts.append(
            f'Clinic data is still preparing for this window: status {status_text}, ready at {ready_status}. '
            'Diagnosis is intentionally deferred until Clinic data is ready so ticket comments and attachments do not become substitute evidence.'
        )
        return ' '.join(parts)
    if successful:
        first = successful[0]
        dashboard = clinic_dashboard_title(first)
        metrics_overall = clinic_metrics_overall_judgement(first)
        judgement = str(metrics_overall.get('judgement') or '').strip()
        severity = str(metrics_overall.get('severity') or '').strip()
        data_status = first.get('data_status') or {}
        deep_rca = first.get('tikv_write_path_evidence') or {}
        deep_summary = str(deep_rca.get('summary') or '').strip()
        confidence = str(deep_rca.get('confidence') or '').strip()
        specialized = specialized_clinic_rca_summary(context, first)
        leader_drop = leader_drop_mechanism_summary(context, first)
        direct_mechanism = str(leader_drop.get('direct_mechanism') or '').strip()
        candidate_upstream = str(leader_drop.get('candidate_upstream_root_cause') or '').strip()
        ruled_out = leader_drop.get('ruled_out') or []
        earliest_signal = str(leader_drop.get('earliest_signal') or '').strip()
        if direct_mechanism:
            parts.append(f'Leader-drop direct mechanism: {direct_mechanism}')
        if earliest_signal:
            parts.append(earliest_signal)
        if ruled_out:
            parts.append(f'Not supported by current evidence: {ruled_out[0]}')
        if candidate_upstream:
            parts.append(f'Candidate upstream cause: {candidate_upstream}')
        if specialized:
            summary = str(specialized.get('summary') or '').strip()
            if summary:
                parts.append(summary)
            detail_lines = specialized.get('detail_lines') or []
            if detail_lines:
                parts.append(str(detail_lines[0]))
        if deep_summary:
            if confidence:
                parts.append(f'Clinic RCA summary ({confidence} confidence): {deep_summary}')
            else:
                parts.append(f'Clinic RCA summary: {deep_summary}')
        if severity or dashboard:
            parts.append(
                f'Clinic-based assessment'
                f'{f" on {dashboard}" if dashboard else ""}'
                f'{f", overall severity={severity}" if severity else ""}.'
            )
        if data_status and not data_status.get('ready'):
            statuses = data_status.get('status_values') or []
            ready_status = data_status.get('ready_status')
            status_text = ", ".join(str(value) for value in statuses) if statuses else "no status item yet"
            parts.append(f'Clinic data is still preparing for this window: status {status_text}, ready at {ready_status}.')
        if judgement and not deep_summary:
            parts.append(f'Clinic judgement: {judgement}')
        findings = first.get('headline_findings') or []
        if findings and not deep_summary:
            parts.append(f'Clinic headline findings: {"; ".join(str(x) for x in findings[:3])}')
    elif grafana_results:
        first = next((item for item in grafana_results if not item.get('error')), None)
        if first:
            cluster_summary = first.get('cluster_summary') or {}
            dashboard = grafana_dashboard_title(first)
            cluster_name = str(cluster_summary.get('cluster_name') or '').strip()
            qps_5m = cluster_summary.get('qps_5m')
            parts.append(
                f'Grafana-based assessment'
                f'{f" on {dashboard}" if dashboard else ""}'
                f'{f" for cluster {cluster_name}" if cluster_name else ""}.'
            )
            if qps_5m is not None:
                parts.append(f'Current 5m QPS is {float(qps_5m):.2f}.')
    else:
        phenomenon = str(context.get('phenomenon') or '').strip()
        if phenomenon and phenomenon != 'TBD':
            parts.append(f'Ticket-reported phenomenon: {phenomenon}')

        clinic_refs = context.get('clinic_references') or []
        if clinic_refs:
            parts.append('A Clinic link was found, but no successful Clinic analysis is available yet, so ticket comments should be treated only as clues.')
        else:
            parts.append('No Clinic link was found, so this is a clue-based triage summary from the ticket only.')

    successful_bundles = [item for item in bundle_results if not item.get('error')]
    if successful_bundles:
        first_bundle = successful_bundles[0]
        tidb_version = str(first_bundle.get('tidb_version') or '').strip()
        extract_dir = str(first_bundle.get('extract_dir') or '').strip()
        if tidb_version:
            parts.append(f'Clinic bundle metadata confirms cluster version {tidb_version}.')
        if extract_dir:
            parts.append(f'Clinic bundle artifacts were downloaded locally under {extract_dir}.')

    successful_attachments = [item for item in attachment_results if not item.get('error')]
    if successful_attachments:
        attachment_dir = str(Path(successful_attachments[0].get('path') or '').parent)
        parts.append(
            f'Issue attachments were downloaded locally under '
            f'{attachment_dir}.'
        )
        slow_log_hits = [
            item for item in successful_attachments
            if ((item.get('analysis') or {}).get('slow_log_summary') or {}).get('detected')
        ]
        text_hits = [item for item in successful_attachments if ((item.get('analysis') or {}).get('kind') == 'text')]
        profile_hits = [item for item in successful_attachments if ((item.get('analysis') or {}).get('kind') == 'profile')]
        if slow_log_hits:
            first_slow_log = ((slow_log_hits[0].get('analysis') or {}).get('slow_log_summary') or {})
            summary_lines = first_slow_log.get('summary_lines') or []
            parts.append(f'{len(slow_log_hits)} slow-log attachment(s) were parsed into structured stage and time-window evidence.')
            if summary_lines:
                parts.append(summary_lines[0])
        if text_hits:
            parts.append(f'{len(text_hits)} text-like attachment(s) were analyzed for log or config clues.')
        if profile_hits:
            parts.append(f'{len(profile_hits)} profile-like attachment(s) were downloaded for deeper follow-up.')

    skills = context.get('recommended_skills') or []
    if skills:
        parts.append(f'Recommended diagnosis skills: {", ".join(skills)}')

    return ' '.join(parts) or 'No diagnosis signal was extracted from the ticket yet.'


def build_attachment_analysis_tracks(attachment_results: list[dict]) -> list[dict]:
    tracks = []
    for item in attachment_results or []:
        analysis = (item or {}).get('analysis') or {}
        slow_log_summary = analysis.get('slow_log_summary') or {}
        if not slow_log_summary:
            continue
        windows = slow_log_summary.get('windows') or []
        top_stages = slow_log_summary.get('top_stages') or []
        dominant_store = str(slow_log_summary.get('dominant_store') or '').strip()
        if not windows:
            windows = [slow_log_summary.get('time_range') or {}]
        for index, window in enumerate(windows, start=1):
            top_stage = top_stages[0] if top_stages else {}
            stage_name = str(top_stage.get('name') or '').strip()
            stage_max = format_seconds(top_stage.get('max_seconds'))
            summary = []
            if stage_name:
                summary.append(f'Slow log shows `{stage_name}` as the dominant tail stage (max {stage_max}).')
            if dominant_store:
                summary.append(f'The most repeated store hint is `{dominant_store}`.')
            tracks.append(
                {
                    'source': 'slow_log_attachment',
                    'label': f'{item.get("filename") or "slow-log"}#{index}',
                    'from': window.get('from') or (slow_log_summary.get('time_range') or {}).get('from') or 'N/A',
                    'to': window.get('to') or (slow_log_summary.get('time_range') or {}).get('to') or 'N/A',
                    'sample_count': int(window.get('sample_count') or slow_log_summary.get('row_count') or 0),
                    'dominant_stage': stage_name,
                    'dominant_store': dominant_store,
                    'summary': ' '.join(summary).strip(),
                    'focus_hints': list(slow_log_summary.get('focus_hints') or []),
                }
            )
    return tracks


def build_clinic_analysis_tracks(clinic_results: list[dict]) -> list[dict]:
    tracks = []
    for index, item in enumerate(clinic_results or [], start=1):
        if item.get('error'):
            continue
        deep_rca = item.get('tikv_write_path_evidence') or {}
        incident_window = deep_rca.get('incident_window') or {}
        primary_entity = deep_rca.get('primary_entity') or {}
        summary = str(deep_rca.get('summary') or '').strip()
        if not (incident_window or summary):
            continue
        dominant_store = str(primary_entity.get('instance') or primary_entity.get('store') or '').strip()
        tracks.append(
            {
                'source': 'clinic',
                'label': f'clinic#{index}',
                'from': incident_window.get('from') or 'N/A',
                'to': incident_window.get('to') or 'N/A',
                'sample_count': 0,
                'dominant_stage': '',
                'dominant_store': dominant_store,
                'summary': summary,
                'focus_hints': [],
            }
        )
    return tracks


def merge_analysis_tracks(attachment_tracks: list[dict], clinic_tracks: list[dict]) -> list[dict]:
    tracks = []
    tracks.extend(attachment_tracks or [])
    tracks.extend(clinic_tracks or [])
    tracks.sort(key=lambda item: (str(item.get('from') or ''), str(item.get('label') or '')))
    return tracks


def clinic_dashboard_title(item: dict) -> str:
    dashboard = ((item.get('dashboard_discovery') or {}).get('dashboard') or {})
    return str(dashboard.get('title') or '').strip()


def grafana_dashboard_title(item: dict) -> str:
    dashboard = ((item.get('dashboard_discovery') or {}).get('dashboard') or {})
    return str(dashboard.get('title') or '').strip()


def clinic_metrics_overall_judgement(item: dict) -> dict:
    return ((item.get('metrics_evidence') or {}).get('overall_judgement')) or {}


def clinic_headline_findings(item: dict) -> list[str]:
    overview_items = (((item.get('metrics_evidence') or {}).get('performance_overview_summary') or {}).get('items')) or []
    return [str(entry.get('judgement') or '').strip() for entry in overview_items[:3] if str(entry.get('judgement') or '').strip()]


def clinic_follow_up_steps(item: dict) -> list[str]:
    steps = []
    metrics_overall = clinic_metrics_overall_judgement(item)
    severity = str(metrics_overall.get('severity') or '').strip()
    judgement = str(metrics_overall.get('judgement') or '').strip()
    if judgement:
        steps.append(f'Use metrics evidence as the baseline assessment: {judgement}')
    if severity in {'high', 'medium'}:
        steps.append('Correlate the abnormal metric modules with ticket symptoms before drawing a final RCA conclusion.')
    access_warnings = item.get('access_warnings') or []
    steps.extend(str(warning) for warning in access_warnings[:1] if str(warning).strip())
    return steps


def is_critical_evidence_missing_error(message: str) -> bool:
    return str(message or '').strip().lower().startswith(CRITICAL_EVIDENCE_MISSING_PREFIX.lower())


def first_blocking_evidence_error(*result_groups: list[dict]) -> str:
    for group in result_groups:
        for item in group or []:
            error = str((item or {}).get('error') or '').strip()
            if is_critical_evidence_missing_error(error):
                return error
    return ''


def clinic_data_pending(item: dict) -> bool:
    data_status = item.get('data_status') or {}
    return bool(data_status) and not bool(data_status.get('ready'))


def clinic_data_ready(item: dict) -> bool:
    data_status = item.get('data_status') or {}
    if data_status:
        return bool(data_status.get('ready'))
    return not bool(item.get('error'))


def context_text_blob(context: dict) -> str:
    return '\n'.join(
        str(context.get(key) or '')
        for key in ['title', 'problem_statement', 'phenomenon', 'routing_context_text']
    )


def context_is_leader_drop(context: dict) -> bool:
    text = context_text_blob(context)
    return bool(
        re.search(
            r'leader drop|leader missing|re-?election|掉leader|掉主|leader丢失|选主',
            text,
            re.I,
        )
    )


def context_is_network_issue(context: dict) -> bool:
    text = context_text_blob(context)
    return bool(
        re.search(
            r'tikvrpc|batch receive|wait connection establish|rpc layer latency|unreachable|'
            r'network jitter|network latency|network interruption|network saturation|'
            r'带宽打满|网络抖动|网络异常|网络问题',
            text,
            re.I,
        )
    )


def context_is_flow_control(context: dict) -> bool:
    text = context_text_blob(context)
    return bool(
        re.search(
            r'serverisbusy|server is busy|deadline exceeded|flow control|throttle duration|'
            r'discard ratio|scheduler pending write|pending compaction|l0 file|memtable|'
            r'memory_quota_exceeded|max_pending_tasks_exceeded|kv backoff|channelfull|'
            r'raftstore is busy|限流|流控',
            text,
            re.I,
        )
    )


def clinic_basic_check_entry(item: dict, title: str) -> dict:
    entries = ((((item.get('metrics_evidence') or {}).get('basic_healthy_check')) or {}).get('items')) or []
    want = str(title or '').strip()
    for entry in entries:
        if str((entry or {}).get('check_item') or '').strip() == want:
            return entry
    return {}


def clinic_pd_slow_store_total(item: dict) -> float | None:
    entry = clinic_basic_check_entry(item, 'PD -> Abnormal stores -> Slow stores')
    observed = (entry or {}).get('observed') or {}
    value = observed.get('total')
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def format_unix_ts_local(ts) -> str:
    try:
        value = int(ts)
    except (TypeError, ValueError):
        return 'N/A'
    return datetime.fromtimestamp(value, tz=timezone.utc).astimezone().strftime('%Y-%m-%d %H:%M:%S %z')


def format_seconds(value: float | None) -> str:
    if value is None:
        return 'N/A'
    if value >= 1:
        return f'{value:.2f}s'
    if value >= 0.001:
        return f'{value * 1000:.1f}ms'
    return f'{value * 1000_000:.0f}us'


def format_number(value: float | None, digits: int = 2) -> str:
    if value is None:
        return 'N/A'
    return f'{value:.{digits}f}'


def clinic_deep_metric_peak(item: dict, metric_name: str) -> dict:
    metric = ((((item.get('tikv_write_path_evidence') or {}).get('raw_metrics')) or {}).get(metric_name)) or {}
    return (metric.get('peak') or {}) if isinstance(metric, dict) else {}


def clinic_deep_metric_peak_value(item: dict, metric_name: str) -> float | None:
    peak = clinic_deep_metric_peak(item, metric_name)
    try:
        return float(peak.get('peak_value'))
    except (TypeError, ValueError):
        return None


def clinic_server_is_busy_groups(item: dict) -> list[dict]:
    entry = clinic_basic_check_entry(item, 'TiKV -> Errors -> Server is busy')
    observed = (entry or {}).get('observed') or []
    groups = []
    for group in observed:
        try:
            series_count = int(group.get('active_series_count') or 0)
        except (TypeError, ValueError):
            series_count = 0
        if series_count <= 0:
            continue
        top5 = group.get('top5') or []
        value_sum = 0.0
        instances = []
        for item_ in top5:
            try:
                value_sum += float(item_.get('last_value') or 0)
            except (TypeError, ValueError):
                pass
            instance = str(item_.get('instance') or '').strip()
            if instance and instance not in instances:
                instances.append(instance)
        groups.append(
            {
                'name': str(group.get('name') or '').strip(),
                'active_series_count': series_count,
                'value_sum': value_sum,
                'instances': instances[:3],
            }
        )
    groups.sort(key=lambda entry: (entry['value_sum'], entry['active_series_count']), reverse=True)
    return groups


def network_issue_summary(context: dict, item: dict) -> dict:
    if not context_is_network_issue(context):
        return {}

    deep_rca = item.get('tikv_write_path_evidence') or {}
    primary_entity = deep_rca.get('primary_entity') or {}
    instance = str(primary_entity.get('instance') or '').strip()
    store = str(primary_entity.get('store') or '').strip()
    rpc_peak = clinic_deep_metric_peak_value(item, 'rpc_layer_latency')
    commit_peak = clinic_deep_metric_peak_value(item, 'commit_log')
    append_peak = clinic_deep_metric_peak_value(item, 'append_log')
    process_ready_peak = clinic_deep_metric_peak_value(item, 'process_ready')
    grpc_peak = clinic_deep_metric_peak_value(item, 'grpc_poll_cpu')

    failure_entry = clinic_basic_check_entry(item, 'TiKV -> Errors -> Server report failures')
    failure_observed = (failure_entry or {}).get('observed') or {}
    try:
        failure_count = int(failure_observed.get('failure_series_count') or 0)
    except (TypeError, ValueError):
        failure_count = 0
    failure_reasons = []
    for series in (failure_observed.get('top5') or [])[:3]:
        metric = series.get('metric') or {}
        reason = str(metric.get('type') or metric.get('reason') or metric.get('result') or '').strip()
        if reason and reason not in failure_reasons:
            failure_reasons.append(reason)

    affected_hop = 'TiDB <-> TiKV'
    issue_type = 'latency jitter'
    direct_mechanism = 'Transport latency between TiDB and TiKV is more likely than TiKV-local append/apply inflation.'
    if (
        failure_count > 0
        and commit_peak is not None
        and append_peak is not None
        and commit_peak > max(append_peak * 3, append_peak + 0.03)
    ):
        affected_hop = 'TiKV <-> TiKV'
        issue_type = 'transport instability'
        direct_mechanism = (
            'Raft replication transport or follower-response delay is more likely than a local append/apply bottleneck.'
        )
    elif failure_count > 0:
        affected_hop = 'TiKV <-> TiKV'
        issue_type = 'interruption or severe jitter'
        direct_mechanism = (
            'Peer communication instability is visible, but it still needs to be separated from gRPC-thread false positives.'
        )

    evidence = []
    if rpc_peak is not None:
        evidence.append(
            f'TiDB RPC layer latency peaked at {format_seconds(rpc_peak)}'
            f'{f" on store {store}" if store else ""}, so transport time was visible before full TiKV processing.'
        )
    if commit_peak is not None and append_peak is not None and commit_peak > append_peak:
        evidence.append(
            f'Commit log duration peaked at {format_seconds(commit_peak)} while append log was only {format_seconds(append_peak)}, '
            'which fits transport or follower-response delay better than local append saturation.'
        )
    if failure_count > 0:
        reason_text = f' ({", ".join(failure_reasons)})' if failure_reasons else ''
        evidence.append(f'TiKV server-report failures had {failure_count} active series in the same window{reason_text}.')
    if not evidence and deep_rca.get('summary'):
        evidence.append(str(deep_rca.get('summary') or '').strip())

    counterevidence = []
    if append_peak is not None and process_ready_peak is not None:
        counterevidence.append(
            f'Append log stayed at {format_seconds(append_peak)} and process-ready at {format_seconds(process_ready_peak)}, '
            'so the Clinic window does not support a broad local append/process-ready bottleneck.'
        )
    if grpc_peak is not None and grpc_peak < 1:
        counterevidence.append(
            f'gRPC poll CPU peak was {format_number(grpc_peak)}, which is not a comparably large saturation signal in the collected peaks.'
        )

    not_supported = []
    if failure_count == 0:
        not_supported.append(
            'A hard TiKV-to-TiKV network interruption is not proven because the collected Clinic summary does not show active server-report failures.'
        )
    if grpc_peak is not None and grpc_peak >= 1:
        not_supported.append(
            '`unreachable` or transport delay should not be claimed as a pure network fault without checking gRPC-thread false positives in logs or host evidence.'
        )

    actions = [
        'Correlate TiDB slow logs, batch-client reconnect metrics, and RPC layer latency before calling the issue a TiKV-local slowdown.',
        'Check node-level network allowance, retransmit, or drop metrics on the affected nodes in the same narrow window.',
        'If the incident is on versions before v8.5.1 and clearly matches TiDB <-> TiKV jitter, consider whether a lower `tikv-client.grpc-keepalive-timeout` is an appropriate temporary mitigation.',
    ]

    summary = (
        f'Current Clinic evidence most strongly points to {issue_type} on the {affected_hop} hop. '
        f'{direct_mechanism}'
    )

    return {
        'heading': '### Network RCA',
        'summary': summary,
        'detail_lines': [
            f'Affected hop: {affected_hop}',
            f'Network symptom type: {issue_type}',
            f'Direct mechanism: {direct_mechanism}',
        ],
        'evidence': evidence,
        'counterevidence': counterevidence,
        'not_supported': not_supported,
        'recommended_actions': actions,
    }


def flow_control_summary(context: dict, item: dict) -> dict:
    if not context_is_flow_control(context):
        return {}

    groups = clinic_server_is_busy_groups(item)
    deep_rca = item.get('tikv_write_path_evidence') or {}
    before_write_peak = clinic_deep_metric_peak_value(item, 'before_write')
    propose_peak = clinic_deep_metric_peak_value(item, 'propose_wait')
    commit_persist_peak = clinic_deep_metric_peak_value(item, 'commit_and_persist')
    append_peak = clinic_deep_metric_peak_value(item, 'append_log')
    process_ready_peak = clinic_deep_metric_peak_value(item, 'process_ready')
    write_stall_peak = clinic_deep_metric_peak_value(item, 'write_stall')

    dominant = groups[0] if groups else {}
    dominant_name = str(dominant.get('name') or '').strip()
    mechanism_map = {
        'scheduler_too_busy': (
            'Write-side scheduler admission control fired.',
            'write',
            'scheduler pending-write pressure or a short burst that effectively crossed the threshold',
            'scheduler entry pressure or a downstream write-path backlog',
        ),
        'channel_full': (
            'Write-side raftstore channel-full protection fired.',
            'write',
            'raftstore router/channel saturation',
            'raftstore processing backlog or a downstream write-path stall',
        ),
        'store_write_block_wait': (
            'Write-side store-write blocking protection fired.',
            'write',
            'store-write block wait or async-write backlog',
            'the write / persist path could not drain fast enough',
        ),
        'raftstore_process_busy': (
            'Write-side raftstore process-busy protection fired.',
            'write',
            'raftstore processing backlog',
            'raftstore CPU pressure or downstream raft processing delay',
        ),
        'coprocessor_full': (
            'Read-side coprocessor or unified-read-pool protection fired.',
            'read',
            'read-pool or coprocessor admission pressure',
            'read-pool queueing or slow coprocessor tasks',
        ),
    }
    direct_mechanism, affected_path, trigger_text, bottleneck_text = mechanism_map.get(
        dominant_name,
        (
            'Request protection fired before the affected requests could complete.',
            'write or read',
            'a flow-control or deadline budget threshold',
            'a lower-layer bottleneck that still needs to be pinned down',
        ),
    )

    evidence = []
    if groups:
        group_bits = []
        for group in groups[:3]:
            group_bits.append(
                f'{group["name"]} ({group["active_series_count"]} active series'
                f'{f", top instances: {", ".join(group["instances"])}" if group.get("instances") else ""})'
            )
        evidence.append('`TiKV -> Errors -> Server is busy` was active in the same window: ' + '; '.join(group_bits) + '.')
    if before_write_peak is not None:
        evidence.append(
            f'Store before write peaked at {format_seconds(before_write_peak)}, which means the write path was already backed up before the final persist stage.'
        )
    if propose_peak is not None:
        evidence.append(
            f'Store propose wait peaked at {format_seconds(propose_peak)}, which supports queueing before raft processing completed.'
        )
    if commit_persist_peak is not None:
        evidence.append(
            f'Store commit and persist peaked at {format_seconds(commit_persist_peak)}, so the protected requests were not just rejected at the API edge; the downstream write path was also slow.'
        )
    if not evidence and deep_rca.get('summary'):
        evidence.append(str(deep_rca.get('summary') or '').strip())

    counterevidence = []
    if append_peak is not None and process_ready_peak is not None:
        counterevidence.append(
            f'Append log stayed at {format_seconds(append_peak)} and process-ready at {format_seconds(process_ready_peak)}, '
            'so the Clinic window does not support blaming a broad append/process-ready bottleneck by default.'
        )
    if write_stall_peak is not None:
        counterevidence.append(
            f'Write stall remained at {format_number(write_stall_peak)}, so sustained write-stall state is not proven from this window alone.'
        )

    not_supported = [
        '`ServerIsBusy` itself is a protection signal, not the final root cause.',
        'Threshold tuning alone should not be treated as the RCA unless the downstream bottleneck is also explained.',
    ]

    actions = []
    if dominant_name == 'scheduler_too_busy':
        actions.append('Check scheduler writing bytes and raft-engine written bytes with higher quantiles to distinguish a real threshold crossing from a short burst that a gauge may miss.')
    elif dominant_name in {'channel_full', 'raftstore_process_busy', 'store_write_block_wait'}:
        actions.append('Treat raftstore and downstream write-path stages as the next bottleneck split instead of stopping at the `ServerIsBusy` symptom.')
    elif dominant_name == 'coprocessor_full':
        actions.append('Check unified read-pool wait, running tasks, and slow query shape before changing read-pool thresholds.')
    else:
        actions.append('Use TiDB or TiKV logs to capture the exact `reason` field if the Clinic summary alone does not fully pin down the threshold family.')
    actions.append('When proposing threshold changes, state clearly whether they are emergency mitigation only or part of a proven root-cause fix.')

    summary = (
        f'Current Clinic evidence most strongly points to a {affected_path}-side flow-control or admission-protection event. '
        f'{direct_mechanism} The more important RCA question is which downstream bottleneck triggered that protection.'
    )

    return {
        'heading': '### Flow Control RCA',
        'summary': summary,
        'detail_lines': [
            f'Affected path: {affected_path}',
            f'Direct mechanism: {direct_mechanism}',
            f'Observed trigger: {trigger_text}',
            f'Candidate downstream bottleneck: {bottleneck_text}',
        ],
        'evidence': evidence,
        'counterevidence': counterevidence,
        'not_supported': not_supported,
        'recommended_actions': actions,
    }


def specialized_clinic_rca_summary(context: dict, item: dict) -> dict:
    if context_is_leader_drop(context):
        return {}
    if context_is_flow_control(context):
        return flow_control_summary(context, item)
    if context_is_network_issue(context):
        return network_issue_summary(context, item)
    return {}


def leader_drop_mechanism_summary(context: dict, item: dict) -> dict:
    if not context_is_leader_drop(context):
        return {}

    pd_slow_total = clinic_pd_slow_store_total(item)
    deep_rca = item.get('tikv_write_path_evidence') or {}
    primary_entity = deep_rca.get('primary_entity') or {}
    instance = str(primary_entity.get('instance') or '').strip()
    store = str(primary_entity.get('store') or '').strip()

    direct_mechanism = (
        'Follower-side re-election after delayed heartbeat or delayed raft processing is more likely than '
        'PD-driven evict-slow-store.'
    )
    if pd_slow_total and pd_slow_total > 0:
        direct_mechanism = (
            'PD-driven evict-slow-store is not proven by this workflow. The collected Clinic evidence still '
            'looks more like follower-side re-election after delayed heartbeat or delayed raft processing, '
            'but PD scheduler/operator evidence would still be required before naming a PD-driven mechanism.'
        )

    ruled_out = []
    if pd_slow_total == 0:
        ruled_out.append('PD abnormal-store checks reported 0 slow stores in the same window.')
        ruled_out.append('No PD scheduler or operator evidence for evict-slow-store was collected in this workflow.')
    elif pd_slow_total is not None:
        ruled_out.append(
            f'PD slow-store count was {pd_slow_total:.0f}, but slow-store count alone is not enough to prove evict-slow-store.'
        )
        ruled_out.append('PD scheduler or operator evidence would still be required before naming evict-slow-store.')
    else:
        ruled_out.append('PD slow-store count was not available in the collected Clinic summary, so evict-slow-store remains unproven.')

    candidates = []
    thresholds = {
        'rpc_layer_latency': 0.05,
        'before_write': 0.05,
        'propose_wait': 0.05,
        'commit_and_persist': 0.05,
        'commit_log': 0.05,
        'grpc_poll_cpu': 1.0,
    }
    labels = {
        'rpc_layer_latency': 'TiDB RPC layer latency',
        'before_write': 'Store before write',
        'propose_wait': 'Store propose wait',
        'commit_and_persist': 'Store commit and persist',
        'commit_log': 'TiKV commit log duration',
        'grpc_poll_cpu': 'gRPC poll CPU',
    }
    for metric_name, min_value in thresholds.items():
        peak = clinic_deep_metric_peak(item, metric_name)
        try:
            peak_value = float(peak.get('peak_value'))
            peak_ts = int(peak.get('peak_ts'))
        except (TypeError, ValueError):
            continue
        if peak_value < min_value:
            continue
        candidates.append((peak_ts, labels[metric_name], peak_value, metric_name))
    candidates.sort(key=lambda entry: entry[0])

    earliest_signal = ''
    if candidates:
        first_ts = candidates[0][0]
        first_items = [entry for entry in candidates if entry[0] == first_ts]
        signal_bits = []
        for _, label, peak_value, metric_name in first_items:
            if metric_name == 'grpc_poll_cpu':
                signal_bits.append(f'{label} reached {peak_value:.2f}')
            else:
                signal_bits.append(f'{label} reached {peak_value:.2f}s')
        earliest_signal = (
            f'The earliest strong signal in the collected 15s-aligned peaks was at '
            f'{format_unix_ts_local(first_ts)}: ' + '; '.join(signal_bits) + '.'
        )

    candidate_upstream = str(deep_rca.get('summary') or '').strip()
    if instance:
        candidate_upstream = (
            f'The strongest upstream candidate remains a localized node stall on {instance}'
            f'{f" (store {store})" if store else ""}, which slowed the raft/write path enough to delay heartbeats or raft processing.'
        )

    return {
        'direct_mechanism': direct_mechanism,
        'earliest_signal': earliest_signal,
        'ruled_out': ruled_out,
        'candidate_upstream_root_cause': candidate_upstream,
    }


def grafana_follow_up_steps(item: dict) -> list[str]:
    cluster_summary = item.get('cluster_summary') or {}
    steps = []
    cluster_name = str(cluster_summary.get('cluster_name') or '').strip()
    if cluster_name:
        steps.append(f'Use the Grafana cluster summary for {cluster_name} as monitoring context before diagnosis.')
    else:
        steps.append('Use the Grafana cluster summary as monitoring context before diagnosis.')
    access_warnings = item.get('access_warnings') or []
    steps.extend(str(warning) for warning in access_warnings[:1] if str(warning).strip())
    return steps


def build_analyze_payload(
    context: dict,
    clinic_refs: list[dict],
    selected_refs: list[dict],
    clinic_results: list[dict],
    grafana_results: list[dict],
    clinic_download_refs: list[dict],
    selected_download_refs: list[dict],
    bundle_results: list[dict],
    attachment_refs: list[dict],
    attachment_results: list[dict],
) -> dict:
    fatal_error = first_blocking_evidence_error(clinic_results, bundle_results)
    missing_evidence_reasons = analyze_missing_evidence_reasons(context, clinic_results)
    attachment_tracks = build_attachment_analysis_tracks(attachment_results)
    clinic_tracks = build_clinic_analysis_tracks(clinic_results)
    attachment_policy = context.get('attachment_policy') or {}
    diagnosis_result = build_diagnosis_result(
        context,
        clinic_results=clinic_results,
        grafana_results=grafana_results,
        bundle_results=bundle_results,
        attachment_results=attachment_results,
        missing_evidence_reasons=missing_evidence_reasons,
        context_is_leader_drop=context_is_leader_drop,
        context_is_network_issue=context_is_network_issue,
        context_is_flow_control=context_is_flow_control,
    ).to_dict()
    return {
        'mode': 'analyze',
        'issue_ref': context.get('issue_ref'),
        'source': context.get('source'),
        'title': context.get('title'),
        'ticket_url': context.get('ticket_url'),
        'problem_statement': context.get('problem_statement'),
        'fatal_error': fatal_error,
        'diagnosis_summary': build_preliminary_diagnosis(context, clinic_results, grafana_results, bundle_results, attachment_results),
        'ticket_context': {
            'phenomenon': context.get('phenomenon'),
            'owner_conclusion': context.get('owner_conclusion') or {},
            'recent_comments': context.get('recent_comments') or [],
            'comment_corpus': context.get('comment_corpus') or [],
            'clinic_references': clinic_refs,
            'grafana_references': context.get('grafana_references') or [],
            'clinic_download_references': clinic_download_refs,
            'attachment_references': attachment_refs,
        },
        'diagnosis': {
            'recommended_skills': context.get('recommended_skills') or [],
            'routing_reasons': context.get('routing_reasons') or [],
            'diagnosis_result': diagnosis_result,
            'missing_evidence_reasons': missing_evidence_reasons,
            'attachment_policy': attachment_policy,
            'selected_clinic_links': [item.get('link') for item in selected_refs],
            'selected_grafana_links': [item.get('link') for item in (context.get('grafana_references') or [])[:1]],
            'selected_clinic_download_links': [item.get('link') for item in selected_download_refs],
            'clinic_analysis': clinic_results,
            'grafana_analysis': grafana_results,
            'clinic_bundle_analysis': bundle_results,
            'attachment_analysis': attachment_results,
            'analysis_tracks': merge_analysis_tracks(attachment_tracks, clinic_tracks),
        },
    }


def markdown_bullet_lines(items: list[str]) -> list[str]:
    return [f'- {item}' for item in items if str(item or '').strip()]


def render_cluster_summary_lines(cluster_summary: dict) -> list[str]:
    if not cluster_summary:
        return []
    lines = ['### Cluster Summary']
    cluster_name = str(cluster_summary.get('cluster_name') or '').strip()
    cluster_version = str(cluster_summary.get('cluster_version') or '').strip()
    if cluster_name or cluster_version:
        label = cluster_name or 'N/A'
        if cluster_version:
            label = f'{label} ({cluster_version})'
        lines.append(f'- Cluster: {label}')

    for key, label in [
        ('store_count', 'TiKV stores'),
        ('tidb_count', 'TiDB nodes'),
        ('pd_count', 'PD nodes'),
        ('region_count', 'Region count'),
        ('leader_count', 'Leader count'),
    ]:
        value = cluster_summary.get(key)
        if value is not None:
            lines.append(f'- {label}: {int(value)}')

    for key, label in [
        ('store_used_bytes', 'Data size used'),
        ('store_available_bytes', 'Available size'),
        ('store_capacity_bytes', 'Capacity size'),
    ]:
        value = cluster_summary.get(key)
        if value is not None:
            gib = float(value) / (1024 ** 3)
            lines.append(f'- {label}: {float(value):.0f} bytes ({gib:.2f} GiB)')

    qps_1m = cluster_summary.get('qps_1m')
    qps_5m = cluster_summary.get('qps_5m')
    if qps_1m is not None:
        lines.append(f'- QPS (1m): {float(qps_1m):.2f}')
    if qps_5m is not None:
        lines.append(f'- QPS (5m): {float(qps_5m):.2f}')

    stores = cluster_summary.get('stores_used_by_instance') or []
    if stores:
        summaries = []
        for item in stores[:6]:
            value = item.get('value')
            if value is None:
                continue
            summaries.append(f'{item.get("instance")}: {float(value):.0f} bytes ({float(value)/(1024**3):.2f} GiB)')
        if summaries:
            lines.append(f'- TiKV store used size by instance: {"; ".join(summaries)}')

    sql_types = cluster_summary.get('top_sql_types_5m') or []
    if sql_types:
        summaries = [f'{item.get("type")}: {float(item.get("value") or 0):.2f}' for item in sql_types[:5]]
        if summaries:
            lines.append(f'- Top SQL types (5m): {"; ".join(summaries)}')

    commands = cluster_summary.get('top_tikv_commands_5m') or []
    if commands:
        summaries = [f'{item.get("type")}: {float(item.get("value") or 0):.2f}' for item in commands[:5]]
        if summaries:
            lines.append(f'- Top TiKV commands (5m): {"; ".join(summaries)}')
    if len(lines) == 1:
        lines.append('- No cluster summary metrics were returned for the requested window or datasource query set.')
    return lines


def render_analyze_markdown(payload: dict) -> str:
    lines = []
    title = str(payload.get('title') or payload.get('issue_ref') or 'N/A').strip()
    issue_ref = str(payload.get('issue_ref') or '').strip()
    ticket_url = str(payload.get('ticket_url') or '').strip()

    if ticket_url and issue_ref:
        lines.append(f'## Analyze {issue_ref}')
        lines.append(f'[{title}]({ticket_url})')
    elif issue_ref:
        lines.append(f'## Analyze {issue_ref}')
        lines.append(title)
    else:
        lines.append('## Analyze')
        lines.append(title)

    lines.append('')
    lines.append('### Summary')
    lines.append(str(payload.get('diagnosis_summary') or 'No diagnosis signal was extracted from the ticket yet.'))

    problem_statement = str(payload.get('problem_statement') or '').strip()
    if problem_statement:
        lines.append('')
        lines.append('### Problem Statement')
        lines.append(problem_statement)

    ticket_context = payload.get('ticket_context') or {}
    diagnosis = payload.get('diagnosis') or {}
    diagnosis_result = diagnosis.get('diagnosis_result') or {}
    phenomenon = str(ticket_context.get('phenomenon') or '').strip()
    owner_conclusion = ticket_context.get('owner_conclusion') or {}
    recent_comments = ticket_context.get('recent_comments') or []
    comment_corpus = ticket_context.get('comment_corpus') or []
    clinic_results = diagnosis.get('clinic_analysis') or []
    grafana_results = diagnosis.get('grafana_analysis') or []
    bundle_results = diagnosis.get('clinic_bundle_analysis') or []
    attachment_results = diagnosis.get('attachment_analysis') or []
    attachment_policy = diagnosis.get('attachment_policy') or {}
    analysis_tracks = diagnosis.get('analysis_tracks') or []
    clinic_pending = any(clinic_data_pending(item) for item in clinic_results if not item.get('error'))
    clinic_ready = any(clinic_data_ready(item) for item in clinic_results if not item.get('error'))
    analysis_context = {
        'title': title,
        'problem_statement': problem_statement,
        'phenomenon': phenomenon,
        'routing_context_text': problem_statement,
    }
    leader_drop_mode = context_is_leader_drop(analysis_context)

    if clinic_results:
        lines.append('')
        lines.append('### Clinic-Based Assessment')
        for item in clinic_results:
            link = str(item.get('link') or '').strip()
            if item.get('error'):
                lines.append(f'- Clinic analysis failed for [{link}]({link}): {item["error"]}')
                continue
            cluster_summary = item.get('cluster_summary') or {}
            data_status = item.get('data_status') or {}
            deep_rca = item.get('tikv_write_path_evidence') or {}
            metrics_overall = clinic_metrics_overall_judgement(item)
            severity = str(metrics_overall.get('severity') or 'N/A').strip()
            dashboard = clinic_dashboard_title(item) or 'N/A'
            judgement = str(metrics_overall.get('judgement') or 'N/A').strip()
            specialized = specialized_clinic_rca_summary(analysis_context, item)
            lines.append(f'- Source: [{link}]({link})')
            cluster_summary_lines = render_cluster_summary_lines(cluster_summary)
            if cluster_summary_lines:
                lines.append('')
                lines.extend(cluster_summary_lines)
                lines.append('')
            lines.append(f'- Dashboard: {dashboard}')
            lines.append(f'- Overall severity: {severity}')
            lines.append(f'- Clinic judgement: {judgement}')
            if data_status and not data_status.get('ready'):
                statuses = data_status.get('status_values') or []
                ready_status = data_status.get('ready_status')
                status_text = ', '.join(str(value) for value in statuses) if statuses else 'no status item yet'
                lines.append(f'- Clinic data status: {status_text} (ready when status reaches {ready_status})')
                lines.append('- Diagnosis state: pending Clinic readiness; do not treat ticket comments or attachments as final evidence yet.')
            deep_summary = str(deep_rca.get('summary') or '').strip()
            if deep_summary and not specialized:
                lines.append(f'- RCA summary: {deep_summary}')
            confidence = str(deep_rca.get('confidence') or '').strip()
            if confidence and not specialized:
                lines.append(f'- RCA confidence: {confidence}')

            leader_drop = leader_drop_mechanism_summary(analysis_context, item)
            if leader_drop_mode and leader_drop:
                lines.append('')
                lines.append('### Leader Drop Mechanism')
                direct_mechanism = str(leader_drop.get('direct_mechanism') or '').strip()
                earliest_signal = str(leader_drop.get('earliest_signal') or '').strip()
                candidate_upstream = str(leader_drop.get('candidate_upstream_root_cause') or '').strip()
                if direct_mechanism:
                    lines.append(f'- Direct mechanism: {direct_mechanism}')
                if earliest_signal:
                    lines.append(f'- Earliest collected signal: {earliest_signal}')
                ruled_out = leader_drop.get('ruled_out') or []
                if ruled_out:
                    lines.append(f'- Why PD evict-slow-store is not supported: {ruled_out[0]}')
                    for detail in ruled_out[1:3]:
                        lines.append(f'- Additional rule-out note: {detail}')
                if candidate_upstream:
                    lines.append(f'- Candidate upstream root cause: {candidate_upstream}')

            if specialized:
                lines.append('')
                lines.append(str(specialized.get('heading') or '### RCA').strip())
                summary = str(specialized.get('summary') or '').strip()
                if summary:
                    lines.append(summary)
                detail_lines = specialized.get('detail_lines') or []
                if detail_lines:
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in detail_lines]))

            incident_window = deep_rca.get('incident_window') or {}
            primary_entity = deep_rca.get('primary_entity') or {}
            if incident_window or primary_entity:
                lines.append('')
                lines.append('### RCA Scope')
                if incident_window.get('from') or incident_window.get('to'):
                    lines.append(f'- Window: {incident_window.get("from", "N/A")} to {incident_window.get("to", "N/A")}')
                if incident_window.get('cluster_name') or incident_window.get('cluster_version'):
                    lines.append(
                        f'- Cluster: {incident_window.get("cluster_name", "N/A")} '
                        f'({incident_window.get("cluster_version", "N/A")})'
                    )
                if primary_entity.get('instance'):
                    lines.append(f'- Primary TiKV instance: {primary_entity["instance"]}')
                if primary_entity.get('store'):
                    lines.append(f'- Primary TiDB-side store signal: store {primary_entity["store"]}')

            if specialized:
                evidence = specialized.get('evidence') or []
                if evidence:
                    lines.append('')
                    lines.append('### Primary Evidence')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in evidence]))

                counterevidence = specialized.get('counterevidence') or []
                if counterevidence:
                    lines.append('')
                    lines.append('### Counterevidence')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in counterevidence]))

                not_supported = specialized.get('not_supported') or []
                if not_supported:
                    lines.append('')
                    lines.append('### What Is Not Supported')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in not_supported]))

                actions = specialized.get('recommended_actions') or []
                if actions:
                    lines.append('')
                    lines.append('### Recommended Actions')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in actions]))
            else:
                root_cause = str(deep_rca.get('root_cause_hypothesis') or '').strip()
                if root_cause:
                    lines.append('')
                    lines.append('### Candidate Upstream Root Cause' if leader_drop_mode else '### Root Cause Hypothesis')
                    lines.append(root_cause)

                evidence = deep_rca.get('evidence') or []
                if evidence:
                    lines.append('')
                    lines.append('### Primary Evidence')
                    for entry in evidence:
                        title_text = str(entry.get('title') or 'Evidence').strip()
                        peak_value_text = str(entry.get('peak_value_text') or 'N/A').strip()
                        peak_at = str(entry.get('peak_at') or 'N/A').strip()
                        conclusion_text = str(entry.get('conclusion') or '').strip()
                        lines.append(f'- {title_text}: peak {peak_value_text} at {peak_at}. {conclusion_text}')

                counterevidence = deep_rca.get('counterevidence') or []
                if counterevidence:
                    lines.append('')
                    lines.append('### Counterevidence')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in counterevidence]))

                not_supported = deep_rca.get('what_is_not_supported') or []
                if not_supported:
                    lines.append('')
                    lines.append('### What Is Not Supported')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in not_supported]))

                actions = deep_rca.get('recommended_actions') or []
                if actions:
                    lines.append('')
                    lines.append('### Recommended Actions')
                    lines.extend(markdown_bullet_lines([str(item_) for item_ in actions]))

            if not deep_rca:
                findings = clinic_headline_findings(item)
                if findings:
                    lines.append(f'- Headline findings: {"; ".join(str(finding) for finding in findings[:3])}')
                next_steps = clinic_follow_up_steps(item)
                if next_steps:
                    lines.append(f'- Suggested next steps from clinic evidence: {"; ".join(str(step) for step in next_steps[:2])}')

    if grafana_results:
        lines.append('')
        lines.append('### Grafana-Based Assessment')
        for item in grafana_results:
            link = str(item.get('link') or '').strip()
            if item.get('error'):
                lines.append(f'- Grafana analysis failed for [{link}]({link}): {item["error"]}')
                continue
            cluster_summary = item.get('cluster_summary') or {}
            dashboard = grafana_dashboard_title(item) or 'N/A'
            judgement = 'Collected Grafana access evidence and normalized cluster summary from the datasource.'
            lines.append(f'- Source: [{link}]({link})')
            cluster_summary_lines = render_cluster_summary_lines(cluster_summary)
            if cluster_summary_lines:
                lines.append('')
                lines.extend(cluster_summary_lines)
                lines.append('')
            lines.append(f'- Dashboard: {dashboard}')
            lines.append(f'- Grafana judgement: {judgement}')
            next_steps = grafana_follow_up_steps(item)
            if next_steps:
                lines.append(f'- Suggested next steps from Grafana evidence: {"; ".join(str(step) for step in next_steps[:2])}')

    if bundle_results:
        lines.append('')
        lines.append('### Clinic Bundle Evidence')
        for item in bundle_results:
            link = str(item.get('download_url') or '').strip()
            if item.get('error'):
                lines.append(f'- Clinic bundle download failed for [{link}]({link}): {item["error"]}')
                continue
            archive_path = str(item.get('archive_path') or 'N/A').strip()
            extract_dir = str(item.get('extract_dir') or 'N/A').strip()
            tidb_version = str(item.get('tidb_version') or 'N/A').strip()
            lines.append(f'- Source bundle: [{link}]({link})')
            lines.append(f'- Local archive: `{archive_path}`')
            lines.append(f'- Extracted directory: `{extract_dir}`')
            lines.append(f'- TiDB version from meta.yaml: {tidb_version}')

            component_counts = item.get('component_counts') or {}
            if component_counts:
                counts_text = ', '.join(f'{key}={value}' for key, value in component_counts.items())
                lines.append(f'- Component counts: {counts_text}')

            if item.get('tikv_hosts_with_logs'):
                lines.append(f'- TiKV hosts with extracted logs: {", ".join(item["tikv_hosts_with_logs"])}')
            if item.get('pd_hosts_with_logs'):
                lines.append(f'- PD hosts with extracted logs: {", ".join(item["pd_hosts_with_logs"])}')
            if item.get('tidb_hosts_with_logs'):
                lines.append(f'- TiDB hosts with extracted logs: {", ".join(item["tidb_hosts_with_logs"])}')
            slow_log_files = item.get('slow_log_files') or []
            if slow_log_files:
                lines.append(f'- Slow log files: {len(slow_log_files)} extracted')
            slow_metric_count = int(item.get('slow_metric_count') or 0)
            if slow_metric_count:
                lines.append(f'- Slow-related metric files: {slow_metric_count}')
            warnings = item.get('warnings') or []
            if warnings:
                lines.append('- Bundle warnings:')
                for warning in warnings[:5]:
                    lines.append(f'  - {warning}')

    if attachment_results:
        lines.append('')
        lines.append('### Attachment Evidence')
        if clinic_ready and not clinic_pending:
            successful_attachments = [item for item in attachment_results if not item.get('error')]
            failed_attachments = [item for item in attachment_results if item.get('error')]
            kind_counts = {}
            for item in successful_attachments:
                kind = str(((item.get('analysis') or {}).get('kind') or 'attachment')).strip() or 'attachment'
                kind_counts[kind] = kind_counts.get(kind, 0) + 1
            if successful_attachments:
                lines.append(
                    f'- Downloaded {len(successful_attachments)} attachment(s); Clinic evidence remains primary and attachments are kept as supplementary context only.'
                )
            if kind_counts:
                kind_summary = ', '.join(f'{kind}={count}' for kind, count in sorted(kind_counts.items()))
                lines.append(f'- Attachment types: {kind_summary}')
            suspicious_hits = []
            profile_hints = []
            for item in successful_attachments:
                filename = str(item.get('filename') or 'N/A').strip()
                path = str(item.get('path') or 'N/A').strip()
                suspicious = ((item.get('analysis') or {}).get('suspicious_lines') or [])[:2]
                if suspicious:
                    suspicious_hits.append((filename, path, suspicious))
                profile_hint = str(((item.get('analysis') or {}).get('profile_hint') or '')).strip()
                if profile_hint:
                    profile_hints.append((filename, path, profile_hint))
            if suspicious_hits:
                lines.append('- Text-like attachment clues:')
                for filename, path, suspicious in suspicious_hits[:3]:
                    lines.append(f'  - {filename}: `{path}`')
                    for line in suspicious[:2]:
                        lines.append(f'    - {line}')
            if profile_hints:
                lines.append('- Profile-like attachment notes:')
                for filename, path, profile_hint in profile_hints[:3]:
                    lines.append(f'  - {filename}: {profile_hint} (`{path}`)')
            if failed_attachments:
                lines.append(f'- Attachment download failures: {len(failed_attachments)}')
                for item in failed_attachments[:3]:
                    link = str(item.get('download_url') or '').strip()
                    lines.append(f'  - [{link}]({link}): {item["error"]}')
        else:
            for item in attachment_results:
                link = str(item.get('download_url') or '').strip()
                if item.get('error'):
                    lines.append(f'- Attachment download failed for [{link}]({link}): {item["error"]}')
                    continue
                filename = str(item.get('filename') or 'N/A').strip()
                path = str(item.get('path') or 'N/A').strip()
                kind = str(((item.get('analysis') or {}).get('kind') or '')).strip()
                lines.append(f'- {filename} ({kind or "attachment"})')
                lines.append(f'- Source: [{link}]({link})')
                lines.append(f'- Local path: `{path}`')
                suspicious = ((item.get('analysis') or {}).get('suspicious_lines') or [])[:5]
                if suspicious:
                    lines.append('- Suspicious lines:')
                    for line in suspicious:
                        lines.append(f'  - {line}')
                profile_hint = str(((item.get('analysis') or {}).get('profile_hint') or '')).strip()
                if profile_hint:
                    lines.append(f'- Profile note: {profile_hint}')
    elif attachment_policy:
        mode = str(attachment_policy.get('mode') or '').strip()
        reason = str(attachment_policy.get('reason') or '').strip()
        total_known_bytes = int(attachment_policy.get('total_known_bytes') or 0)
        total_known_mb = total_known_bytes / (1024 * 1024)
        limit_mb = int((attachment_policy.get('auto_download_limit_bytes') or 0) / (1024 * 1024)) if attachment_policy.get('auto_download_limit_bytes') else 0
        unknown_size_count = int(attachment_policy.get('unknown_size_count') or 0)
        if mode in ('blocked', 'skipped'):
            lines.append('')
            lines.append('### Attachment Policy')
            if reason == 'clinic_evidence_sufficient':
                lines.append('- Clinic-first analysis already produced enough evidence, so attachments were not downloaded automatically.')
            elif reason == 'attachments_not_indicated_for_analysis':
                lines.append('- Attachments were available, but the current case did not require attachment-based follow-up after Clinic-first analysis.')
            elif reason == 'attachments_skipped_by_flag':
                lines.append('- Attachment analysis was skipped by `--skip-attachments`.')
            elif reason == 'attachment_size_exceeds_auto_limit':
                lines.append(
                    f'- Attachment-backed analysis stopped because the attachment set is {total_known_mb:.1f} MB, which exceeds the {limit_mb} MB auto-download limit.'
                )
                lines.append('- Re-run with `--download-attachments` to force attachment-based analysis.')
            elif reason == 'attachment_size_unknown':
                lines.append(
                    f'- Attachment-backed analysis stopped because {unknown_size_count} attachment(s) do not expose a known size for the auto-download policy.'
                )
                lines.append('- Re-run with `--download-attachments` to force attachment-based analysis.')

    ticket_clue_lines = []
    if phenomenon:
        ticket_clue_lines.append(f'- Phenomenon: {phenomenon}')
    root_cause = str(owner_conclusion.get('root_cause') or '').strip()
    workaround = str(owner_conclusion.get('workaround') or '').strip()
    todo = str(owner_conclusion.get('todo') or '').strip()
    if root_cause:
        ticket_clue_lines.append(f'- Ticket owner clue on possible root cause: {root_cause}')
    if workaround:
        ticket_clue_lines.append(f'- Ticket owner clue on workaround: {workaround}')
    if todo:
        ticket_clue_lines.append(f'- Ticket owner clue on next action: {todo}')
    if ticket_clue_lines:
        lines.append('')
        lines.append('### Ticket Clues')
        if clinic_pending:
            lines.append('These are reference clues only. Final diagnosis should wait for Clinic readiness and be based on Clinic metrics, logs, or bundle evidence.')
        lines.extend(ticket_clue_lines)

    if recent_comments:
        lines.append('')
        lines.append('### Recent Comments')
        if clinic_pending:
            lines.append('These comments are preserved as hints only and are intentionally not used as the current conclusion because Clinic data is still pending.')
        for item in recent_comments:
            created = str(item.get('created') or 'N/A').strip()
            author = str(item.get('author') or 'N/A').strip()
            summary = str(item.get('summary') or 'N/A').strip()
            suffix = ''
            if 'is_internal' in item:
                suffix = ' | internal=yes' if item.get('is_internal') else ' | internal=no'
            lines.append(f'- {created} | {author}{suffix} | {summary}')

    if comment_corpus:
        lines.append('')
        lines.append('### Full Comment Corpus')
        lines.append(f'- Total collected comment summaries used as analyzer input: {len(comment_corpus)}')
        for item in comment_corpus[:12]:
            created = str(item.get('created') or 'N/A').strip()
            author = str(item.get('author') or 'N/A').strip()
            summary = str(item.get('summary') or 'N/A').strip()
            suffix = ''
            if 'is_internal' in item:
                suffix = ' | internal=yes' if item.get('is_internal') else ' | internal=no'
            lines.append(f'- {created} | {author}{suffix} | {summary}')

    recommended_skills = diagnosis.get('recommended_skills') or []
    routing_reasons = diagnosis.get('routing_reasons') or []
    if recommended_skills:
        lines.append('')
        lines.append('### Recommended Skills')
        lines.extend(markdown_bullet_lines([str(skill) for skill in recommended_skills]))
    if routing_reasons:
        lines.append('')
        lines.append('### Routing Reasons')
        lines.extend(markdown_bullet_lines([str(reason) for reason in routing_reasons]))

    if diagnosis_result:
        lines.append('')
        lines.append('### Structured Diagnosis Result')
        lines.append(f'- entry_mode: {diagnosis_result.get("entry_mode") or "unknown"}')
        activated = diagnosis_result.get('activated_subcapabilities') or []
        lines.append(f'- activated_subcapabilities: {", ".join(str(item) for item in activated) if activated else "N/A"}')
        lines.append(f'- is_tikv_side: {diagnosis_result.get("is_tikv_side")}')
        lines.append(f'- scope: {diagnosis_result.get("scope") or "unknown"}')
        lines.append(f'- path: {diagnosis_result.get("path") or "unknown"}')
        lines.append(f'- earliest_proven_bottleneck: {diagnosis_result.get("earliest_proven_bottleneck") or "unproven"}')
        key_evidence = diagnosis_result.get('key_evidence') or []
        if key_evidence:
            lines.append('- key_evidence:')
            for item in key_evidence[:3]:
                lines.append(f'  - {item}')
        missing_evidence = diagnosis_result.get('missing_evidence') or []
        if missing_evidence:
            lines.append('- missing_evidence:')
            for item in missing_evidence[:4]:
                lines.append(f'  - {item}')
        uncertainty_statement = str(diagnosis_result.get('uncertainty_statement') or '').strip()
        if uncertainty_statement:
            lines.append(f'- uncertainty_statement: {uncertainty_statement}')
        provisional_conclusion = str(diagnosis_result.get('provisional_conclusion') or '').strip()
        if provisional_conclusion:
            lines.append(f'- provisional_conclusion: {provisional_conclusion}')
        reference_files = diagnosis_result.get('reference_files_consulted') or []
        if reference_files:
            lines.append(f'- reference_files_consulted: {", ".join(str(item) for item in reference_files)}')

    if analysis_tracks:
        lines.append('')
        lines.append('### Analysis Tracks')
        lines.append('The analyzer splits evidence by time window and source so separate slow periods do not get blended into one conclusion.')
        for track in analysis_tracks:
            label = str(track.get('label') or 'track').strip()
            from_text = str(track.get('from') or 'N/A').strip()
            to_text = str(track.get('to') or 'N/A').strip()
            sample_count = int(track.get('sample_count') or 0)
            dominant_stage = str(track.get('dominant_stage') or '').strip()
            dominant_store = str(track.get('dominant_store') or '').strip()
            summary = str(track.get('summary') or '').strip()
            lines.append(f'- {label}: {from_text} -> {to_text}')
            if sample_count:
                lines.append(f'  - samples: {sample_count}')
            if dominant_stage:
                lines.append(f'  - dominant_stage: {dominant_stage}')
            if dominant_store:
                lines.append(f'  - dominant_store: {dominant_store}')
            if summary:
                lines.append(f'  - summary: {summary}')
            for hint in (track.get('focus_hints') or [])[:3]:
                lines.append(f'  - clinic_focus_hint: {hint}')

    clinic_refs = ticket_context.get('clinic_references') or []
    if clinic_refs:
        lines.append('')
        lines.append('### Clinic Links')
        for item in clinic_refs:
            link = str(item.get('link') or '').strip()
            source = str(item.get('source') or 'unknown').strip()
            author = str(item.get('author') or '').strip()
            created = str(item.get('created') or '').strip()
            meta = [f'source={source}']
            if author:
                meta.append(f'author={author}')
            if created:
                meta.append(f'created={created}')
            lines.append(f'- [{link}]({link}) | {" | ".join(meta)}')

    grafana_refs = ticket_context.get('grafana_references') or []
    if grafana_refs:
        lines.append('')
        lines.append('### Grafana Links')
        for item in grafana_refs:
            link = str(item.get('link') or '').strip()
            source = str(item.get('source') or 'unknown').strip()
            author = str(item.get('author') or '').strip()
            created = str(item.get('created') or '').strip()
            meta = [f'source={source}']
            if author:
                meta.append(f'author={author}')
            if created:
                meta.append(f'created={created}')
            lines.append(f'- [{link}]({link}) | {" | ".join(meta)}')

    clinic_download_refs = ticket_context.get('clinic_download_references') or []
    if clinic_download_refs:
        lines.append('')
        lines.append('### Clinic Bundle Links')
        for item in clinic_download_refs:
            link = str(item.get('link') or '').strip()
            source = str(item.get('source') or 'unknown').strip()
            author = str(item.get('author') or '').strip()
            created = str(item.get('created') or '').strip()
            meta = [f'source={source}']
            if author:
                meta.append(f'author={author}')
            if created:
                meta.append(f'created={created}')
            lines.append(f'- [{link}]({link}) | {" | ".join(meta)}')

    attachment_refs = ticket_context.get('attachment_references') or []
    if attachment_refs:
        lines.append('')
        lines.append('### Attachment Links')
        if clinic_ready and not clinic_pending:
            source_counts = {}
            for item in attachment_refs:
                source = str(item.get('source') or 'unknown').strip() or 'unknown'
                source_counts[source] = source_counts.get(source, 0) + 1
            lines.append(f'- Total attachment links: {len(attachment_refs)}')
            if source_counts:
                source_summary = ', '.join(f'{source}={count}' for source, count in sorted(source_counts.items()))
                lines.append(f'- Attachment link sources: {source_summary}')
            if attachment_refs:
                lines.append('- Example attachment links:')
                for item in attachment_refs[:3]:
                    link = str(item.get('link') or '').strip()
                    source = str(item.get('source') or 'unknown').strip()
                    author = str(item.get('author') or '').strip()
                    created = str(item.get('created') or '').strip()
                    filename = str(item.get('filename') or '').strip()
                    meta = [f'source={source}']
                    if filename:
                        meta.append(f'filename={filename}')
                    if author:
                        meta.append(f'author={author}')
                    if created:
                        meta.append(f'created={created}')
                    lines.append(f'- [{link}]({link}) | {" | ".join(meta)}')
        else:
            for item in attachment_refs:
                link = str(item.get('link') or '').strip()
                source = str(item.get('source') or 'unknown').strip()
                author = str(item.get('author') or '').strip()
                created = str(item.get('created') or '').strip()
                filename = str(item.get('filename') or '').strip()
                meta = [f'source={source}']
                if filename:
                    meta.append(f'filename={filename}')
                if author:
                    meta.append(f'author={author}')
                if created:
                    meta.append(f'created={created}')
                lines.append(f'- [{link}]({link}) | {" | ".join(meta)}')

    return '\n'.join(lines).strip() + '\n'


def analyze_cmd(args):
    try:
        context = resolve_analyze_context(
            args,
            is_pingkai_issue=is_pingkai_issue,
            is_pingkai_report_issue=is_pingkai_report_issue,
            build_pingkai_context=build_pingkai_analyze_context,
            build_jira_context=build_jira_analyze_context,
            build_grafana_context=build_grafana_analyze_context,
            build_clinic_context=build_clinic_analyze_context,
            apply_window_to_clinic_link=apply_window_to_clinic_link,
            is_grafana_link=is_grafana_link,
            prepend_reference=prepend_reference,
        )
    except (RuntimeError, ValueError) as exc:
        print(str(exc))
        return 1

    payload, exit_code = execute_analyze_context(
        context,
        args,
        run_grafana_analyze=run_grafana_analyze,
        run_clinic_analyze=run_clinic_analyze,
        run_clinic_bundle_download=run_clinic_bundle_download,
        context_requires_attachment_analysis=context_requires_attachment_analysis,
        analyze_missing_evidence_reasons=analyze_missing_evidence_reasons,
        run_attachment_downloads=run_attachment_downloads,
        build_analyze_payload=build_analyze_payload,
    )
    if getattr(args, 'json_output', False):
        print_json(payload)
    else:
        sys.stdout.write(render_analyze_markdown(payload))
    return exit_code


def build_jira_analyze_context(issue_id: str) -> dict:
    raw = get_issue_raw(issue_id)
    if not raw:
        raise RuntimeError(f'Failed to load Jira issue: {issue_id}')

    fields = (raw or {}).get('fields') or {}
    title = fields.get('summary') or issue_id
    description_text = extract_text(fields.get('description'))
    owner_comment_text = collect_owner_comment_text(raw, fields)
    root_cause, workaround, todo = summarize_owner_comments(owner_comment_text)
    clinic_refs = extract_jira_clinic_references(raw)
    grafana_refs = extract_jira_grafana_references(raw)
    clinic_download_refs = extract_jira_clinic_download_references(raw)
    attachment_refs = extract_jira_attachment_references(raw)
    recent_comments = recent_jira_comment_context(raw)
    comment_corpus = jira_comment_corpus_context(raw)
    problem_statement = summarize_text(f'{title}. {description_text}', 320) or title

    context_text = '\n'.join(
        [
            title,
            description_text,
            owner_comment_text,
            '\n'.join(item.get('summary', '') for item in comment_corpus),
        ]
    )
    recommended_skills, reasons = recommend_diagnosis_skills(context_text, clinic_refs or clinic_download_refs or grafana_refs)

    return {
        'source': 'jira',
        'issue_ref': issue_id,
        'title': title,
        'problem_statement': problem_statement,
        'ticket_url': issue_url(issue_id),
        'phenomenon': summarize_description_for_phenomenon(description_text),
        'owner_conclusion': {
            'root_cause': root_cause,
            'workaround': workaround,
            'todo': todo,
        },
        'recent_comments': recent_comments,
        'comment_corpus': comment_corpus,
        'clinic_references': clinic_refs,
        'grafana_references': grafana_refs,
        'clinic_download_references': clinic_download_refs,
        'attachment_references': attachment_refs,
        'recommended_skills': recommended_skills,
        'routing_reasons': reasons,
        'routing_context_text': context_text,
    }


def build_pingkai_analyze_context(issue_ref: str) -> dict:
    client, ticket = load_pingkai_ticket(issue_ref, include_internal=True)
    if not client or not ticket:
        raise RuntimeError(f'Failed to load Pingkai ticket: {issue_ref}')

    title = ticket.get('title') or issue_ref
    description_text = strip_text_for_pingkai_comment(ticket.get('description', ''))
    owner_entries = pingkai_owner_comment_entries(ticket)
    root_cause, workaround, todo = extract_pingkai_owner_conclusions(owner_entries)
    clinic_refs = extract_pingkai_clinic_references(ticket)
    grafana_refs = extract_pingkai_grafana_references(ticket)
    clinic_download_refs = extract_pingkai_clinic_download_references(ticket)
    attachment_refs = extract_pingkai_attachment_references(ticket)
    recent_comments = recent_pingkai_comment_context(ticket)
    comment_corpus = pingkai_comment_corpus_context(ticket)
    problem_statement = summarize_text(f'{title}. {description_text}', 320) or title

    context_text = '\n'.join(
        [
            title,
            description_text,
            '\n'.join(item.get('body', '') for item in owner_entries),
            '\n'.join(item.get('summary', '') for item in comment_corpus),
        ]
    )
    recommended_skills, reasons = recommend_diagnosis_skills(context_text, clinic_refs or clinic_download_refs or grafana_refs)

    return {
        'source': 'pingkai',
        'issue_ref': pingkai_ticket_ref(ticket),
        'title': title,
        'problem_statement': problem_statement,
        'ticket_url': pingkai_ticket_link(ticket, client),
        'phenomenon': summarize_description_for_phenomenon(description_text or title),
        'owner_conclusion': {
            'root_cause': root_cause,
            'workaround': workaround,
            'todo': todo,
        },
        'recent_comments': recent_comments,
        'comment_corpus': comment_corpus,
        'clinic_references': clinic_refs,
        'grafana_references': grafana_refs,
        'clinic_download_references': clinic_download_refs,
        'attachment_references': attachment_refs,
        'recommended_skills': recommended_skills,
        'routing_reasons': reasons,
        'routing_context_text': context_text,
    }


def build_grafana_analyze_context(link: str, args) -> dict:
    from_ts = str(getattr(args, 'from_ts', '') or '').strip()
    to_ts = str(getattr(args, 'to_ts', '') or '').strip()
    title = 'Grafana analysis'
    time_bits = []
    if from_ts:
        time_bits.append(f'from={from_ts}')
    if to_ts:
        time_bits.append(f'to={to_ts}')
    problem_statement = f'Analyze Grafana link {link}'
    if time_bits:
        problem_statement += f' with window {" ".join(time_bits)}'
    recommended_skills = ['tidb-perf-diagnosis', 'op-clinic-grafana-access']
    reasons = [
        'Use tidb-perf-diagnosis as the default performance triage skill for monitor-driven diagnosis.',
        'Use op-clinic-grafana-access because it provides the shared Clinic / Grafana access and evidence-normalization flow.',
    ]
    return {
        'source': 'grafana',
        'issue_ref': 'grafana-link',
        'title': title,
        'problem_statement': problem_statement,
        'ticket_url': link,
        'phenomenon': '',
        'owner_conclusion': {},
        'recent_comments': [],
        'clinic_references': [],
        'clinic_download_references': [],
        'attachment_references': [],
        'grafana_references': [{'link': link, 'source': 'direct-input', 'author': '', 'created': ''}],
        'recommended_skills': recommended_skills,
        'routing_reasons': reasons,
        'routing_context_text': problem_statement,
    }


def build_clinic_analyze_context(link: str, args) -> dict:
    title = 'Clinic analysis'
    problem_statement = f'Analyze Clinic link {link}'
    if not clinic_link_has_window(link):
        from_ts = str(getattr(args, 'from_ts', '') or '').strip()
        to_ts = str(getattr(args, 'to_ts', '') or '').strip()
        if from_ts and to_ts:
            problem_statement += f' with window from={from_ts} to={to_ts}'
    recommended_skills = ['tidb-perf-diagnosis', 'op-clinic-grafana-access']
    reasons = [
        'Use tidb-perf-diagnosis as the default performance triage skill for monitor-driven diagnosis.',
        'Use op-clinic-grafana-access because it provides the shared Clinic / Grafana access and evidence-normalization flow.',
    ]
    return {
        'source': 'clinic',
        'issue_ref': 'clinic-link',
        'title': title,
        'problem_statement': problem_statement,
        'ticket_url': link,
        'phenomenon': '',
        'owner_conclusion': {},
        'recent_comments': [],
        'clinic_references': [{'link': link, 'source': 'cli-flag', 'author': '', 'created': ''}],
        'clinic_download_references': [],
        'attachment_references': [],
        'grafana_references': [],
        'recommended_skills': recommended_skills,
        'routing_reasons': reasons,
        'routing_context_text': problem_statement,
    }


def is_tikv_slow_focus_text(title: str, description_text: str) -> bool:
    return workflow_is_tikv_slow_focus_text(
        title,
        description_text,
        keywords=TIKV_SLOW_EVALUATE_KEYWORDS,
        negative_keywords=TIKV_SLOW_EVALUATE_NEGATIVE_KEYWORDS,
    )


def build_tikv_slow_eval_case_from_jira(raw: dict) -> dict:
    return build_eval_case_from_jira_raw(
        raw,
        extract_text=extract_text,
        extract_clinic_links=extract_jira_initial_clinic_links,
        extract_grafana_links=extract_jira_initial_grafana_links,
        summarize_description=summarize_description_for_phenomenon,
        parse_sort_datetime=parse_sort_datetime,
        issue_url=issue_url,
        focus_keywords=TIKV_SLOW_EVALUATE_KEYWORDS,
        negative_keywords=TIKV_SLOW_EVALUATE_NEGATIVE_KEYWORDS,
    ).to_dict()


def build_tikv_slow_eval_case_from_pingkai(ticket: dict, client: TicketAPIClient | None = None) -> dict:
    return build_eval_case_from_pingkai_ticket(
        ticket,
        ticket_ref=pingkai_ticket_ref,
        ticket_link=pingkai_ticket_link,
        strip_text=strip_text_for_pingkai_comment,
        extract_clinic_links=extract_pingkai_initial_clinic_links,
        extract_grafana_links=extract_pingkai_initial_grafana_links,
        summarize_description=summarize_description_for_phenomenon,
        parse_datetime=parse_pingkai_datetime,
        client=client,
        focus_keywords=TIKV_SLOW_EVALUATE_KEYWORDS,
        negative_keywords=TIKV_SLOW_EVALUATE_NEGATIVE_KEYWORDS,
    ).to_dict()


def evaluate_exclusion_reason(
    case: dict,
    args,
    enforce_focus: bool = True,
    explicit_case: bool = False,
) -> str:
    case_obj = EvaluateCase(
        incident_id=str(case.get('incident_id') or ''),
        source=str(case.get('source') or ''),
        title=str(case.get('title') or ''),
        description_text=str(case.get('description_text') or ''),
        initial_symptom=str(case.get('initial_symptom') or ''),
        clinic_links=list(case.get('clinic_links') or []),
        grafana_links=list(case.get('grafana_links') or []),
        ticket_url=str(case.get('ticket_url') or ''),
        created_at=case.get('created_at'),
        l2_assignee=str(case.get('l2_assignee') or 'N/A'),
        assignee_or_l3=str(case.get('assignee_or_l3') or 'N/A'),
        focus_match=bool(case.get('focus_match')),
    )
    return workflow_evaluate_exclusion_reason(
        case_obj,
        args,
        enforce_focus=enforce_focus,
        explicit_case=explicit_case,
    )


def discover_tikv_slow_jira_cases(args) -> tuple[list[dict], list[dict]]:
    jira_source = make_jira_source_adapter()

    def build_case_from_raw(raw):
        return build_eval_case_from_jira_raw(
            raw,
            extract_text=extract_text,
            extract_clinic_links=extract_jira_initial_clinic_links,
            extract_grafana_links=extract_jira_initial_grafana_links,
            summarize_description=summarize_description_for_phenomenon,
            parse_sort_datetime=parse_sort_datetime,
            issue_url=issue_url,
            focus_keywords=TIKV_SLOW_EVALUATE_KEYWORDS,
            negative_keywords=TIKV_SLOW_EVALUATE_NEGATIVE_KEYWORDS,
        )

    cases, excluded = discover_jira_cases(args, jira_source, build_case_from_raw=build_case_from_raw)
    return [item.to_dict() for item in cases], [item.to_dict() for item in excluded]


def discover_tikv_slow_pingkai_cases(args) -> tuple[list[dict], list[dict]]:
    pingkai_source = PingkaiSourceAdapter(
        list_entries=list_pingkai_entries,
        load_ticket=load_pingkai_ticket,
        ticket_ref=pingkai_ticket_ref,
    )

    def build_case_from_ticket(ticket, *, client=None):
        return build_eval_case_from_pingkai_ticket(
            ticket,
            ticket_ref=pingkai_ticket_ref,
            ticket_link=pingkai_ticket_link,
            strip_text=strip_text_for_pingkai_comment,
            extract_clinic_links=extract_pingkai_initial_clinic_links,
            extract_grafana_links=extract_pingkai_initial_grafana_links,
            summarize_description=summarize_description_for_phenomenon,
            parse_datetime=parse_pingkai_datetime,
            client=client,
            focus_keywords=TIKV_SLOW_EVALUATE_KEYWORDS,
            negative_keywords=TIKV_SLOW_EVALUATE_NEGATIVE_KEYWORDS,
        )

    cases, excluded = discover_pingkai_cases(args, pingkai_source, build_case_from_ticket=build_case_from_ticket)
    return [item.to_dict() for item in cases], [item.to_dict() for item in excluded]


def render_evaluate_markdown(payload: dict) -> str:
    lines = [f'## Evaluate {payload["profile"]}', '']
    scope = payload.get('scope') or {}
    replay_contract = payload.get('replay_contract') or {}
    replay_ready_cases = payload.get('replay_ready_cases') or []
    comparison_contract = payload.get('comparison_contract') or {}
    case_reviews = payload.get('case_reviews') or []
    workflow_improvements = payload.get('workflow_improvements') or []
    knowledge_improvements = payload.get('knowledge_improvements') or []
    output_improvements = payload.get('output_improvements') or []
    do_not_change_yet = payload.get('do_not_change_yet') or []
    lines.append('### Scope')
    lines.append(f'- source: {scope.get("source", "N/A")}')
    lines.append(f'- time_range: {scope.get("time_range", "N/A")}')
    lines.append(f'- assignee: {scope.get("assignee", "N/A")}')
    lines.append(f'- sample_size: {scope.get("sample_size", "N/A")}')
    lines.append(f'- clinic_only: {scope.get("clinic_only", False)}')
    lines.append(f'- ignore_grafana_only: {scope.get("ignore_grafana_only", False)}')
    lines.append('')
    lines.append('### Sample Summary')
    lines.append(f'- included: {len(payload.get("included_cases") or [])}')
    lines.append(f'- excluded: {len(payload.get("excluded_cases") or [])}')
    if replay_contract:
        lines.append(f'- replay_entry_modes: {", ".join(replay_contract.get("allowed_entry_modes") or [])}')
    if comparison_contract:
        lines.append(f'- comparison_match_levels: {", ".join(comparison_contract.get("allowed_match_levels") or [])}')
    lines.append('')
    lines.append('### Included Cases')
    included = payload.get('included_cases') or []
    if not included:
        lines.append('- None')
    for case in included:
        clinic_count = len(case.get('clinic_links') or [])
        grafana_count = len(case.get('grafana_links') or [])
        lines.append(
            f'- {case["incident_id"]} | {case["source"]} | clinic_links={clinic_count} | '
            f'grafana_links={grafana_count} | symptom={case.get("initial_symptom") or "N/A"}'
        )
    if replay_ready_cases:
        lines.append('')
        lines.append('### Replay-Ready Cases')
        for case in replay_ready_cases:
            canonical = case.get('canonical_incident') or {}
            replay_inputs = case.get('replay_inputs') or {}
            lines.append(
                f'- {case.get("incident_id", "N/A")} | cluster_id={canonical.get("cluster_id") or "N/A"} | '
                f'time_window={canonical.get("time_window") or "N/A"} | used_fields={", ".join(replay_inputs.get("used_fields") or [])}'
            )
    if case_reviews:
        lines.append('')
        lines.append('### Case Reviews')
        for review in case_reviews:
            notes = review.get('comparison_notes') or []
            note_text = notes[-1] if notes else 'N/A'
            skill_conclusion = review.get('skill_conclusion') or {}
            final_human = review.get('final_human_conclusion') or {}
            match_level = review.get('match_level') or 'pending'
            human_status = final_human.get('status') or 'pending'
            entry_mode = skill_conclusion.get('entry_mode') or 'pending'
            mismatch_text = ','.join(review.get('mismatch_type') or []) or 'none'
            improvement = ((review.get('improvement_suggestions') or [])[:1] or [''])[0]
            line = (
                f'- {review.get("incident_id", "N/A")} | replay_entry_mode={entry_mode} | '
                f'match_level={match_level} | mismatch_type={mismatch_text} | '
                f'human_conclusion_status={human_status} | note={note_text}'
            )
            if improvement:
                line += f' | improvement={improvement}'
            lines.append(line)
    if workflow_improvements:
        lines.append('')
        lines.append('### Workflow Improvements')
        for item in workflow_improvements:
            lines.append(
                f'- {item.get("title", "N/A")} | merge_readiness={item.get("merge_readiness", "N/A")} | '
                f'cases={", ".join(item.get("evidence_case_ids") or []) or "N/A"}'
            )
    if knowledge_improvements:
        lines.append('')
        lines.append('### Knowledge Improvements')
        for item in knowledge_improvements:
            lines.append(
                f'- {item.get("title", "N/A")} | merge_readiness={item.get("merge_readiness", "N/A")} | '
                f'cases={", ".join(item.get("evidence_case_ids") or []) or "N/A"}'
            )
    if output_improvements:
        lines.append('')
        lines.append('### Output Improvements')
        for item in output_improvements:
            lines.append(
                f'- {item.get("title", "N/A")} | merge_readiness={item.get("merge_readiness", "N/A")} | '
                f'cases={", ".join(item.get("evidence_case_ids") or []) or "N/A"}'
            )
    if do_not_change_yet:
        lines.append('')
        lines.append('### Do Not Change Yet')
        for item in do_not_change_yet:
            lines.append(
                f'- {item.get("title", "N/A")} | merge_readiness={item.get("merge_readiness", "N/A")} | '
                f'cases={", ".join(item.get("evidence_case_ids") or []) or "N/A"}'
            )
    lines.append('')
    lines.append('### Excluded Cases')
    excluded = payload.get('excluded_cases') or []
    if not excluded:
        lines.append('- None')
    for item in excluded:
        lines.append(f'- {item.get("incident_id", "N/A")} | {item.get("reason", "N/A")}')
    lines.append('')
    lines.append('### Execution Hint')
    lines.append('- Use the embedded `oncall-toolkit evaluate tikv-slow` replay rules to review each included case from the initial ticket-visible inputs only.')
    return '\n'.join(lines).rstrip() + '\n'


def build_evaluate_live_analyze_args(args):
    return argparse.Namespace(
        clinic_base_url=str(getattr(args, 'clinic_base_url', '') or '').strip(),
        clinic_profile=str(getattr(args, 'clinic_profile', '') or '').strip(),
        clinic_dashboard_name=str(getattr(args, 'clinic_dashboard_name', '') or '').strip(),
        grafana_profile=str(getattr(args, 'grafana_profile', '') or '').strip(),
        grafana_datasource_id=str(getattr(args, 'grafana_datasource_id', '') or '1').strip(),
        grafana_dashboard_uid=str(getattr(args, 'grafana_dashboard_uid', '') or '').strip(),
        from_ts=str(getattr(args, 'from_ts', '') or '').strip(),
        to_ts=str(getattr(args, 'to_ts', '') or '').strip(),
        max_clinic_links=1,
        skip_clinic=False,
        skip_clinic_download=True,
        skip_attachments=True,
        download_attachments=False,
        attachment_download_root='',
        clinic_download_root='',
        full_clinic_json=False,
    )


def extract_live_clinic_diagnosis_result(payload: dict) -> dict:
    diagnosis = (payload or {}).get('diagnosis') or {}
    diagnosis_result = diagnosis.get('diagnosis_result') or {}
    clinic_analysis = diagnosis.get('clinic_analysis') or []
    successful_ready = [
        item for item in clinic_analysis
        if not item.get('error') and (item.get('data_status') or {}).get('ready') is not False
    ]
    if not successful_ready:
        return {}
    return diagnosis_result if isinstance(diagnosis_result, dict) else {}


def evaluate_cmd(args):
    profile = str(getattr(args, 'profile', '') or '').strip().lower()
    if profile != TIKV_SLOW_EVALUATE_PROFILE:
        print(f'unsupported evaluate profile: {args.profile}')
        return 1

    try:
        sources = parse_sources(getattr(args, 'source', TIKV_SLOW_EVALUATE_DEFAULT_SOURCE), default=TIKV_SLOW_EVALUATE_DEFAULT_SOURCE)
    except ValueError as e:
        print(str(e))
        return 1

    explicit_ids = [str(x or '').strip() for x in getattr(args, 'issue_ids', []) if str(x or '').strip()]
    included = []
    excluded = []

    if explicit_ids:
        for issue_id in explicit_ids:
            if issue_id.upper().startswith('GTOC-'):
                raw = get_issue_raw(issue_id)
                if not raw:
                    excluded.append({'incident_id': issue_id, 'reason': 'jira issue is not accessible or does not exist'})
                    continue
                case = build_tikv_slow_eval_case_from_jira(raw)
            elif is_pingkai_issue(issue_id):
                _, ticket = load_pingkai_ticket(issue_id, include_internal=True)
                if not ticket:
                    excluded.append({'incident_id': issue_id, 'reason': 'failed to load pingkai ticket body'})
                    continue
                case = build_tikv_slow_eval_case_from_pingkai(ticket)
            else:
                excluded.append({'incident_id': issue_id, 'reason': 'unsupported issue id for evaluate tikv-slow'})
                continue
            reason = evaluate_exclusion_reason(case, args, enforce_focus=False, explicit_case=True)
            if reason:
                excluded.append({'incident_id': case['incident_id'], 'reason': reason})
                continue
            included.append(case)
    else:
        if 'jira' in sources:
            jira_cases, jira_excluded = discover_tikv_slow_jira_cases(args)
            included.extend(jira_cases)
            excluded.extend(jira_excluded)
        if 'pingkai' in sources:
            pingkai_cases, pingkai_excluded = discover_tikv_slow_pingkai_cases(args)
            included.extend(pingkai_cases)
            excluded.extend(pingkai_excluded)

    included.sort(key=lambda item: item.get('created_at') or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
    sample_size = max(1, int(getattr(args, 'sample_size', TIKV_SLOW_EVALUATE_DEFAULT_SAMPLE_SIZE) or TIKV_SLOW_EVALUATE_DEFAULT_SAMPLE_SIZE))
    if len(included) > sample_size:
        overflow = included[sample_size:]
        included = included[:sample_size]
        for item in overflow:
            excluded.append({'incident_id': item['incident_id'], 'reason': f'not selected into sample_size={sample_size}'})

    jira_source = make_jira_source_adapter()
    pingkai_source = make_pingkai_source_adapter()
    live_analyze_args = build_evaluate_live_analyze_args(args)
    maintainer_grafana_links = [str(item or '').strip() for item in (getattr(args, 'maintainer_grafana_link', []) or []) if str(item or '').strip()]
    maintainer_window_specs = []
    for raw_window in getattr(args, 'maintainer_window', []) or []:
        text = str(raw_window or '').strip()
        if not text:
            continue
        parts = [part.strip() for part in text.split('|')]
        while len(parts) < 6:
            parts.append('')
        label, start_time, end_time, summary, inferred_path, inferred_bottleneck = parts[:6]
        evidence = [item for item in [summary, start_time, end_time] if item]
        maintainer_window_specs.append(
            {
                'label': label or f'window-{len(maintainer_window_specs) + 1}',
                'start_time': start_time,
                'end_time': end_time,
                'summary': summary,
                'inferred_path': inferred_path,
                'inferred_bottleneck': inferred_bottleneck,
                'evidence': evidence,
            }
        )
    replay_ready_case_objects = [build_replay_ready_case(item) for item in included]
    replay_ready_cases = [item.to_dict() for item in replay_ready_case_objects]
    case_reviews = []
    for replay_case in replay_ready_case_objects:
        if replay_case.source == 'jira_gtoc':
            raw = jira_source.get_issue_raw(replay_case.incident_id)
            if raw:
                review = build_case_review_from_jira_case(
                    replay_case,
                    raw,
                    extract_comment_entries_with_time=extract_comment_entries_with_time,
                    author_set=author_set,
                    is_owner_comment=is_owner_comment,
                    clean_line=clean_line,
                    summarize_owner_comments=summarize_owner_comments,
                    context_is_leader_drop=context_is_leader_drop,
                    context_is_network_issue=context_is_network_issue,
                    context_is_flow_control=context_is_flow_control,
                )
                try:
                    live_context = build_jira_analyze_context(replay_case.incident_id)
                    live_payload, _ = execute_analyze_context(
                        live_context,
                        live_analyze_args,
                        run_grafana_analyze=run_grafana_analyze,
                        run_clinic_analyze=run_clinic_analyze,
                        run_clinic_bundle_download=run_clinic_bundle_download,
                        context_requires_attachment_analysis=context_requires_attachment_analysis,
                        analyze_missing_evidence_reasons=analyze_missing_evidence_reasons,
                        run_attachment_downloads=run_attachment_downloads,
                        build_analyze_payload=build_analyze_payload,
                    )
                    live_diagnosis_result = extract_live_clinic_diagnosis_result(live_payload)
                    if live_diagnosis_result:
                        review = apply_live_diagnosis_result_to_case_review(
                            review,
                            live_diagnosis_result,
                            source_note='Skill conclusion was generated from live Clinic-assisted analyze output.',
                        )
                except Exception:
                    pass
                if explicit_ids and (maintainer_grafana_links or maintainer_window_specs):
                    review = apply_maintainer_evidence_hints_to_case_review(
                        review,
                        grafana_links=maintainer_grafana_links,
                        window_specs=maintainer_window_specs,
                        source_note='Maintainer-only evidence hints were attached to this case review for window-level evaluation.',
                    )
                case_reviews.append(review.to_dict())
                continue
        if replay_case.source == 'pingkai':
            _, ticket = pingkai_source.load_ticket(replay_case.incident_id, include_internal=True)
            if ticket:
                review = build_case_review_from_pingkai_case(
                    replay_case,
                    ticket,
                    pingkai_owner_comment_entries=pingkai_owner_comment_entries,
                    clean_line=clean_line,
                    extract_pingkai_owner_conclusions=extract_pingkai_owner_conclusions,
                    context_is_leader_drop=context_is_leader_drop,
                    context_is_network_issue=context_is_network_issue,
                    context_is_flow_control=context_is_flow_control,
                )
                try:
                    live_context = build_pingkai_analyze_context(replay_case.incident_id)
                    live_payload, _ = execute_analyze_context(
                        live_context,
                        live_analyze_args,
                        run_grafana_analyze=run_grafana_analyze,
                        run_clinic_analyze=run_clinic_analyze,
                        run_clinic_bundle_download=run_clinic_bundle_download,
                        context_requires_attachment_analysis=context_requires_attachment_analysis,
                        analyze_missing_evidence_reasons=analyze_missing_evidence_reasons,
                        run_attachment_downloads=run_attachment_downloads,
                        build_analyze_payload=build_analyze_payload,
                    )
                    live_diagnosis_result = extract_live_clinic_diagnosis_result(live_payload)
                    if live_diagnosis_result:
                        review = apply_live_diagnosis_result_to_case_review(
                            review,
                            live_diagnosis_result,
                            source_note='Skill conclusion was generated from live Clinic-assisted analyze output.',
                        )
                except Exception:
                    pass
                if explicit_ids and (maintainer_grafana_links or maintainer_window_specs):
                    review = apply_maintainer_evidence_hints_to_case_review(
                        review,
                        grafana_links=maintainer_grafana_links,
                        window_specs=maintainer_window_specs,
                        source_note='Maintainer-only evidence hints were attached to this case review for window-level evaluation.',
                    )
                case_reviews.append(review.to_dict())
                continue
        case_reviews.append(build_case_review_stub(replay_case).to_dict())

    improvement_buckets = aggregate_case_review_improvements(case_reviews)

    payload = {
        'mode': 'evaluate',
        'profile': TIKV_SLOW_EVALUATE_PROFILE,
        'scope': {
            'source': ','.join(sources),
            'time_range': getattr(args, 'time_range', TIKV_SLOW_EVALUATE_DEFAULT_TIME_RANGE),
            'assignee': getattr(args, 'assignee', ''),
            'sample_size': sample_size,
            'clinic_only': bool(getattr(args, 'clinic_only', False)),
            'ignore_grafana_only': bool(getattr(args, 'ignore_grafana_only', False)),
            'explicit_case_set': explicit_ids,
            'maintainer_grafana_links': maintainer_grafana_links,
            'maintainer_window_count': len(maintainer_window_specs),
        },
        'included_cases': included,
        'replay_contract': build_replay_contract(
            profile=TIKV_SLOW_EVALUATE_PROFILE,
            mismatch_types=TIKV_SLOW_EVALUATE_MISMATCH_TYPES,
            entry_modes=TIKV_SLOW_EVALUATE_ENTRY_MODES,
            reference_files=TIKV_SLOW_EVALUATE_REFERENCE_FILES,
            skill_conclusion_fields=TIKV_SLOW_EVALUATE_SKILL_CONCLUSION_FIELDS,
            merge_readiness_values=TIKV_SLOW_EVALUATE_MERGE_READINESS_VALUES,
        ).to_dict(),
        'comparison_contract': build_comparison_contract(
            profile=TIKV_SLOW_EVALUATE_PROFILE,
            match_levels=TIKV_SLOW_EVALUATE_MATCH_LEVEL_VALUES,
            mismatch_types=TIKV_SLOW_EVALUATE_MISMATCH_TYPES,
            human_conclusion_status_values=TIKV_SLOW_EVALUATE_HUMAN_CONCLUSION_STATUS_VALUES,
            skill_conclusion_fields=TIKV_SLOW_EVALUATE_SKILL_CONCLUSION_FIELDS,
        ).to_dict(),
        'replay_ready_cases': replay_ready_cases,
        'case_reviews': case_reviews,
        'workflow_improvements': improvement_buckets.get('workflow_improvements') or [],
        'knowledge_improvements': improvement_buckets.get('knowledge_improvements') or [],
        'output_improvements': improvement_buckets.get('output_improvements') or [],
        'do_not_change_yet': improvement_buckets.get('do_not_change_yet') or [],
        'excluded_cases': excluded,
    }
    if getattr(args, 'json_output', False):
        print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
    else:
        sys.stdout.write(render_evaluate_markdown(payload))
    return 0


def summary_cmd(args):
    if getattr(args, 'storage_weekly', False):
        return render_storage_weekly_report_from_args(args)

    if args.issue_ids:
        result = build_explicit_report_blocks(
            args.issue_ids,
            jira_source=make_jira_source_adapter(),
            pingkai_source=make_pingkai_source_adapter(),
            is_pingkai_issue=is_pingkai_issue,
            is_pingkai_report_issue=is_pingkai_report_issue,
            issue_url=issue_url,
            pingkai_base_url=API_BASE_URL,
            render_summary_block_from_raw=render_summary_block_from_raw,
            render_pingkai_summary_block=render_pingkai_summary_block,
        )
        if result.get('returncode'):
            print(result.get('message') or 'failed to build report blocks')
            return result['returncode']
        print('\n\n'.join(result.get('blocks') or []))
        return 0

    try:
        sources = parse_sources(getattr(args, 'source', 'jira'))
    except ValueError as e:
        print(str(e))
        return 1

    result = collect_report_entries(
        args,
        sources,
        jira_source=make_jira_source_adapter(),
        pingkai_source=make_pingkai_source_adapter(),
        build_jira_entries=lambda current_args, jira_source: workflow_build_jira_report_entries(
            current_args,
            jira_source,
            parse_sort_datetime=parse_sort_datetime,
            render_summary_block_from_raw=render_summary_block_from_raw,
        ),
        build_pingkai_entries=lambda current_args, pingkai_source: workflow_build_pingkai_report_entries(
            current_args,
            pingkai_source,
            render_pingkai_summary_block=render_pingkai_summary_block,
        ),
    )
    if result.get('error_json'):
        print(json.dumps(result['error_json'], ensure_ascii=False, indent=2))
        return 0
    if result.get('returncode'):
        sys.stdout.write(result.get('stdout') or '')
        if result.get('stderr'):
            sys.stderr.write(result['stderr'])
        return result['returncode']

    entries = result.get('entries') or []

    if not entries:
        if len(sources) == 1 and sources[0] == 'pingkai':
            print('No Pingkai tickets found for the requested report filters.')
        elif len(sources) == 1 and sources[0] == 'jira':
            print('No Jira issues found for the requested report filters.')
        else:
            print('No Jira or Pingkai tickets found for the requested report filters.')
        return 0

    print('\n\n'.join(item['block'] for item in entries))
    return 0






def current_jira_user_aliases():
    aliases = []
    seen = set()
    try:
        p = run(['jira', 'me'], check=False)
        raw = (p.stdout or '').strip()
    except Exception:
        raw = ''

    def add(v):
        s = str(v or '').strip()
        if not s:
            return
        k = s.lower()
        if k in seen:
            return
        seen.add(k)
        aliases.append(s)

    add(raw)
    if '@' in raw:
        local = raw.split('@', 1)[0]
        if '.' in local:
            parts = [x for x in local.split('.') if x]
        elif '_' in local:
            parts = [x for x in local.split('_') if x]
        elif '-' in local:
            parts = [x for x in local.split('-') if x]
        else:
            parts = []
            m = re.match(r'^([a-z]+?)(mei|yan|xin|lei|liang|hong|wen|yang|gao|zhu)([a-z]*)$', local, flags=re.I)
            if m:
                parts = [m.group(1), m.group(2) + m.group(3)]
            else:
                parts = [local]
        if parts:
            add(' '.join(p.capitalize() for p in parts))
            if len(parts) >= 2:
                add(f'{parts[-1].capitalize()} {parts[0].capitalize()}')
                add(f'{parts[0].capitalize()} {parts[-1].capitalize()}')
    for name in ('JIRA_EMAIL', 'ATLASSIAN_EMAIL', 'JIRA_ACCOUNT_ID', 'ATLASSIAN_ACCOUNT_ID'):
        add(os.environ.get(name))
    if jira_api_available():
        myself = jira_api_json('/rest/api/3/myself')
        if isinstance(myself, dict):
            add(myself.get('emailAddress'))
            add(myself.get('displayName'))
            add(myself.get('accountId'))
    return aliases


def jira_identity_candidates_from_actor(actor):
    candidates = set()
    if not isinstance(actor, dict):
        return candidates
    for key in ('accountId', 'displayName', 'emailAddress', 'email', 'name'):
        value = str(actor.get(key) or '').strip()
        if value:
            candidates.add(value.lower())
    return candidates


def tempo_worklog_author(item: dict):
    tempo_item = (item or {}).get('_tempo') or {}
    for key in ('author', 'worker'):
        actor = tempo_item.get(key)
        if isinstance(actor, dict):
            display_name = str(actor.get('displayName') or actor.get('name') or '').strip()
            if display_name:
                return display_name
    jira_author = ((item or {}).get('author') or {})
    return str(jira_author.get('displayName') or 'N/A').strip()


def tempo_worklog_identity_candidates(item: dict):
    candidates = set()
    tempo_item = (item or {}).get('_tempo') or {}
    for key in ('author', 'worker'):
        candidates.update(jira_identity_candidates_from_actor(tempo_item.get(key)))
    candidates.update(jira_identity_candidates_from_actor((item or {}).get('author') or {}))
    return candidates


def tempo_worklog_started(item: dict):
    tempo_item = (item or {}).get('_tempo') or {}
    start_date = str(tempo_item.get('startDate') or '').strip()
    start_time = str(tempo_item.get('startTime') or '').strip()
    if start_date and start_time:
        return f'{start_date} {start_time}'
    if start_date:
        return start_date
    return str((item or {}).get('started') or 'N/A')


def tempo_worklog_created(item: dict):
    tempo_item = (item or {}).get('_tempo') or {}
    return str(tempo_item.get('createdAt') or (item or {}).get('created') or 'N/A')


def tempo_issue_worklogs(raw: dict):
    if not isinstance(raw, dict):
        return []
    cached = raw.get('_tempo_issue_worklogs')
    if isinstance(cached, list):
        return cached

    issue_id = str(raw.get('id') or '').strip()
    if not issue_id or not tempo_api_available():
        raw['_tempo_issue_worklogs'] = []
        return []

    data = tempo_api_json('/4/worklogs', query={'issueId': issue_id})
    results = data.get('results') if isinstance(data, dict) else []
    raw['_tempo_issue_worklogs'] = results if isinstance(results, list) else []
    return raw['_tempo_issue_worklogs']


def tempo_enriched_worklogs(raw: dict):
    if not isinstance(raw, dict):
        return []
    cached = raw.get('_tempo_enriched_worklogs')
    if isinstance(cached, list):
        return cached

    fields = raw.get('fields') or {}
    worklog = fields.get('worklog') or {}
    worklogs = worklog.get('worklogs') or []
    enriched = []
    for item in worklogs:
        merged = dict(item or {})
        if tempo_api_available():
            tempo_item = tempo_get_worklog_by_jira_id((item or {}).get('id') or '')
            if isinstance(tempo_item, dict):
                merged['_tempo'] = tempo_item
        enriched.append(merged)

    raw['_tempo_enriched_worklogs'] = enriched
    return enriched


def current_jira_user_worklogs(raw: dict):
    aliases = {x.lower() for x in current_jira_user_aliases()}
    tempo_items = tempo_issue_worklogs(raw)
    if tempo_items:
        matched = []
        for item in tempo_items:
            candidates = set()
            candidates.update(jira_identity_candidates_from_actor((item or {}).get('author') or {}))
            candidates.update(jira_identity_candidates_from_actor((item or {}).get('worker') or {}))
            if aliases.intersection(candidates):
                matched.append(item)
        return matched

    matched = []
    for item in tempo_enriched_worklogs(raw):
        if aliases.intersection(tempo_worklog_identity_candidates(item)):
            matched.append(item)
    return matched

def parse_duration_text(value: str) -> str:
    s = (value or '').strip()
    if not s:
        raise ValueError('duration is required')
    compact = re.sub(r'\s+', '', s.lower())
    if re.fullmatch(r'\d+(?:\.\d+)?h', compact):
        num = float(compact[:-1])
        hours = int(num)
        minutes = int(round((num - hours) * 60))
        parts = []
        if hours:
            parts.append(f'{hours}h')
        if minutes:
            parts.append(f'{minutes}m')
        return ' '.join(parts) or '0m'
    parts = re.findall(r'\d+(?:\.\d+)?[dhm]', compact)
    if not parts:
        raise ValueError(f'invalid duration: {value}')
    normalized = []
    for part in parts:
        if part.endswith('h') and '.' in part:
            num = float(part[:-1])
            hours = int(num)
            minutes = int(round((num - hours) * 60))
            if hours:
                normalized.append(f'{hours}h')
            if minutes:
                normalized.append(f'{minutes}m')
        else:
            normalized.append(part)
    return ' '.join(normalized)


def parse_time_to_started(value: str | None) -> str:
    if not value:
        return None
    s = value.strip()
    if re.fullmatch(r'\d{4}-\d{2}-\d{2}', s):
        return f'{s} 09:00:00'
    if re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}', s):
        return f'{s}:00'
    if re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', s):
        return s
    return s


def is_pingkai_issue(value: str) -> bool:
    s = (value or '').strip()
    return bool(re.fullmatch(r'(?i)ticket-\d+', s) or re.fullmatch(r'\d+', s) or re.fullmatch(r'[0-9a-f]{32}', s))


def is_pingkai_report_issue(value: str) -> bool:
    s = (value or '').strip()
    return bool(re.fullmatch(r'(?i)ticket-\d+', s))


def is_pingkai_labor_issue(value: str) -> bool:
    return is_pingkai_report_issue(value)


def parse_pingkai_ref(value: str):
    s = (value or '').strip()
    if re.fullmatch(r'(?i)ticket-(\d+)', s):
        return ('alias', int(s.split('-', 1)[1]))
    if re.fullmatch(r'\d+', s):
        return ('alias', int(s))
    if re.fullmatch(r'[0-9a-f]{32}', s):
        return ('id', s)
    raise ValueError(f'unsupported Pingkai ticket reference: {value}')


def load_pingkai_ticket(ref: str, include_internal: bool = True):
    ref_type, ref_value = parse_pingkai_ref(ref)
    client = TicketAPIClient(verbose=False)
    if not client.login():
        return (None, None)
    ticket = (
        client.get_ticket_by_id(ref_value, include_internal=include_internal)
        if ref_type == 'id'
        else client.get_ticket_by_alias(ref_value, include_internal=include_internal)
    )
    return (client, ticket)


def duration_to_hours(value: str) -> float:
    compact = re.sub(r'\s+', '', (value or '').lower())
    if not compact:
        raise ValueError('duration is required')

    parts = re.findall(r'\d+(?:\.\d+)?[dhm]', compact)
    if not parts or ''.join(parts) != compact:
        raise ValueError(f'invalid duration: {value}')

    hours = 0.0
    for part in parts:
        amount = float(part[:-1])
        unit = part[-1]
        if unit == 'd':
            hours += amount * 24
        elif unit == 'h':
            hours += amount
        elif unit == 'm':
            hours += amount / 60

    hours = round(hours, 2)
    if hours <= 0:
        raise ValueError(f'invalid duration: {value}')
    return int(hours) if hours.is_integer() else hours


def parse_pingkai_working_date(value: str | None) -> str:
    if not value:
        return datetime.now().strftime('%Y-%m-%d')
    s = value.strip()
    if re.fullmatch(r'\d{4}-\d{2}-\d{2}', s):
        return s
    if re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}', s):
        return s.split(' ', 1)[0]
    if re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', s):
        return s.split(' ', 1)[0]
    raise ValueError(f'invalid Pingkai labor date: {value}')


def parse_datetime_loose(value: str):
    s = str(value or '').strip()
    if not s:
        return None
    if s.endswith('Z'):
        s = s[:-1] + '+00:00'
    s = re.sub(r'([+-]\d{2})(\d{2})$', r'\1:\2', s)
    try:
        dt = datetime.fromisoformat(s)
    except ValueError:
        return None
    if dt.tzinfo is None:
        dt = dt.astimezone()
    return dt


def format_datetime_local(value: str) -> str:
    dt = parse_datetime_loose(value)
    if not dt:
        return str(value or 'N/A')
    return dt.astimezone().strftime('%Y-%m-%d %H:%M:%S %z')


def round_hours_to_half(value: float) -> float:
    return max(0.0, round(value * 2) / 2.0)


def hours_to_duration_text(value: float) -> str:
    rounded = round_hours_to_half(value)
    if rounded <= 0:
        return '0m'
    hours = int(rounded)
    minutes = int(round((rounded - hours) * 60))
    parts = []
    if hours:
        parts.append(f'{hours}h')
    if minutes:
        parts.append(f'{minutes}m')
    return ''.join(parts) if parts else '0m'


def seconds_to_duration_text(value) -> str:
    try:
        seconds = int(float(value or 0))
    except (TypeError, ValueError):
        return 'N/A'
    if seconds <= 0:
        return '0m'
    hours, remainder = divmod(seconds, 3600)
    minutes = remainder // 60
    parts = []
    if hours:
        parts.append(f'{hours}h')
    if minutes:
        parts.append(f'{minutes}m')
    return ''.join(parts) if parts else '0m'


def current_jira_display_name() -> str:
    myself = jira_api_json('/rest/api/3/myself') if jira_api_available() else None
    if isinstance(myself, dict):
        name = str(myself.get('displayName') or '').strip()
        if name:
            return name
    aliases = current_jira_user_aliases()
    for alias in aliases:
        text = str(alias or '').strip()
        if text and '@' not in text and ':' not in text:
            return text
    return aliases[0] if aliases else 'N/A'


def estimate_comment_hours(text: str) -> float:
    body = str(text or '').strip()
    if not body:
        return 0.0
    char_count = len(body)
    line_count = max(1, body.count('\n') + 1)
    extra = 0
    if re.search(r'https?://|curl |tiup |select |update |insert |delete |alter ', body, flags=re.I):
        extra += 80
    if re.search(r'error|exception|timeout|panic|stack|trace|监控|日志|参数|配置', body, flags=re.I):
        extra += 60
    score = char_count + min(line_count, 8) * 12 + extra
    if score <= 80:
        return 0.5
    if score <= 220:
        return 1.0
    if score <= 420:
        return 1.5
    if score <= 700:
        return 2.0
    return 3.0


def current_jira_owner_comment_entries(raw: dict):
    fields = (raw or {}).get('fields', {})
    current_aliases = {x.lower() for x in current_jira_user_aliases()}
    entries = []
    comments = (((fields.get('comment') or {}).get('comments')) or [])
    for entry, comment in zip(extract_comment_entries(raw), comments):
        author = (entry.get('author') or '').strip()
        if not author or author.lower() not in current_aliases:
            continue
        entries.append({
            'created': (comment or {}).get('created') or '',
            'author': author,
            'body': entry.get('body', ''),
        })
    return entries


def jira_current_user_logged_hours_by_date(raw: dict):
    totals = {}
    for item in current_jira_user_worklogs(raw):
        if 'startDate' in (item or {}) or 'startTime' in (item or {}):
            started_text = f"{str((item or {}).get('startDate') or '').strip()}T{str((item or {}).get('startTime') or '').strip()}".strip('T')
            started = parse_datetime_loose(started_text)
            spent_seconds = (item or {}).get('timeSpentSeconds')
            spent = round(float(spent_seconds or 0) / 3600, 2) if spent_seconds not in (None, '') else 0.0
        else:
            started = parse_datetime_loose(str(tempo_worklog_started(item)).replace(' ', 'T', 1))
            if not started:
                started = parse_datetime_loose((item or {}).get('started') or '')
            spent = duration_to_hours((item or {}).get('timeSpent') or '')
        if not started:
            continue
        date_key = started.astimezone().strftime('%Y-%m-%d')
        totals[date_key] = totals.get(date_key, 0.0) + float(spent)
    return totals


def current_pingkai_owner_comment_entries(ticket: dict, client: TicketAPIClient):
    aliases = {x.lower() for x in current_pingkai_user_aliases(client)}
    entries = []
    for item in (ticket.get('comments') or []):
        if item.get('is_system_generated'):
            continue
        author = ((item.get('author') or {}).get('full_name') or item.get('author_name') or '').strip()
        author_username = ((item.get('author') or {}).get('username') or '').strip()
        author_email = ((item.get('author') or {}).get('email') or '').strip()
        author_id = ((item.get('author') or {}).get('id') or item.get('author_id') or '').strip()
        candidates = {
            x.lower()
            for x in [author, author_username, author_email, author_id]
            if str(x or '').strip()
        }
        if not aliases.intersection(candidates):
            continue
        entries.append({
            'created': item.get('created_at') or '',
            'author': author or author_username or 'N/A',
            'body': strip_text_for_pingkai_comment(item.get('content', '')),
        })
    return entries


def pingkai_current_user_logged_hours_by_date(result: dict, client: TicketAPIClient):
    aliases = {x.lower() for x in current_pingkai_user_aliases(client)}
    totals = {}
    for item in pingkai_working_times(result):
        author_candidates = {
            str(x or '').strip().lower()
            for x in [
                pingkai_working_time_field(item, 'user_name', default=''),
                pingkai_working_time_field(item, 'author_name', default=''),
                pingkai_working_time_field(item, 'creator_name', default=''),
                pingkai_working_time_field(item, 'created_by_name', default=''),
            ]
            if str(x or '').strip()
        }
        if aliases and not aliases.intersection(author_candidates):
            continue
        date_key = str(pingkai_working_time_field(item, 'date', 'work_date', 'working_date', default='')).strip()
        if not date_key:
            continue
        spent = pingkai_working_time_field(item, 'working_time', 'duration', 'hours', default=0)
        try:
            spent_hours = float(spent)
        except (TypeError, ValueError):
            continue
        totals[date_key] = totals.get(date_key, 0.0) + spent_hours
    return totals


def build_labor_suggestions(comment_entries, logged_hours_by_date: dict, issue_ref: str, pingkai: bool = False):
    if not comment_entries:
        return ['No labor add suggestions.']

    grouped = {}
    for item in comment_entries:
        created = parse_datetime_loose(item.get('created', ''))
        if not created:
            continue
        local_dt = created.astimezone()
        date_key = local_dt.strftime('%Y-%m-%d')
        grouped.setdefault(date_key, []).append({
            'datetime': local_dt,
            'body': str(item.get('body') or ''),
        })

    suggestions = []
    for date_key in sorted(grouped):
        entries = grouped[date_key]
        entries.sort(key=lambda item: item['datetime'])
        estimated = sum(estimate_comment_hours(item['body']) for item in entries)
        estimated = min(round_hours_to_half(estimated), 4.0)
        logged = round_hours_to_half(float(logged_hours_by_date.get(date_key, 0.0)))
        delta = round_hours_to_half(max(0.0, estimated - logged))
        if delta <= 0:
            continue

        first_time = entries[0]['datetime']
        char_total = sum(len(item['body']) for item in entries)
        if pingkai:
            command = f'oncall-toolkit labor {issue_ref} {hours_to_duration_text(delta)} --time "{date_key}" -m "follow-up based on owner comments"'
        else:
            started = first_time.strftime('%Y-%m-%d %H:%M:%S')
            command = f'oncall-toolkit labor {issue_ref} {hours_to_duration_text(delta)} --time "{started}" -m "follow-up based on owner comments"'
        suggestions.append(
            f'- {first_time.strftime("%Y-%m-%d %H:%M:%S %z")} | duration={hours_to_duration_text(delta)} | '
            f'basis={len(entries)} comments / {char_total} chars | command={command}'
        )

    return suggestions or ['No labor add suggestions.']


def render_worklog_lines(raw: dict):
    worklogs = current_jira_user_worklogs(raw)
    if not worklogs:
        return ['No labor records found.']
    lines = []
    for item in worklogs:
        if 'tempoWorklogId' in (item or {}):
            author_info = (item or {}).get('author') or {}
            author = (
                author_info.get('displayName')
                or (
                    current_jira_display_name()
                    if str(author_info.get('accountId') or '').strip().lower() in {x.lower() for x in current_jira_user_aliases()}
                    else author_info.get('accountId')
                )
                or 'N/A'
            )
            start_date = str((item or {}).get('startDate') or '').strip()
            start_time = str((item or {}).get('startTime') or '').strip()
            started = f'{start_date} {start_time}'.strip() or 'N/A'
            created = format_datetime_local((item or {}).get('createdAt') or 'N/A')
            spent = seconds_to_duration_text((item or {}).get('timeSpentSeconds'))
            comment = compact_summary(str((item or {}).get('description') or ''), 120) or 'N/A'
        else:
            author = tempo_worklog_author(item) or 'N/A'
            started = tempo_worklog_started(item) or 'N/A'
            created = format_datetime_local(tempo_worklog_created(item) or 'N/A')
            spent = (item or {}).get('timeSpent') or 'N/A'
            comment = compact_summary(extract_text((item or {}).get('comment')), 120) or 'N/A'
        lines.append(f'- started={started} | spent={spent} | author={author} | created_at={created} | comment={comment}')
    return lines


def render_owner_comment_labor_lines(raw: dict):
    fields = (raw or {}).get('fields', {})
    current_aliases = {x.lower() for x in current_jira_user_aliases()}
    lines = []
    for entry, comment in zip(extract_comment_entries(raw), (((fields.get('comment') or {}).get('comments')) or [])):
        author = (entry.get('author') or '').strip()
        if not author or author.lower() not in current_aliases:
            continue
        created = (comment or {}).get('created') or 'N/A'
        summary = compact_summary(entry.get('body', ''), 120) or 'N/A'
        lines.append(f'- {created} | owner={author} | comment={summary}')
    return lines or ['No owner comments found for current Jira user.']


def pingkai_working_times(result: dict):
    data = (result or {}).get('data') or {}
    items = data.get('working_times') or data.get('items') or []
    return items if isinstance(items, list) else []


def pingkai_working_time_field(item: dict, *names, default=''):
    for name in names:
        value = (item or {}).get(name)
        if value not in (None, ''):
            return value
    return default


def render_pingkai_worklog_lines(result: dict):
    items = pingkai_working_times(result)
    if not items:
        return ['No labor records found.']

    lines = []
    for item in items:
        work_date = pingkai_working_time_field(item, 'date', 'work_date', 'working_date', default='N/A')
        spent = pingkai_working_time_field(item, 'working_time', 'duration', 'hours', default='N/A')
        spent_text = str(spent)
        if spent_text != 'N/A' and not re.search(r'[a-zA-Z]$', spent_text):
            spent_text = f'{spent_text}h'
        author = pingkai_working_time_field(
            item,
            'user_name',
            'author_name',
            'creator_name',
            'created_by_name',
            default='N/A',
        )
        comment = compact_summary(
            str(pingkai_working_time_field(item, 'comment', 'remark', 'description', default='')),
            120,
        ) or 'N/A'
        lines.append(f'- {work_date} | {spent_text} | author={author} | comment={comment}')
    return lines


def current_pingkai_user_aliases(client: TicketAPIClient):
    user = getattr(client, 'user_info', {}) or {}
    aliases = []
    seen = set()

    def add(value):
        s = str(value or '').strip()
        if not s:
            return
        key = s.lower()
        if key in seen:
            return
        seen.add(key)
        aliases.append(s)

    add(user.get('id'))
    add(user.get('username'))
    add(user.get('email'))
    add(user.get('full_name'))
    return aliases


def render_pingkai_owner_comment_lines(ticket: dict, client: TicketAPIClient):
    aliases = {x.lower() for x in current_pingkai_user_aliases(client)}
    comments = (ticket or {}).get('comments') or []
    lines = []
    for item in comments:
        author = ((item.get('author') or {}).get('full_name') or item.get('author_name') or '').strip()
        author_username = ((item.get('author') or {}).get('username') or '').strip()
        author_email = ((item.get('author') or {}).get('email') or '').strip()
        author_id = ((item.get('author') or {}).get('id') or item.get('author_id') or '').strip()
        candidates = {
            x.lower()
            for x in [author, author_username, author_email, author_id]
            if str(x or '').strip()
        }
        if not aliases.intersection(candidates):
            continue
        created = item.get('created_at') or 'N/A'
        summary = compact_summary(strip_text_for_pingkai_comment(item.get('content', '')), 120) or 'N/A'
        lines.append(f'- {created} | owner={author or author_username or "N/A"} | comment={summary}')
    return lines or ['No owner comments found for current Pingkai user.']


def strip_text_for_pingkai_comment(value):
    text = str(value or '')
    text = re.sub(r'<script[^>]*>.*?</script>', '', text, flags=re.DOTALL | re.IGNORECASE)
    text = re.sub(r'<style[^>]*>.*?</style>', '', text, flags=re.DOTALL | re.IGNORECASE)
    text = re.sub(r'<br\s*/?>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'</p>', '\n\n', text, flags=re.IGNORECASE)
    text = re.sub(r'</div>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'</li>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'<[^>]+>', '', text)
    return re.sub(r'\s+', ' ', text).strip()


def labor_cmd(args):
    if is_pingkai_issue(args.issue) and not is_pingkai_labor_issue(args.issue):
        print(f'unsupported Pingkai labor reference: {args.issue}. Use ticket-xxxx.')
        return 1

    if is_pingkai_labor_issue(args.issue):
        try:
            ref_type, ref_value = parse_pingkai_ref(args.issue)
        except ValueError as e:
            print(str(e))
            return 1

        try:
            client = TicketAPIClient(verbose=False)
            if not client.login():
                return 1
            ticket = (
                client.get_ticket_by_id(ref_value, include_internal=True)
                if ref_type == 'id'
                else client.get_ticket_by_alias(ref_value, include_internal=True)
            )
        except RuntimeError as e:
            print(str(e))
            return 1

        if not ticket:
            print(f'Failed to load Pingkai ticket: {args.issue}')
            return 1

        alias = ticket.get('alias') or args.issue
        title = ticket.get('title') or 'N/A'

        if args.action == 'show':
            result = client.get_working_times(ticket['id'])
            if not result:
                return 1
            print(f'### Labor ticket-{alias} - {title}')
            print('- Worklogs:')
            for line in render_pingkai_worklog_lines(result):
                print(line)
            print('- Owner comments:')
            for line in render_pingkai_owner_comment_lines(ticket, client):
                print(line)
            print()
            print('### Suggested labor add')
            for line in build_labor_suggestions(
                current_pingkai_owner_comment_entries(ticket, client),
                pingkai_current_user_logged_hours_by_date(result, client),
                f'ticket-{alias}',
                pingkai=True,
            ):
                print(line)
            return 0

        try:
            working_time = duration_to_hours(args.duration)
            work_date = parse_pingkai_working_date(args.time)
        except ValueError as e:
            print(str(e))
            return 1

        payload = {
            'ticket_ref': args.issue,
            'date': work_date,
            'working_time': working_time,
            'comment': args.comment or '',
        }

        if args.dry_run:
            print(json.dumps({
                'mode': 'pingkai-labor-add',
                'request': payload,
            }, ensure_ascii=False, indent=2))
            return 0

        result = client.add_working_time(
            ticket_id=ticket['id'],
            date=work_date,
            working_time=working_time,
            comment=args.comment or '',
        )

        if not result:
            return 1

        print(f'### Labor ticket-{alias} - {title}')
        print(f'- date={work_date} | working_time={working_time} | comment={args.comment or ""}')
        print('- add working time: success')
        return 0

    if args.action == 'show':
        raw = get_issue_raw(args.issue)
        if not raw:
            print(f'Failed to load Jira issue: {args.issue}')
            return 1
        fields = (raw or {}).get('fields', {})
        title = fields.get('summary') or args.issue
        print(f'### Labor {args.issue} - {title}')
        print('- Worklogs:')
        for line in render_worklog_lines(raw):
            print(line)
        print('- Owner comments:')
        for line in render_owner_comment_labor_lines(raw):
            print(line)
        print()
        print('### Suggested labor add')
        for line in build_labor_suggestions(
            current_jira_owner_comment_entries(raw),
            jira_current_user_logged_hours_by_date(raw),
            args.issue,
            pingkai=False,
        ):
            print(line)
        return 0

    try:
        duration = parse_duration_text(args.duration)
    except ValueError as e:
        print(str(e))
        return 1

    comment = args.comment or 'time-tracking'
    cmd = ['jira', 'issue', 'worklog', 'add', args.issue, duration, '--comment', comment, '--no-input']
    started = parse_time_to_started(args.time)
    if started:
        cmd.extend(['--started', started, '--timezone', 'Asia/Shanghai'])
    if args.dry_run:
        print(json.dumps({
            'mode': 'jira-labor-add',
            'command': cmd,
        }, ensure_ascii=False, indent=2))
        return 0
    p = run(cmd, check=False)
    sys.stdout.write(p.stdout)
    if p.stderr:
        sys.stderr.write(p.stderr)
    return p.returncode



def extract_comment_entries_with_time(raw: dict):
    comments = (((raw or {}).get('fields') or {}).get('comment') or {}).get('comments') or []
    entries = []
    for c in comments:
        body = extract_text(c.get('body'))
        author = ((c.get('author') or {}).get('displayName') or '').strip()
        created = (c.get('created') or '').strip()
        if body.strip():
            entries.append({
                'author': author,
                'body': body.strip(),
                'created': created,
            })
    return entries


def rca_summary_from_issue(raw: dict, key: str) -> str:
    fields = (raw or {}).get('fields', {})
    title = fields.get('summary') or key
    priority = derive_priority(fields)
    version = derive_version(fields)
    desc = summarize_description_for_phenomenon(extract_text(fields.get('description')))
    return '\n'.join([
        f'- Issue: [{key}]({issue_url(key)})',
        f'- Title: {title}',
        f'- Severity: {priority}',
        f'- Version: {version}',
        f'- User impact: {desc}',
    ])


def rca_timeline_from_issue(raw: dict) -> str:
    fields = (raw or {}).get('fields', {})
    desc = compact_summary(extract_text(fields.get('description')), 300) or 'TBD'
    comments = extract_comment_entries_with_time(raw)

    trigger = []
    detection = []
    diagnosis = []
    mitigation = []
    recovery = []
    followup = []

    if desc and desc != 'TBD':
        trigger.append(f'- Description: {desc}')

    owner_names = {a.lower() for a in author_set(fields)}

    for item in comments:
        body = compact_summary(item['body'], 220) or 'TBD'
        ts = item['created'] or 'N/A'
        author = item['author'] or 'N/A'
        line = f'- {ts} | {author}: {body}'
        ll = item['body'].lower()
        is_owner = author.lower() in owner_names
        is_question = 'question 1' in ll or 'question 2' in ll or 'question 3' in ll or ll.startswith('thank you')

        if any(x in ll for x in ['504', 'detected', 'observed', 'application error', 'timeout']) and not is_owner:
            detection.append(line)
        elif any(x in ll for x in ['root cause', 'split vote', 'packet loss', 'network abnormal', 'higher term', 'step down', 'no region leader']) and is_owner and not is_question:
            diagnosis.append(line)
        elif any(x in ll for x in ['mitigation', 'workaround', 'timeout value', 'max-leader-lease', '15s', '12s']) and is_owner:
            mitigation.append(line)
        elif any(x in ll for x in ['recovered', 'recovery', 'not detected any 504 errors']) and not is_owner:
            recovery.append(line)
        elif any(x in ll for x in ['investigate', 'follow up', 'need some time', 'still need some time']) or is_question:
            followup.append(line)

    def block(title, items):
        if not items:
            items = ['- TBD']
        return f'#### {title}\n\n' + '\n'.join(items)

    return '\n\n'.join([
        block('Trigger', trigger),
        block('Detection', detection),
        block('Diagnosis', diagnosis),
        block('Mitigation', mitigation),
        block('Recovery', recovery),
        block('Postmortem / Follow-up', followup),
    ])

def rca_analysis_sections(raw: dict):
    fields = (raw or {}).get('fields', {})
    comments = extract_comment_entries_with_time(raw)
    owner_names = {a.lower() for a in author_set(fields)}
    analysis = []
    conclusion = 'TBD'
    solution = []
    other = []

    root_candidates = []
    solution_candidates = []
    followup_candidates = []

    for item in comments:
        raw_body = item['body']
        body = compact_summary(raw_body, 240) or 'TBD'
        ll = raw_body.lower()
        author = item.get('author') or 'N/A'
        ts = item.get('created') or 'N/A'
        is_owner = author.lower() in owner_names
        is_question = 'question 1' in ll or 'question 2' in ll or 'question 3' in ll or ll.startswith('thank you')
        line = f'- {ts} | {author}: {body}'

        if is_owner and not is_question and any(x in ll for x in ['root cause', 'split vote', 'packet loss', 'network abnormal', 'higher term', 'step down', 'no region leader', 're-election']):
            root_candidates.append((ts, body, line))
        elif is_owner and any(x in ll for x in ['mitigation', 'workaround', 'timeout value', 'max-leader-lease', '15s', '12s']):
            solution_candidates.append((ts, body, line))
        elif any(x in ll for x in ['investigate', 'follow up', 'need some time', 'still need some time']):
            followup_candidates.append((ts, body, line))

    if root_candidates:
        analysis = [x[2] for x in root_candidates]
        conclusion = root_candidates[-1][1]
    else:
        root_cause, workaround, todo = summarize_owner_comments('\n'.join(c['body'] for c in comments))
        conclusion = root_cause or 'TBD'
        if conclusion != 'TBD':
            analysis = [f'- {conclusion}']
        if workaround and workaround != 'None yet':
            solution_candidates.append(('fallback', workaround, f'- {workaround}'))
        if todo and todo != 'TBD':
            followup_candidates.append(('fallback', todo, f'- {todo}'))

    if solution_candidates:
        solution = [solution_candidates[-1][2]]
    else:
        solution = ['- TBD']

    if followup_candidates:
        other = [followup_candidates[-1][2]]
    else:
        other = ['- TBD']

    if not analysis:
        analysis = ['- TBD']

    return analysis, conclusion, solution, other

def build_rca_markdown(raw: dict, key: str) -> str:
    fields = (raw or {}).get('fields', {})
    assignee = ((fields.get('assignee') or {}).get('displayName') or 'TBD').strip()
    l2_assignee = ((fields.get('customfield_10053') or {}).get('displayName') or 'TBD').strip()
    analysis, conclusion, solution, other = rca_analysis_sections(raw)
    timeline = rca_timeline_from_issue(raw)

    summary_block = '## Summary\n\n' + rca_summary_from_issue(raw, key)
    timeline_block = '## Timeline\n\n' + timeline
    analysis_block = '## Root Cause Analysis\n\n### Analysis\n\n' + '\n'.join(analysis) + '\n\n### Conclusion\n\n' + conclusion + '\n\n### Other\n\n' + '\n'.join(other)
    solution_block = '## Solution\n\n' + '\n'.join(solution)
    postmortem_table = (
        '## TCOC Postmortem\n\n'
        '<lark-table header-row="true">\n'
        '| Item | Value |\n'
        '|---|---|\n'
        f'| Assignee | {assignee} |\n'
        f'| L2 assignee | {l2_assignee} |\n'
        f'| Root cause | {conclusion.replace("|", "/")} |\n'
        f'| Follow-up | {("; ".join(x.lstrip("- ").strip() for x in other) or "TBD").replace("|", "/")} |\n'
        '</lark-table>'
    )
    header_table = (
        '<lark-table header-row="true">\n'
        '| Field | Value |\n'
        '|---|---|\n'
        '| Author | AI |\n'
        '| Reviewer | TBD |\n'
        '| Last updated | TBD |\n'
        '</lark-table>'
    )
    return '\n\n'.join([
        header_table,
        summary_block,
        timeline_block,
        analysis_block,
        solution_block,
        postmortem_table,
    ])


def get_issue_status_info(raw: dict):
    fields = (raw or {}).get('fields', {})
    status = fields.get('status') or {}
    category = (status.get('statusCategory') or {}) if isinstance(status, dict) else {}
    return {
        'name': str(status.get('name') or 'Unknown').strip(),
        'category': str(category.get('name') or '').strip(),
        'category_key': str(category.get('key') or '').strip(),
    }


def issue_project_key(issue_id: str) -> str:
    return str(issue_id or '').split('-', 1)[0].upper()


def is_supported_close_project(issue_id: str) -> bool:
    return issue_project_key(issue_id) in {'TCOC', 'GTOC'}


def jira_issue_transitions(issue_id: str):
    data = jira_api_json(
        f'/rest/api/3/issue/{quote(issue_id)}/transitions',
        query={'expand': 'transitions.fields'},
    )
    transitions = (data or {}).get('transitions') or []
    return transitions if isinstance(transitions, list) else []


def transition_label(transition: dict) -> str:
    to_status = (transition.get('to') or {}).get('name') or ''
    name = str(transition.get('name') or '').strip()
    to_status = str(to_status).strip()
    if name and to_status and name.lower() != to_status.lower():
        return f'{name} -> {to_status}'
    return name or to_status or 'unknown transition'


def find_transition(transitions, preferred_names=(), preferred_to_status=(), preferred_to_category_keys=()):
    name_set = {str(x).strip().lower() for x in preferred_names if str(x).strip()}
    status_set = {str(x).strip().lower() for x in preferred_to_status if str(x).strip()}
    category_set = {str(x).strip().lower() for x in preferred_to_category_keys if str(x).strip()}
    for transition in transitions:
        name = str(transition.get('name') or '').strip().lower()
        to_status = str(((transition.get('to') or {}).get('name') or '')).strip().lower()
        to_category = str((((transition.get('to') or {}).get('statusCategory') or {}).get('key') or '')).strip().lower()
        if name_set and name in name_set:
            return transition
        if status_set and to_status in status_set:
            return transition
        if category_set and to_category in category_set:
            return transition
    return None


def close_needs_under_investigation(status_info: dict) -> bool:
    status_name = str((status_info or {}).get('name') or '').strip().lower()
    category_key = str((status_info or {}).get('category_key') or '').strip().lower()
    return status_name in {'pending for support', 'to do'} or category_key == 'new'


def close_field_drafts(raw: dict):
    fields = (raw or {}).get('fields', {})
    description_text = extract_text(fields.get('description'))
    owner_comment_text = collect_owner_comment_text(raw, fields)
    phenomenon = summarize_description_for_phenomenon(description_text)
    root_cause, workaround, _todo = summarize_owner_comments(owner_comment_text)
    return [
        f'resolution-path = Resolved',
        f'Troubleshooting = {phenomenon}',
        f'Rediscovery = {phenomenon}',
        'Is product issue? = No',
        'Fix versions = ',
        'Github Issue(s) = ',
        f'Root Cause (migrated) = {root_cause}',
        f'Workaround (migrated) = {workaround}',
        f'close comment body = {DEFAULT_CLOSE_COMMENT}',
    ]


def build_close_review(issue_id: str, raw: dict, transitions):
    status_info = get_issue_status_info(raw)
    path = []
    open_questions = []
    if close_needs_under_investigation(status_info):
        intermediate = find_transition(
            transitions,
            preferred_names=('under investigation',),
            preferred_to_status=('under investigation',),
        )
        if intermediate:
            path.append(transition_label(intermediate))
        else:
            open_questions.append('Missing transition to UNDER INVESTIGATION.')

    final_transition = find_transition(
        transitions,
        preferred_names=('resolved', 'resolve issue'),
        preferred_to_status=('resolved',),
        preferred_to_category_keys=('done',),
    )
    if final_transition:
        path.append(transition_label(final_transition))
    else:
        open_questions.append('No resolve path available.')

    proposed_transition = ' | '.join(path) if path else 'waiting for user confirmation'
    summary_block = render_summary_block_from_raw(issue_id, raw).replace('### ', '', 1)
    lines = [
        f'### Review {issue_id}',
        f'- Issue summary: {summary_block}',
        f'- Proposed transition: {proposed_transition}',
        f'- Fields to fill: {"; ".join(close_field_drafts(raw))}',
        f'- Draft close comment: {DEFAULT_CLOSE_COMMENT}',
        f'- Open questions: {"; ".join(open_questions) if open_questions else "None."}',
    ]
    return '\n'.join(lines)


def jira_transition_issue(issue_id: str, transition_id: str, fields: dict | None = None):
    payload = {'transition': {'id': str(transition_id)}}
    if fields:
        payload['fields'] = fields
    return jira_api_request('POST', f'/rest/api/3/issue/{quote(issue_id)}/transitions', payload=payload)


def jira_add_comment(issue_id: str, comment: str):
    return jira_api_request(
        'POST',
        f'/rest/api/3/issue/{quote(issue_id)}/comment',
        payload={'body': comment},
    )


def execute_close_issue(issue_id: str):
    if not jira_api_available():
        print(f'- [{issue_id}]({issue_url(issue_id)}) | missing Jira API credentials')
        return 1

    raw = get_issue_raw(issue_id)
    if not raw:
        print(f'- [{issue_id}]({issue_url(issue_id)}) | failed to load issue')
        return 1

    if not is_supported_close_project(issue_id):
        print(f'- [{issue_id}]({issue_url(issue_id)}) | unsupported project')
        return 1

    status_info = get_issue_status_info(raw)
    if status_info.get('category_key', '').lower() == 'done':
        print(f'- [{issue_id}]({issue_url(issue_id)}) | already closed')
        return 0

    transitions = jira_issue_transitions(issue_id)
    path = []

    if close_needs_under_investigation(status_info):
        intermediate = find_transition(
            transitions,
            preferred_names=('under investigation',),
            preferred_to_status=('under investigation',),
        )
        if not intermediate:
            print(f'- [{issue_id}]({issue_url(issue_id)}) | no resolve path available')
            return 1
        jira_transition_issue(issue_id, intermediate.get('id'))
        path.append(transition_label(intermediate))
        transitions = jira_issue_transitions(issue_id)

    final_transition = find_transition(
        transitions,
        preferred_names=('resolved', 'resolve issue'),
        preferred_to_status=('resolved',),
        preferred_to_category_keys=('done',),
    )
    if not final_transition:
        print(f'- [{issue_id}]({issue_url(issue_id)}) | no resolve path available')
        return 1

    jira_add_comment(issue_id, DEFAULT_CLOSE_COMMENT)
    jira_transition_issue(issue_id, final_transition.get('id'))
    path.append(transition_label(final_transition))
    print(f'- [{issue_id}]({issue_url(issue_id)}) | {" | ".join(path)}')
    return 0

def close_cmd(args):
    issues = split_issue_inputs(args.issues)
    preview = not args.no_preview
    if preview:
        blocks = []
        exit_code = 0
        for issue_id in issues:
            if not is_supported_close_project(issue_id):
                blocks.append('\n'.join([
                    f'### Review {issue_id}',
                    '- Issue summary: unsupported project',
                    '- Proposed transition: waiting for user confirmation',
                    '- Fields to fill: resolution-path = TBD',
                    '- Draft close comment: TBD',
                    '- Open questions: Non-TCOC/GTOC issue requires explicit user confirmation.',
                ]))
                exit_code = 1
                continue

            raw = get_issue_raw(issue_id)
            if not raw:
                blocks.append('\n'.join([
                    f'### Review {issue_id}',
                    '- Issue summary: failed to load issue',
                    '- Proposed transition: waiting for user confirmation',
                    '- Fields to fill: resolution-path = TBD',
                    '- Draft close comment: TBD',
                    '- Open questions: Jira issue data is unavailable.',
                ]))
                exit_code = 1
                continue

            status_info = get_issue_status_info(raw)
            if status_info.get('category_key', '').lower() == 'done':
                blocks.append('\n'.join([
                    f'### Review {issue_id}',
                    '- Issue summary: already closed',
                    '- Proposed transition: already closed',
                    '- Fields to fill: resolution-path = none',
                    '- Draft close comment: N/A',
                    '- Open questions: None.',
                ]))
                continue

            blocks.append(build_close_review(issue_id, raw, jira_issue_transitions(issue_id)))

        print('\n\n'.join(blocks))
        return exit_code

    rc = 0
    for issue_id in issues:
        result = execute_close_issue(issue_id)
        if result != 0:
            rc = result
    return rc


def rca_cmd(args):
    raw = get_issue_raw(args.issue)
    if not raw:
        print(f'Failed to load Jira issue: {args.issue}')
        return 1

    markdown = build_rca_markdown(raw, args.issue)
    title = f'RCA {args.issue} by AI'
    rca_dir = (args.rca_link_dir or DEFAULT_RCA_LINK_DIR).strip()

    print(json.dumps({
        'mode': 'RCA',
        'issue': args.issue,
        'title': title,
        'dir': rca_dir,
        'markdown': markdown,
    }, ensure_ascii=False, indent=2))
    return 0


CAPABILITY_LAYERS = (
    (
        'discovery',
        'discover candidate incidents, filter sources, and build evaluator-ready bundles',
        ('list', 'evaluate'),
    ),
    (
        'diagnosis',
        'analyze one incident or monitoring link and route it through diagnosis evidence workflows',
        ('analyze',),
    ),
    (
        'reporting',
        'render human-readable summaries, RCA payloads, and weekly reporting outputs',
        ('report', 'RCA'),
    ),
    (
        'execution',
        'execute or preview operational actions such as close flows and labor updates',
        ('close', 'labor'),
    ),
)


def capability_layer_description() -> str:
    lines = ['Capability layers:']
    for name, desc, commands in CAPABILITY_LAYERS:
        lines.append(f'- {name}: {desc} ({", ".join(commands)})')
    return '\n'.join(lines)


def add_discovery_commands(sub):
    p_list = sub.add_parser('list', help='list Jira issues, Pingkai tickets, or both')
    p_list.add_argument(
        '--source',
        default='jira',
        help='ticket source: jira, pingkai, or a comma-separated combination such as pingkai,jira (default: jira)',
    )
    p_list.add_argument('--status', default='all', help='jira status filter; pingkai currently supports all only')
    p_list.add_argument(
        '--components',
        default='tikv,tidb,pd',
        help='jira components filter; for pingkai this matches module_name',
    )
    p_list.add_argument('--time_range', default='1w', help='relative time range such as 1w, 7d, 10d, this_week')
    p_list.add_argument(
        '--assignee',
        default='',
        help='jira assignee filter; use stable Jira identifiers such as email, English display name, account id, or currentUser(); for pingkai this matches l2_assignee or l3_assignee',
    )
    p_list.add_argument('--page_size', type=int, default=100, help='pingkai page size (1-100)')
    p_list.add_argument('--module', default='', help=argparse.SUPPRESS)
    p_list.add_argument('--l2_assignee', default='', help=argparse.SUPPRESS)
    p_list.add_argument('--l3_assignee', default='', help=argparse.SUPPRESS)
    p_list.set_defaults(func=list_cmd)

    p_evaluate = sub.add_parser('evaluate', help='build an evaluator-ready case set for a diagnosis profile')
    p_evaluate.add_argument('profile', help='evaluation profile, for example tikv-slow')
    p_evaluate.add_argument('issue_ids', nargs='*', help='optional explicit issue ids such as GTOC-8286 TICKET-8412')
    p_evaluate.add_argument(
        '--source',
        default=TIKV_SLOW_EVALUATE_DEFAULT_SOURCE,
        help='ticket source: jira, pingkai, or a comma-separated combination such as jira,pingkai (default: jira,pingkai)',
    )
    p_evaluate.add_argument('--time_range', default=TIKV_SLOW_EVALUATE_DEFAULT_TIME_RANGE, help='relative time range such as 14d, 30d, this_week')
    p_evaluate.add_argument(
        '--assignee',
        default=TIKV_SLOW_EVALUATE_DEFAULT_ASSIGNEE,
        help='candidate owner filter for discovery mode; Jira uses stable identifiers such as email, English display name, account id, or currentUser(), while Pingkai matches Chinese l2/l3 names',
    )
    p_evaluate.add_argument('--sample-size', type=int, default=TIKV_SLOW_EVALUATE_DEFAULT_SAMPLE_SIZE, help='maximum included cases (default: 5)')
    p_evaluate.add_argument('--status', default=TIKV_SLOW_EVALUATE_DEFAULT_STATUS, help='jira status filter for discovery mode')
    p_evaluate.add_argument('--page_size', type=int, default=TIKV_SLOW_EVALUATE_DEFAULT_PAGE_SIZE, help='pingkai page size for discovery mode')
    p_evaluate.add_argument('--clinic-only', action='store_true', help='include only cases whose initial ticket body contains a Clinic link; enabled by default for tikv-slow')
    p_evaluate.add_argument('--ignore-grafana-only', action='store_true', help='exclude cases that have Grafana links but no Clinic link in the initial ticket body; enabled by default for tikv-slow')
    p_evaluate.add_argument(
        '--maintainer-grafana-link',
        action='append',
        default=[],
        help='maintainer-only evidence assist: attach one or more explicit Grafana links to explicit-case review without changing the replay boundary',
    )
    p_evaluate.add_argument(
        '--maintainer-window',
        action='append',
        default=[],
        help='maintainer-only window hint in the format label|start_time|end_time|summary|inferred_path|inferred_bottleneck',
    )
    p_evaluate.add_argument('--json', dest='json_output', action='store_true', help='emit the structured evaluation bundle instead of markdown')
    p_evaluate.set_defaults(
        func=evaluate_cmd,
        clinic_only=TIKV_SLOW_EVALUATE_DEFAULT_CLINIC_ONLY,
        ignore_grafana_only=TIKV_SLOW_EVALUATE_DEFAULT_IGNORE_GRAFANA_ONLY,
    )


def add_diagnosis_commands(sub):
    p_analyze = sub.add_parser('analyze', help='extract ticket context and run diagnosis routing, including Clinic or Grafana analysis when links are present')
    p_analyze.add_argument('issue', nargs='?', help='optional Jira issue key or Pingkai ticket reference such as GTOC-1234 or TICKET-8412')
    p_analyze.add_argument('--online-grafana', '--grafana-link', dest='grafana_link', default='', help='analyze one online Grafana link directly, optionally together with a ticket reference')
    p_analyze.add_argument('--clinic-link', default='', help='analyze one Clinic link directly, or override the Clinic link used together with a ticket reference')
    p_analyze.add_argument('--from', dest='from_ts', default='', help='optional unix timestamp window start for direct Grafana analysis')
    p_analyze.add_argument('--to', dest='to_ts', default='', help='optional unix timestamp window end for direct Grafana analysis')
    p_analyze.add_argument('--skip-clinic-download', action='store_true', help='extract clinic bundle download links but do not download/extract bundle archives')
    p_analyze.add_argument('--skip-attachments', action='store_true', help='do not download ticket attachments even when logs/profile evidence is needed')
    p_analyze.add_argument('--download-attachments', action='store_true', help='force attachment download and lightweight analysis when attachment references exist')
    p_analyze.set_defaults(func=analyze_cmd)


def add_reporting_commands(sub):
    p_report = sub.add_parser('report', help='generate one or more incident report blocks')
    p_report.add_argument('issue_ids', nargs='*')
    p_report.add_argument('--storage-weekly', action='store_true', help='render the storage weekly report markdown body for PD, TXN, and TiKV')
    p_report.add_argument(
        '--start-time',
        default='',
        help='optional start time for --storage-weekly; defaults to 7 days before --end-time or now',
    )
    p_report.add_argument(
        '--end-time',
        default='',
        help='optional end time for --storage-weekly; defaults to 7 days after --start-time or now',
    )
    p_report.add_argument(
        '--source',
        default='jira',
        help='ticket source: jira, pingkai, or a comma-separated combination such as pingkai,jira (default: jira)',
    )
    p_report.add_argument('--components', default='tikv,tidb,pd', help='jira components filter')
    p_report.add_argument('--time_range', default='1w', help='relative time range such as 1w, 7d, 30d, this_week')
    p_report.add_argument(
        '--assignee',
        default='',
        help='jira assignee filter; use stable Jira identifiers such as email, English display name, account id, or currentUser(); for pingkai this matches l2_assignee or l3_assignee',
    )
    p_report.set_defaults(func=summary_cmd)

    p_rca = sub.add_parser('RCA', help='generate RCA payload JSON for one Jira issue')
    p_rca.add_argument('issue', help='Jira issue key, for example TCOC-4930')
    p_rca.add_argument('rca_link_dir', nargs='?', default=DEFAULT_RCA_LINK_DIR, help='optional Feishu wiki directory link')
    p_rca.set_defaults(func=rca_cmd)


def add_execution_commands(sub):
    p_close = sub.add_parser('close', help='preview or execute a Jira close flow')
    g = p_close.add_mutually_exclusive_group()
    g.add_argument('--preview', action='store_true', help='print the proposed close flow without changing Jira')
    g.add_argument('--no-preview', dest='no_preview', action='store_true', help='execute the close flow immediately')
    p_close.add_argument('issues', nargs='+', help='one or more Jira issue keys, for example TCOC-4930')
    p_close.set_defaults(func=close_cmd)

    p_labor = sub.add_parser('labor', help='show or add Jira/Pingkai labor records')
    p_labor.add_argument('issue', help='Jira issue key or Pingkai ticket reference such as ticket-8368')
    p_labor.add_argument('action_or_duration', help="either 'show' or a duration like 45m, 4h, or 1h30m")
    p_labor.add_argument('extra', nargs='?', help='reserved extra positional argument')
    p_labor.add_argument('--time', default='', help='optional start time or date, depending on Jira vs Pingkai mode')
    p_labor.add_argument('-m', '--comment', default='', help='optional labor / worklog comment')
    p_labor.add_argument('--dry-run', action='store_true', help='print the computed Pingkai labor payload without writing')

    def labor_dispatch(args):
        if args.action_or_duration == 'show':
            args.action = 'show'
            args.duration = ''
        else:
            args.action = 'add'
            args.duration = args.action_or_duration
        return labor_cmd(args)

    p_labor.set_defaults(func=labor_dispatch)


def build_parser():
    parser = argparse.ArgumentParser(
        prog='oncall-toolkit',
        description='Oncall workflow helper for Jira and Pingkai tickets.\n\n' + capability_layer_description(),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    sub = parser.add_subparsers(dest='cmd', required=True)
    add_discovery_commands(sub)
    add_diagnosis_commands(sub)
    add_reporting_commands(sub)
    add_execution_commands(sub)
    return parser


def main():
    argv = list(sys.argv[1:])
    if argv and argv[0] == 'summary':
        argv[0] = 'report'

    parser = build_parser()
    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == '__main__':
    raise SystemExit(main())