diff --git a/src/TA_main2main_workflow/external_test/__init__.py b/src/TA_main2main_workflow/external_test/__init__.py
new file mode 100644
index 0000000..cf39aea
--- /dev/null
+++ b/src/TA_main2main_workflow/external_test/__init__.py
@@ -0,0 +1,25 @@
+"""External operator repository pluggable test cases.
+
+Provides config loading and a runner for executing pytest suites
+in external (non-vendored) operator repositories, with AI-driven
+fix retry loops on failure.
+
+Core entry points:
+    - :func:`load_external_test_config` — parse a YAML config file
+    - :func:`run_external_tests` — clone, install, test, fix (per repo)
+    - :class:`ExternalTestConfig` / :class:`ExternalTestRepoConfig` — dataclasses
+"""
+
+from TA_main2main_workflow.external_test.config_loader import (
+    ExternalTestConfig,
+    ExternalTestRepoConfig,
+    load_external_test_config,
+)
+from TA_main2main_workflow.external_test.runner import run_external_tests
+
+__all__ = [
+    "ExternalTestConfig",
+    "ExternalTestRepoConfig",
+    "load_external_test_config",
+    "run_external_tests",
+]
diff --git a/src/TA_main2main_workflow/external_test/config_loader.py b/src/TA_main2main_workflow/external_test/config_loader.py
new file mode 100644
index 0000000..05f0fb1
--- /dev/null
+++ b/src/TA_main2main_workflow/external_test/config_loader.py
@@ -0,0 +1,204 @@
+"""External test configuration loader and validation.
+
+Parses ``external_test_config.yaml`` into typed dataclasses and validates
+URL / path correctness before the runner consumes them.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import yaml  # type: ignore[import-untyped]
+
+from TA_main2main_workflow.utils.logging import get_logger
+
+log = get_logger(__name__)
+
+# ── Default config shipped with the package ───────────────────────────────
+_DEFAULT_CONFIG_DIR = Path(__file__).resolve().parent
+_DEFAULT_CONFIG_PATH = _DEFAULT_CONFIG_DIR / "external_test_config.yaml"
+
+
+@dataclass
+class ExternalTestRepoConfig:
+    """Configuration for a single external operator repository."""
+
+    name: str  # display name
+    url: str  # git clone URL
+    branch: str = "main"  # branch / tag to checkout
+    test_cases: list[str] = field(default_factory=list)  # test file relative paths
+    install_cmd: str = ""  # optional dep install (empty = skip)
+
+
+@dataclass
+class ExternalTestConfig:
+    """Top-level external test configuration."""
+
+    enabled: bool = False  # master on/off switch
+    repos: list[ExternalTestRepoConfig] = field(default_factory=list)
+    test_procs: int = 8  # pytest -n <N>
+    mode: str = "inline"  # inline | standalone | off
+    max_retries: int = 5  # AI fix retries per repo
+    timeout: int = 7200  # per-repo test timeout (seconds)
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Public API
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def load_external_test_config(path: str = "") -> ExternalTestConfig | None:
+    """Load external test config from *path* (YAML).
+
+    Resolution order:
+        1. Explicit *path* argument
+        2. ``TA_EXTERNAL_TEST_CONFIG`` environment variable
+        3. Default ``external_test/external_test_config.yaml`` shipped with package
+
+    Returns ``None`` when the config file does not exist (not an error — the
+    caller treats missing config as "no external tests configured").
+    """
+    resolved = _resolve_config_path(path)
+    if resolved is None:
+        return None
+
+    log.info(f"Loading external test config: {resolved}")
+    try:
+        raw = yaml.safe_load(resolved.read_text(encoding="utf-8")) or {}
+    except yaml.YAMLError as exc:
+        log.error(f"Failed to parse external test config: {exc}")
+        return None
+
+    cfg = _dict_to_config(raw)
+
+    # Merge env-var overrides (env takes precedence over YAML)
+    _apply_env_overrides(cfg)
+
+    if not validate_config(cfg):
+        return None
+
+    log.key_value("External test enabled", str(cfg.enabled))
+    log.key_value("External test mode", cfg.mode)
+    log.key_value("External test repos", str(len(cfg.repos)))
+    return cfg
+
+
+def validate_config(cfg: ExternalTestConfig) -> bool:
+    """Validate the loaded config. Returns True if usable."""
+    if cfg.mode not in ("inline", "standalone", "off"):
+        log.error(f"Invalid external test mode: {cfg.mode}")
+        return False
+
+    if cfg.test_procs < 1:
+        log.error("test_procs must be >= 1")
+        return False
+
+    if cfg.max_retries < 0:
+        log.error("max_retries must be >= 0")
+        return False
+
+    for repo in cfg.repos:
+        if not repo.name:
+            log.error("External test repo missing 'name'")
+            return False
+        if not repo.url or not (
+            repo.url.startswith("http") or repo.url.startswith("git@")
+        ):
+            log.error(f"Invalid repo URL for '{repo.name}': {repo.url}")
+            return False
+        if not repo.test_cases:
+            log.warning(f"External repo '{repo.name}' has no test_cases configured")
+
+    return True
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Internal helpers
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def _resolve_config_path(explicit: str) -> Path | None:
+    """Determine which config file to read."""
+    if explicit:
+        p = Path(explicit)
+        if p.exists():
+            return p
+        log.warning(f"External test config not found: {explicit}")
+        return None
+
+    env_path = os.getenv("TA_EXTERNAL_TEST_CONFIG", "")
+    if env_path:
+        p = Path(env_path)
+        if p.exists():
+            return p
+        log.warning(f"TA_EXTERNAL_TEST_CONFIG points to missing file: {env_path}")
+
+    if _DEFAULT_CONFIG_PATH.exists():
+        return _DEFAULT_CONFIG_PATH
+
+    return None
+
+
+def _dict_to_config(raw: dict[str, Any]) -> ExternalTestConfig:
+    """Convert raw YAML dict to ExternalTestConfig."""
+    repos: list[ExternalTestRepoConfig] = []
+    for item in raw.get("external_test_repos", []) or []:
+        repos.append(
+            ExternalTestRepoConfig(
+                name=item.get("name", ""),
+                url=item.get("url", ""),
+                branch=item.get("branch", "main"),
+                test_cases=item.get("test_cases", []),
+                install_cmd=item.get("install_cmd", ""),
+            )
+        )
+
+    return ExternalTestConfig(
+        enabled=bool(raw.get("enabled", False)),
+        repos=repos,
+        test_procs=int(raw.get("test_procs", 8)),
+        mode=str(raw.get("mode", "inline")),
+        max_retries=int(raw.get("max_retries", 5)),
+        timeout=int(raw.get("timeout", 7200)),
+    )
+
+
+def _apply_env_overrides(cfg: ExternalTestConfig) -> None:
+    """Apply environment variable overrides on top of YAML config.
+
+    Environment variables always take precedence over the YAML file.
+    """
+    # Master switch
+    env_enabled = os.getenv("TA_EXTERNAL_TEST_ENABLED", "").lower()
+    if env_enabled in ("true", "1", "yes"):
+        cfg.enabled = True
+    elif env_enabled in ("false", "0", "no"):
+        cfg.enabled = False
+
+    # Mode
+    env_mode = os.getenv("TA_EXTERNAL_TEST_MODE", "").lower()
+    if env_mode in ("inline", "standalone", "off"):
+        cfg.mode = env_mode
+
+    # Parallelism
+    try:
+        cfg.test_procs = int(os.getenv("TA_EXTERNAL_TEST_PROCS", str(cfg.test_procs)))
+    except ValueError:
+        pass
+
+    # Max retries
+    try:
+        cfg.max_retries = int(
+            os.getenv("TA_EXTERNAL_TEST_MAX_RETRIES", str(cfg.max_retries))
+        )
+    except ValueError:
+        pass
+
+    # Timeout
+    try:
+        cfg.timeout = int(os.getenv("TA_EXTERNAL_TEST_TIMEOUT", str(cfg.timeout)))
+    except ValueError:
+        pass
diff --git a/src/TA_main2main_workflow/external_test/external_test_config.yaml b/src/TA_main2main_workflow/external_test/external_test_config.yaml
new file mode 100644
index 0000000..d80a5b5
--- /dev/null
+++ b/src/TA_main2main_workflow/external_test/external_test_config.yaml
@@ -0,0 +1,74 @@
+# ── 全局开关：是否启用外部测试（默认 false，需用户主动开启）──
+enabled: false                  # 设为 true 启用外部测试
+
+# ── 运行模式：inline（阻塞主流程）| standalone（不阻塞）| off（禁用）──
+mode: inline                    # 仅在 enabled=true 时生效
+
+# ── 全局默认值 ────────────────────────────────────────────────────
+test_procs: 8                   # 每个外部算子仓 pytest 并行进程数
+max_retries: 5                  # 每仓最大 AI 修复重试次数
+timeout: 7200                   # 单仓测试超时（秒）
+
+# ── 外部算子仓列表（按配置顺序串行执行）───────────────────────────
+external_test_repos:
+  - name: "Liger-Kernel"
+    url: "https://github.com/linkedin/Liger-Kernel.git"
+    branch: "main"
+    # install_cmd: "pip install -e ."    # 可选，为空则跳过安装
+    test_cases:
+      - "test/transformers/test_attn_res.py"
+      - "test/transformers/test_auto_model.py"
+      - "test/transformers/test_cute_moe_autograd.py"
+      - "test/transformers/test_cutedsl_rms_norm.py"
+      - "test/transformers/test_cutedsl_rms_norm_fastpath.py"
+      - "test/transformers/test_cutedsl_rope.py"
+      - "test/transformers/test_cutile_backend.py"
+      - "test/transformers/test_dyt.py"
+      - "test/transformers/test_embedding.py"
+      - "test/transformers/test_flex_attention.py"
+      - "test/transformers/test_fused_add_rms_norm.py"
+      - "test/transformers/test_fused_linear_cross_entropy.py"
+      - "test/transformers/test_fused_linear_scaled_cross_entropy.py"
+      - "test/transformers/test_group_norm.py"
+      - "test/transformers/test_kl_div.py"
+      - "test/transformers/test_layer_norm.py"
+      - "test/transformers/test_llama4_rope.py"
+      - "test/transformers/test_mhc.py"
+      - "test/transformers/test_mm_int8int2.py"
+      - "test/transformers/test_modulated_rms_norm.py"
+      - "test/transformers/test_moe.py"
+      - "test/transformers/test_monkey_patch.py"
+      - "test/transformers/test_swiglu_cutedsl.py"
+
+  - name: "flash-linear-attention"
+    url: "https://github.com/fla-org/flash-linear-attention.git"
+    branch: "main"
+    test_cases:
+      - "tests/context_parallel/test_cp_conv.py"
+      - "tests/context_parallel/test_cp_dplr.py"
+      - "tests/context_parallel/test_cp_gdn.py"
+      - "tests/context_parallel/test_cp_kda.py"
+      - "tests/context_parallel/test_cp_rwkv7.py"
+      - "tests/context_parallel/test_cp_token_shift.py"
+      - "tests/layers/test_layer_cache_layer_idx.py"
+      - "tests/models/test_cache.py"
+      - "tests/models/test_generation_utils.py"
+      - "tests/models/test_hybrid_attention.py"
+      - "tests/models/test_modeling_bitnet.py"
+      - "tests/models/test_modeling_deltaformer.py"
+      - "tests/models/test_modeling_mla.py"
+      - "tests/models/test_modeling_moba.py"
+      - "tests/models/test_modeling_mom.py"
+      - "tests/models/test_modeling_nsa.py"
+      - "tests/models/test_modeling_rodimus.py"
+      - "tests/models/test_modeling_samba.py"
+      - "tests/models/test_modeling_transformer.py"
+      - "tests/modules/test_grpo.py"
+      - "tests/modules/test_l2norm.py"
+      - "tests/ops/test_cache.py"
+      - "tests/ops/test_forgetting_attn.py"
+      - "tests/ops/test_moba.py"
+      - "tests/ops/test_titans.py"
+      - "tests/test_public_api.py"
+      - "tests/test_split_package_release.py"
+      - "tests/utils/test_ascend_ub_manager.py"
diff --git a/src/TA_main2main_workflow/external_test/runner.py b/src/TA_main2main_workflow/external_test/runner.py
new file mode 100644
index 0000000..8939dac
--- /dev/null
+++ b/src/TA_main2main_workflow/external_test/runner.py
@@ -0,0 +1,549 @@
+"""External test runner — clone, install, test, fix loop per operator repo.
+
+Executes configured external operator repository test suites sequentially,
+one repo at a time.  Each repo follows::
+
+    clone → install_deps → pytest → [fail? → AI fix → retest] → next repo
+
+Results are written to ``workspace/test-logs/`` as JUnit XML + per-repo JSON
+summaries, matching the conventions used by the main pytest_ut pipeline.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shlex
+import shutil
+import signal
+import subprocess
+import time
+import xml.etree.ElementTree as ET
+from pathlib import Path
+
+from TA_main2main_workflow.external_test.config_loader import (
+    ExternalTestConfig,
+    ExternalTestRepoConfig,
+)
+from TA_main2main_workflow.agent.opencode_adapter import run_opencode_adapter
+from TA_main2main_workflow.utils.config import TAConfig
+from TA_main2main_workflow.utils.context import WorkflowContext
+from TA_main2main_workflow.utils.git import run_git
+from TA_main2main_workflow.utils.logging import get_logger
+from TA_main2main_workflow.utils.tracker import timed
+from TA_main2main_workflow.utils import WORKSPACE_DIR
+
+log = get_logger(__name__)
+
+# Top-level directory under WORKSPACE_DIR where external repos are cloned
+_EXTERNAL_REPOS_DIR = "external_repos"
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Public API
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def run_external_tests(
+    ctx: WorkflowContext, config: TAConfig, external_cfg: ExternalTestConfig
+) -> WorkflowContext:
+    """Execute external operator repo tests sequentially.
+
+    1. Ensure ``workspace/external_repos/`` exists
+    2. For each repo in *external_cfg.repos* (in config order):
+        a. ``clone_or_update_repo``
+        b. ``install_dependencies``
+        c. ``run_repo_tests`` — pytest with configured test cases
+        d. On failure: ``_external_test_fix_loop`` — AI fix + retest
+        e. Record result, continue to next repo
+    3. Return updated WorkflowContext
+
+    When *external_cfg.mode* is ``"standalone"``, failures are recorded but
+    never propagated as workflow-fatal (caller decides).
+    """
+    repos_dir = WORKSPACE_DIR / _EXTERNAL_REPOS_DIR
+    repos_dir.mkdir(parents=True, exist_ok=True)
+
+    test_log_dir = WORKSPACE_DIR / "test-logs"
+    test_log_dir.mkdir(parents=True, exist_ok=True)
+
+    all_passed = True
+    results: list[dict] = []
+
+    if not external_cfg.repos:
+        log.warning("No external test repos configured — nothing to run")
+        log.info(
+            "Set TA_EXTERNAL_TEST_CONFIG to point to a config file with repos, "
+            "or edit the default config at external_test/external_test_config.yaml"
+        )
+        return ctx.copy_with(
+            external_test_passed=True,
+            external_test_results=[],
+        )
+
+    for i, repo_cfg in enumerate(external_cfg.repos):
+        log.section(f"External Repo {i + 1}/{len(external_cfg.repos)}: {repo_cfg.name}")
+        repo_result = _process_one_repo(
+            repo_cfg, repos_dir, test_log_dir, ctx, config, external_cfg
+        )
+        results.append(repo_result)
+        if not repo_result.get("passed", False):
+            all_passed = False
+
+    # ── Write aggregate summary ───────────────────────────────────────────
+    summary_file = test_log_dir / "external-test-summary.json"
+    summary = {
+        "all_passed": all_passed,
+        "total_repos": len(external_cfg.repos),
+        "passed_repos": sum(1 for r in results if r.get("passed")),
+        "failed_repos": sum(1 for r in results if not r.get("passed")),
+        "results": results,
+    }
+    summary_file.write_text(
+        json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
+    log.info(f"External test summary: {summary_file}")
+
+    return ctx.copy_with(
+        external_test_passed=all_passed,
+        external_test_results=results,
+    )
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Per-repo processing
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def _process_one_repo(
+    repo_cfg: ExternalTestRepoConfig,
+    repos_dir: Path,
+    test_log_dir: Path,
+    ctx: WorkflowContext,
+    config: TAConfig,
+    external_cfg: ExternalTestConfig,
+) -> dict:
+    """Run the full pipeline for a single external repo. Returns a result dict."""
+    log.key_value("repo", repo_cfg.name)
+    log.key_value("url", repo_cfg.url)
+    log.key_value("branch", repo_cfg.branch)
+
+    repo_path = repos_dir / repo_cfg.name
+    result: dict = {
+        "repo": repo_cfg.name,
+        "passed": False,
+        "failed_cases": [],
+        "fix_count": 0,
+    }
+
+    # ── Step 1: Clone or update ───────────────────────────────────────
+    with timed(f"clone-{repo_cfg.name}"):
+        if not clone_or_update_repo(repo_cfg, repo_path):
+            result["error"] = "clone/update failed"
+            log.error(f"Failed to clone/update repo: {repo_cfg.name}")
+            return result
+
+    # ── Step 2: Install dependencies (skip if install_cmd is empty) ──
+    if repo_cfg.install_cmd:
+        with timed(f"install-{repo_cfg.name}"):
+            if not install_dependencies(repo_path, repo_cfg):
+                result["error"] = "dependency installation failed"
+                log.error(f"Failed to install dependencies for: {repo_cfg.name}")
+                return result
+    else:
+        log.info(f"No install_cmd configured for {repo_cfg.name} — skipping")
+
+    # ── Step 3: Run tests + fix loop ──────────────────────────────────
+    passed, fix_count, failed_cases = _run_repo_tests_with_fix(
+        repo_path, repo_cfg, test_log_dir, ctx, config, external_cfg
+    )
+
+    result["passed"] = passed
+    result["fix_count"] = fix_count
+    result["failed_cases"] = failed_cases
+
+    if passed:
+        log.status(True, f"External repo {repo_cfg.name}: PASSED")
+    else:
+        log.status(False, f"External repo {repo_cfg.name}: FAILED")
+
+    return result
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Git operations
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def clone_or_update_repo(cfg: ExternalTestRepoConfig, repo_path: Path) -> bool:
+    """Clone *cfg.url* into *repo_path*, or ``git pull`` if it already exists."""
+    if repo_path.exists():
+        log.info(f"Repo exists, pulling latest: {repo_path}")
+        try:
+            run_git(repo_path, "fetch", "origin")
+            run_git(repo_path, "checkout", cfg.branch)
+            run_git(repo_path, "pull", "origin", cfg.branch)
+            log.info(f"Updated repo: {cfg.name}")
+            return True
+        except Exception as exc:
+            log.warning(f"git pull failed for {cfg.name}: {exc}")
+            return False
+    else:
+        log.info(f"Cloning {cfg.url} → {repo_path}")
+        try:
+            repo_path.parent.mkdir(parents=True, exist_ok=True)
+            result = subprocess.run(
+                ["git", "clone", "--branch", cfg.branch, cfg.url, str(repo_path)],
+                capture_output=True,
+                text=True,
+                timeout=600,
+            )
+            if result.returncode != 0:
+                log.error(f"Clone failed: {result.stderr.strip()}")
+                return False
+            log.info(f"Cloned repo: {cfg.name}")
+            return True
+        except subprocess.TimeoutExpired:
+            log.error(f"Clone timed out: {cfg.url}")
+            return False
+        except Exception as exc:
+            log.error(f"Clone failed for {cfg.name}: {exc}")
+            return False
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Dependency installation
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def install_dependencies(repo_path: Path, cfg: ExternalTestRepoConfig) -> bool:
+    """Run *cfg.install_cmd* inside *repo_path* via ``bash -c``."""
+    cmd = cfg.install_cmd
+    log.info(f"Installing dependencies: {cmd}")
+
+    try:
+        result = subprocess.run(
+            ["bash", "-c", cmd],
+            cwd=str(repo_path),
+            capture_output=True,
+            text=True,
+            timeout=1800,
+        )
+        if result.returncode != 0:
+            log.error(
+                f"Dependency install failed for {cfg.name}:\n{result.stderr[-2000:]}"
+            )
+            return False
+        log.info(f"Dependencies installed: {cfg.name}")
+        return True
+    except subprocess.TimeoutExpired:
+        log.error(f"Dependency install timed out: {cfg.name}")
+        return False
+    except Exception as exc:
+        log.error(f"Dependency install error for {cfg.name}: {exc}")
+        return False
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Test execution + fix loop
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def _run_repo_tests_with_fix(
+    repo_path: Path,
+    repo_cfg: ExternalTestRepoConfig,
+    test_log_dir: Path,
+    ctx: WorkflowContext,
+    config: TAConfig,
+    external_cfg: ExternalTestConfig,
+) -> tuple[bool, int, list[str]]:
+    """Run tests for a single repo with optional AI fix loop on failure.
+
+    Returns ``(passed, fix_count, failed_cases)``.
+    """
+    procs = external_cfg.test_procs
+    max_retries = external_cfg.max_retries
+    timeout = external_cfg.timeout
+
+    for attempt in range(max_retries + 1):
+        if attempt > 0:
+            if config.skip_ai_analysis:
+                log.info("SKIP_AI_ANALYSIS=true — cannot fix, aborting retries")
+                return False, 0, _list_test_cases(repo_cfg)
+
+            log.header(
+                f"External Fix Attempt {attempt}/{max_retries} — {repo_cfg.name}"
+            )
+            _external_test_ai_fix(
+                repo_path, repo_cfg, ctx, config, attempt, test_log_dir
+            )
+
+        # Run pytest
+        passed, failed_cases = run_repo_tests(
+            repo_path, repo_cfg, test_log_dir, procs, timeout
+        )
+
+        if passed:
+            return True, attempt, []
+        elif attempt == max_retries:
+            log.error(
+                f"External repo {repo_cfg.name} failed after {max_retries} fix attempts"
+            )
+            return False, attempt, failed_cases
+
+        log.info(
+            f"External test failed (attempt {attempt + 1}) — will retry with AI fix"
+        )
+
+    return False, max_retries, _list_test_cases(repo_cfg)
+
+
+def run_repo_tests(
+    repo_path: Path,
+    repo_cfg: ExternalTestRepoConfig,
+    test_log_dir: Path,
+    procs: int = 8,
+    timeout: int = 7200,
+) -> tuple[bool, list[str]]:
+    """Run pytest for the configured test files in *repo_cfg*.
+
+    Returns ``(passed, failed_case_paths)``.
+    """
+    if not repo_cfg.test_cases:
+        log.info(f"No test cases configured for {repo_cfg.name} — treating as passed")
+        return True, []
+
+    test_paths: list[Path] = []
+    for tc in repo_cfg.test_cases:
+        p = repo_path / tc
+        if p.exists():
+            test_paths.append(p)
+        else:
+            log.warning(f"Test file not found in {repo_cfg.name}: {tc}")
+
+    if not test_paths:
+        log.warning(f"No test files found for {repo_cfg.name} — treating as passed")
+        return True, []
+
+    junit_xml = test_log_dir / f"pytest-junit-external-{repo_cfg.name}.xml"
+    output_log = test_log_dir / f"test-output-external-{repo_cfg.name}.log"
+    pytest_bin = shutil.which("pytest")
+    python_exe = os.getenv("PYTHON", "python3.10")
+    cmd = [pytest_bin] if pytest_bin else [python_exe, "-m", "pytest"]
+    cmd += [str(p.relative_to(repo_path)) for p in test_paths]
+    cmd += ["-n", str(procs), f"--junitxml={junit_xml}"]
+
+    log.key_value(
+        f"[{repo_cfg.name}] test files",
+        ", ".join(tc for tc in repo_cfg.test_cases),
+    )
+    log.info(f"[{repo_cfg.name}] cmd: {' '.join(cmd)}")
+    log.info(f"[{repo_cfg.name}] output: {output_log}")
+
+    _start = time.time()
+    # Write header first, then tee pytest output: console (live) + log file
+    with open(output_log, "w", encoding="utf-8") as fh:
+        fh.write(f"=== External Test: {repo_cfg.name} ===\n")
+        fh.write(f"cmd: {' '.join(cmd)}\n\n")
+
+    full_cmd = " ".join(shlex.quote(c) for c in cmd)
+    full_cmd += f" 2>&1 | tee -a {shlex.quote(str(output_log))}"
+    proc = subprocess.Popen(
+        ["bash", "-c", full_cmd],
+        cwd=str(repo_path),
+        start_new_session=True,
+    )
+    try:
+        rc = proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        # Kill the whole process group (bash + tee + pytest + xdist workers)
+        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+        rc = -1
+        log.warning(f"[{repo_cfg.name}] pytest timed out after {timeout}s")
+
+    elapsed = time.time() - _start
+    log.info(f"[{repo_cfg.name}] pytest finished in {elapsed:.0f}s, returncode={rc}")
+
+    # Parse JUnit XML
+    pf = pe = tp = 0
+    failed_cases: list[str] = []
+    if junit_xml.exists():
+        try:
+            tree = ET.parse(junit_xml)
+            root_elem = tree.getroot()
+            suites = (
+                [root_elem]
+                if root_elem.tag != "testsuites"
+                else root_elem.findall("testsuite")
+            )
+            for s in suites:
+                tp += int(s.get("tests", 0))
+                pf += int(s.get("failures", 0))
+                pe += int(s.get("errors", 0))
+                for tc_elem in s.findall("testcase"):
+                    failure = tc_elem.find("failure")
+                    error = tc_elem.find("error")
+                    if failure is not None or error is not None:
+                        failed_cases.append(
+                            f"{tc_elem.get('classname', '')}.{tc_elem.get('name', '')}"
+                        )
+        except Exception:
+            log.warning(f"Could not parse JUnit XML: {junit_xml}")
+
+    # PASS requires: pytest exited cleanly (rc=0), JUnit shows no
+    # failures/errors, AND at least one test was actually collected.
+    # rc=4 (usage/collection error) and rc=5 (nothing collected) with an
+    # empty JUnit would otherwise be misread as "all passed".
+    passed = rc == 0 and pf == 0 and pe == 0 and tp > 0
+
+    # Write per-repo result file
+    result_file = test_log_dir / f"test-result-external-{repo_cfg.name}.json"
+    result_summary = {
+        "repo": repo_cfg.name,
+        "label": f"external-{repo_cfg.name}",
+        "exit_code": 0 if passed else 1,
+        "passed": passed,
+        "test_log": str(junit_xml),
+        "output_log": str(output_log),
+        "test_cases": repo_cfg.test_cases,
+        "passed_count": tp,
+        "failed_count": pf,
+        "error_count": pe,
+        "failed_case_details": failed_cases,
+    }
+    result_file.write_text(
+        json.dumps(result_summary, indent=2, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+    )
+
+    if not passed:
+        if rc != 0:
+            log.error(
+                f"[{repo_cfg.name}] pytest exited with rc={rc} "
+                f"({tp} collected, {pf} failed, {pe} errors) "
+                f"— see {output_log}"
+            )
+        else:
+            log.error(f"[{repo_cfg.name}] Tests FAILED ({pf} failed, {pe} errors)")
+    else:
+        log.status(True, f"[{repo_cfg.name}] All tests passed ({tp} passed)")
+
+    return passed, failed_cases
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# AI fix for external tests
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+def _external_test_ai_fix(
+    repo_path: Path,
+    repo_cfg: ExternalTestRepoConfig,
+    ctx: WorkflowContext,
+    config: TAConfig,
+    attempt: int,
+    test_log_dir: Path,
+) -> None:
+    """Invoke AI to fix test failures in an external operator repository.
+
+    Unlike the main pipeline's ``ai_fix``, this function does NOT gate on
+    ``_ALLOWED_FIX_PREFIXES`` — the entire external repo is fair game for
+    modifications.  Validation only checks that files exist within the repo.
+    """
+    if config.skip_ai_analysis:
+        return
+
+    # Gather error logs for AI context
+    error_log_paths: list[str] = []
+    for pattern in [
+        f"test-output-external-{repo_cfg.name}.log",
+        f"pytest-junit-external-{repo_cfg.name}.xml",
+        f"test-result-external-{repo_cfg.name}.json",
+    ]:
+        p = test_log_dir / pattern
+        if p.exists():
+            error_log_paths.append(str(p))
+
+    fix_dir = WORKSPACE_DIR / "fixes" / f"external-{repo_cfg.name}-fix-{attempt}"
+    fix_dir.mkdir(parents=True, exist_ok=True)
+
+    log.info(f"Invoking AI fix for external repo: {repo_cfg.name}")
+    try:
+        _ = _list_tracked_files(repo_path)
+
+        result = run_opencode_adapter(
+            {
+                "step_id": f"external-{repo_cfg.name}-fix-{attempt}",
+                "previous_step_id": "",
+                "previous_step_summary_path": "",
+                "is_last_step": "true",
+                "step_dir": str(test_log_dir),
+                "fix_dir": str(fix_dir),
+                "conflict_dir": "",
+                "ascend_path": str(repo_path),
+                "triton_path": str(repo_path),
+                "reference_dir": "",
+                "mode": "external_test_fix",
+                "error_logs": json.dumps(error_log_paths, ensure_ascii=False),
+                "target_commit": "",
+                "step_index": f"external/{repo_cfg.name}",
+                "ascend_npu_ir_fix": "false",
+                "ascend_npu_ir_compat_ref": "",
+            }
+        )
+
+        # ── Validate: changes must be inside the external repo ─────────
+        if result.modified_files:
+            repo_path_str = str(repo_path)
+            illegal = [
+                f for f in result.modified_files if not _is_under_path(f, repo_path_str)
+            ]
+            if illegal:
+                log.warning(f"AI fix touched files outside {repo_cfg.name}: {illegal}")
+                rejection_file = fix_dir / "fix_rejection.txt"
+                rejection_file.write_text(
+                    f"VALIDATION REJECTED: files outside repo {repo_cfg.name}\n"
+                    f"Illegal files: {illegal}\n",
+                    encoding="utf-8",
+                )
+                _revert_changes(repo_path)
+
+        log.ai_result(
+            bool(result.modified_files),
+            result.modified_files,
+            (result.step_summary or "")[:500],
+        )
+    except Exception as exc:
+        log.error(f"AI fix failed for external repo {repo_cfg.name}: {exc}")
+
+
+def _is_under_path(file_path: str, parent: str) -> bool:
+    """Check whether *file_path* resides under *parent* directory."""
+    try:
+        Path(file_path).resolve().relative_to(Path(parent).resolve())
+        return True
+    except ValueError:
+        return False
+
+
+def _list_tracked_files(repo: Path) -> set[str]:
+    """Return the set of all tracked files in *repo*."""
+    try:
+        output = run_git(repo, "ls-files")
+        return set(output.strip().splitlines())
+    except Exception:
+        return set()
+
+
+def _revert_changes(repo: Path) -> None:
+    """Revert all uncommitted changes and remove untracked files in *repo*."""
+    try:
+        run_git(repo, "checkout", "--", ".")
+        run_git(repo, "clean", "-fd")
+    except Exception as exc:
+        log.error(f"Failed to revert changes in {repo}: {exc}")
+
+
+def _list_test_cases(repo_cfg: ExternalTestRepoConfig) -> list[str]:
+    """Return a copy of the configured test case paths."""
+    return list(repo_cfg.test_cases)
diff --git a/src/TA_main2main_workflow/flow.py b/src/TA_main2main_workflow/flow.py
index 58e2b18..17945c9 100644
--- a/src/TA_main2main_workflow/flow.py
+++ b/src/TA_main2main_workflow/flow.py
@@ -7,7 +7,7 @@
         if LLVM hash changed: per_step_ir_patch (apply existing → build LLVM →
           build TA → test → supplement IR → loop)
         else: build_and_fix_loop → test_and_fix_loop
-      → commit
+      → [external_test] → commit
     → finalize → [push_pr]
 """
 
@@ -32,6 +32,10 @@
 from TA_main2main_workflow.pipeline.resolve import resolve_conflicts
 from TA_main2main_workflow.pipeline.build import build_and_fix_loop
 from TA_main2main_workflow.pipeline.test import test_and_fix_loop
+from TA_main2main_workflow.external_test import (
+    load_external_test_config,
+    run_external_tests,
+)
 from TA_main2main_workflow.pipeline.commit import commit_step
 from TA_main2main_workflow.pipeline.finalize import finalize
 from TA_main2main_workflow.pipeline.ir_patch import (
@@ -46,7 +50,8 @@ class TA_Main2MainFlow:
     """Orchestrator — builds context, runs pipeline steps, handles PR.
 
     Single-step mode is the only supported mode.  Each step runs the full
-    pipeline: merge → resolve conflicts → build → fix → test → fix → commit.
+    pipeline: merge → resolve conflicts → build → fix → test → fix →
+    external_test → commit.
     """
 
     def __init__(self, config: TAConfig | None = None) -> None:
@@ -163,7 +168,12 @@ def run(self) -> str:
                     ctx = ctx.copy_with(final_status=UpgradeFailed)
                     return UpgradeFailed
 
-            # ── Step D: Commit ──────────────────────────────────
+            # ── Step D: External test ──────────────────────────
+            ctx = self._run_external_test_stage(ctx)
+            if ctx.final_status == UpgradeFailed:
+                return UpgradeFailed
+
+            # ── Step E: Commit ──────────────────────────────────
             with timed("commit"):
                 ctx = commit_step(ctx, self.config)
 
@@ -221,6 +231,42 @@ def run(self) -> str:
         ctx = ctx.copy_with(final_status=UpgradeCompleted)
         return UpgradeCompleted
 
+    def _run_external_test_stage(self, ctx: WorkflowContext) -> WorkflowContext:
+        """Run external operator repo tests (if enabled).
+
+        Loads the YAML config, checks the enabled flag, and executes the
+        external test pipeline.  In ``inline`` mode failures are fatal;
+        in ``standalone`` mode failures are recorded but don't block.
+        """
+        external_cfg = load_external_test_config(self.config.external_test_config)
+
+        if external_cfg is None:
+            # No config file found — not an error, just nothing to do
+            return ctx
+
+        if not external_cfg.enabled:
+            log.info("External test is disabled (enabled=false) — skipped")
+            return ctx
+
+        if external_cfg.mode == "off":
+            log.info("External test mode is 'off' — skipped")
+            return ctx
+
+        log.section(f"External Test — {external_cfg.mode} mode")
+        with timed("external-test"):
+            ctx = run_external_tests(ctx, self.config, external_cfg)
+
+        if external_cfg.mode == "inline" and not ctx.external_test_passed:
+            log.error("External tests failed (inline mode)")
+            ctx = ctx.copy_with(final_status=UpgradeFailed)
+
+        if ctx.external_test_passed:
+            log.status(True, "External tests passed")
+        else:
+            log.status(False, "External tests failed")
+
+        return ctx
+
     def _push_pr(self, ctx: WorkflowContext) -> None:
         """Push work branch and create GitHub PR."""
         from TA_main2main_workflow.pipeline.push_pr import push_and_create_pr
diff --git a/src/TA_main2main_workflow/reference/03-unit-test-failure-diagnosis-and-fixes.md b/src/TA_main2main_workflow/reference/03-unit-test-failure-diagnosis-and-fixes.md
index 980f7d0..81f0b95 100644
--- a/src/TA_main2main_workflow/reference/03-unit-test-failure-diagnosis-and-fixes.md
+++ b/src/TA_main2main_workflow/reference/03-unit-test-failure-diagnosis-and-fixes.md
@@ -344,7 +344,7 @@ tmp = tl.load(in_ptr + ((-NEG_INDEX) + offset), mask=(offset >= NEG_INDEX), othe
 
 **正确修复（pointer rebase）:**
 
-文件：`third_party/ascend/lib/TritonToLinalg/BlockPtrAnalysis.cpp`  
+文件：`third_party/ascend/lib/TritonToLinalg/BlockPtrAnalysis.cpp`
 函数：`BlockDataParser::rewriteAddPtr`（IntToPtr→`pointer_cast` 之后、`createCastOp` 之前）
 
 ```text
@@ -364,7 +364,7 @@ if linear 存在 && linear < 0:
 
 然后：`hivm.hir.pointer_cast(%addr) : memref<?xT>`（动态 `?` 基址；定长 tile 由后续 `reinterpret_cast sizes` 给出）。
 
-等价：`base + (i + S)`（`S<0`）≡ `(base advanced by S elems) + i`。  
+等价：`base + (i + S)`（`S<0`）≡ `(base advanced by S elems) + i`。
 有加有减但 **总和 ≥ 0**：**不要改**。
 
 **禁止的错误修复:**
diff --git a/src/TA_main2main_workflow/utils/config.py b/src/TA_main2main_workflow/utils/config.py
index f95475e..3984773 100644
--- a/src/TA_main2main_workflow/utils/config.py
+++ b/src/TA_main2main_workflow/utils/config.py
@@ -81,6 +81,11 @@ class TAConfig:
     # ── IR patch ──────────────────────────────────────────────────────────
     ir_max_iterations: int = 3
 
+    # ── External test ─────────────────────────────────────────────────────
+    external_test_enabled: bool = False  # TA_EXTERNAL_TEST_ENABLED env
+    external_test_config: str = ""  # TA_EXTERNAL_TEST_CONFIG env
+    external_test_mode: str = "inline"  # TA_EXTERNAL_TEST_MODE env
+
     # ═══════════════════════════════════════════════════════════════════════
     @classmethod
     def from_env(cls) -> TAConfig:
@@ -132,6 +137,11 @@ def from_env(cls) -> TAConfig:
             python_exe=os.getenv("PYTHON", ""),
             single_step_mode=_env_bool("TA_SINGLE_STEP_MODE", True),
             ir_max_iterations=_env_int("TA_IR_MAX_ITERATIONS", 3),
+            external_test_enabled=_env_bool("TA_EXTERNAL_TEST_ENABLED", False),
+            external_test_config=os.getenv("TA_EXTERNAL_TEST_CONFIG", ""),
+            external_test_mode=_env_choice(
+                "TA_EXTERNAL_TEST_MODE", ["inline", "standalone", "off"], "inline"
+            ),
         )
 
     @property
diff --git a/src/TA_main2main_workflow/utils/context.py b/src/TA_main2main_workflow/utils/context.py
index 41e1d06..c72266f 100644
--- a/src/TA_main2main_workflow/utils/context.py
+++ b/src/TA_main2main_workflow/utils/context.py
@@ -84,6 +84,11 @@ class WorkflowContext:
     step_pr_descriptions: list[str] = field(default_factory=list)
     summary_rows: list[tuple] = field(default_factory=list)
 
+    # ── External test results ─────────────────────────────────────────────
+    external_test_passed: bool = False
+    external_test_results: list[dict] = field(default_factory=list)
+    # Per-repo entry: {"repo": "Liger-Kernel", "passed": True, "failed_cases": [], "fix_count": 0}
+
     # ── Final state ───────────────────────────────────────────────────────
     final_status: str = ""
     pr_url: str = ""