fix(expert-skill): align FlyDSL selectors and validation

.github/workflows/ci-l0-checks.yml

@@ -60,7 +60,7 @@ jobs:
       # path (typed scalars, folded multi-line EXTRA_<BE>_ARGS). Without it those
       # tests would silently exercise the degraded line-scanner fallback instead.
       - run: pip install pytest pytest-cov coverage pyyaml
-      # Stdlib-only tests (import run_e2e via importlib; no claude_agent_sdk / GPU).
+      # CPU-only tests (no claude_agent_sdk / GPU); the expert-skill test also uses PyYAML.
       # Files are listed explicitly rather than discovered, so that adding a test
       # under examples/ (torch/triton + GPU) can never silently break this job.
       #
@@ -98,8 +98,19 @@ jobs:
             e2e_workflow/scripts/tests/test_server_teardown.py \
             e2e_workflow/scripts/tests/test_bench_e2e_teardown_lookup.py \
             ci/node/test_run_model_material_audit.py \
+            perf_knowledge/expert_skills/_contribute/test_flydsl_skill_portability.py \
             geak/test_bootstrap.py
 
+      - name: expert skill static validation
+        run: |
+          for skill in \
+            flydsl_decode_moe_stage1_blkmap \
+            flydsl_prefill_moe_stage2_fp8partial \
+            flydsl_fp8_blockscale_gemm
+          do
+            python perf_knowledge/expert_skills/_contribute/validate_skill.py "$skill" --static
+          done
+
       # Standalone contract checks. NOT in the pytest list above: each runs its assertions at import
       # time and exits non-zero, which pytest collection cannot express. They were added with the
       # learned-KB work and were not wired into CI at all — a regression suite nothing runs is a

e2e_workflow/roles/_fragments/expert_skills.md

@@ -1,8 +1,9 @@
 # Fragment: expert_skills (e2e layer) — ADVISORY, injected only when use_expert_skills is ON
 
 > This fragment is appended to a role's prompt by `e2e_workflow.js` **only when `use_expert_skills`
-> is true (opt-in; default OFF)**. When OFF (the default), nothing is injected and behavior is
-> byte-identical to a run without this feature. It is consumed by routing/integration roles (System
+> is true (opt-in; default OFF)**. When OFF, this fragment is not injected; the base role prompt remains active,
+> including general routing and target-backend guidance independent of this registry. It is
+> consumed by routing/integration roles (System
 > Architect, Op Benchmarker, e2e
 > Integrator). It is **advisory**: a matched skill is a high-prior candidate to reproduce, never a
 > mandate, and never overrides your on-box A/B gate.
@@ -18,7 +19,8 @@ the measurement wins (note it so the skill is later marked `stale`).
 
 1. **Read the selector.** Open `EXPERT_SKILLS_DIR/index.yaml`.
 2. **Match against the live bottleneck** you are routing/optimizing. A skill matches when ALL hold:
-   - `match.operator` == the bottleneck operator (same names as `capability_index.yaml`)
+   - `match.operator` is either a scalar equal to the bottleneck operator or a list containing that
+     operator (same names as `capability_index.yaml`)
    - the box `gen` ∈ `match.gens`
    - `env_report.model_arch_class` ∈ `match.arch_class` (or `match.arch_class` contains `'*'`)
    - if the skill is a migration skill (`from_backend`/`to_backend` set), the live path / your author

kernel_workflow/roles/_fragments/expert_skills.md

@@ -1,7 +1,8 @@
 # Fragment: expert_skills (kernel layer) — ADVISORY, injected only when use_expert_skills is ON
 
 > Appended to a role's prompt by `kernel_workflow.js` **only when `use_expert_skills` is true
-> (opt-in; default OFF)**. When OFF (the default), nothing is injected and behavior is byte-identical.
+> (opt-in; default OFF)**. When OFF, this fragment is not injected; the base role prompt remains active,
+> including general target-backend and language guidance from `perf_knowledge`.
 > Consumed by the
 > tech_lead (planning) and author/engineer roles. **Advisory**: a matched skill is a high-prior
 > candidate to reproduce, never a mandate, and never overrides your isolated A/B vs the oracle.
@@ -16,7 +17,8 @@ faster but can never reduce a result below your measured baseline.
 
 1. Read `EXPERT_SKILLS_DIR/index.yaml`.
 2. A skill matches the current op when ALL hold:
-   - `match.operator` == this op's operator (`KK_OPERATOR` / `op_spec.op_kind`)
+   - `match.operator` is either a scalar equal to this op's operator
+     (`KK_OPERATOR` / `op_spec.op_kind`) or a list containing that operator
    - box `gen` ∈ `match.gens`; `op_spec.dtype` ∈ `match.dtypes`; `op_spec.regime` ∈ `match.regimes`
    - migration skills: `from_backend`→`to_backend` fits this run's `mode`/`target_language`
      (e.g. authoring Triton from a TileLang source → a `tilelang→triton` skill applies)

kernel_workflow/roles/tech_lead.md

@@ -120,7 +120,7 @@ analysis below exactly as before.)
      programming model. (Dir names differ from the ids for the others: triton→`triton_amd`,
      hip→`hip_cpp`, ck→`composable_kernel`, asm→`asm_mfma`.)
    - **Cross-backend port / migration** (the TASK asks to rewrite the kernel into a DIFFERENT backend —
-     ANY `source→target`, e.g. `ck→flydsl`, `triton→tilelang`, `hip→ck`, `ck→ck_tile`): keep `kk_language`
+     ANY `source→target`, e.g. `ck→flydsl`, `triton→tilelang`, `hip→ck`): keep `kk_language`
      = the CURRENT editable source, but `kk_refs` MUST ALSO include (a) the TARGET backend card
      `operators/<kk_operator>/backends/<target>.md` and (b) the TARGET language's authoring how-to under
      `languages/<dir>/` — map the language id to its dir: triton→`triton_amd`, hip→`hip_cpp`,

perf_knowledge/expert_skills/README.md

@@ -38,13 +38,14 @@ kernel, tuned config JSONs, a custom validation manifest — can carry those fil
 
 ## How a skill is selected by the workflows
 
-Each skill's frontmatter has a `match:` block. Expert skills are **opt-in** — the workflows ignore this
-directory entirely unless the run passes `use_expert_skills=true` (default OFF; when OFF the workflow
-behaves byte-identically to a build without this feature). When enabled, the workflow filters
-`index.yaml` by the current bottleneck:
+Each skill's frontmatter has a `match:` block. Expert-skill selection is **opt-in** — the workflows do
+not inject or query this registry unless the run passes `use_expert_skills=true` (default OFF). When
+OFF, the base role prompt remains active, including general `perf_knowledge` guidance that is independent
+of expert-skill selection. When enabled, the workflow filters `index.yaml` by the current bottleneck:
 
 ```
-match.operator == bottleneck.operator
+(match.operator is scalar AND match.operator == bottleneck.operator)
+  OR (match.operator is a list AND bottleneck.operator ∈ match.operator)
 AND gen ∈ match.gens
 AND model_arch_class ∈ match.arch_class   (or match.arch_class contains '*')
 AND (migration skills) from_backend/to_backend fit the live path

perf_knowledge/expert_skills/_contribute/scaffold.py

@@ -42,6 +42,23 @@ def known_operators():
     return {c["operator"] for c in (data.get("candidates") or []) if "operator" in c}
 
 
+def skill_index_entry(sub, fm):
+    """Build one selector entry, including non-profile dependency skills."""
+    scope = fm.get("scope", "kernel")
+    entry = {
+        "id": fm["id"],
+        "file": f"skills/{sub}/skill.md",
+        "scope": scope,
+        "match": fm.get("match", {}),
+    }
+    if scope == "dependency":
+        entry["validation_status"] = "n/a"
+    else:
+        entry["expects"] = fm.get("expects", {})
+        entry["validation_status"] = (fm.get("validation") or {}).get("status", "draft")
+    return entry
+
+
 def reindex():
     ops = known_operators()
     entries = []
@@ -57,14 +74,7 @@ def reindex():
             if ops is not None and one != "*" and one not in ops:
                 print(f"  WARN: {sub}/skill.md: operator '{one}' not in capability_index.yaml",
                       file=sys.stderr)
-        entries.append({
-            "id": fm["id"],
-            "file": f"skills/{sub}/skill.md",
-            "scope": fm.get("scope", "kernel"),
-            "match": fm.get("match", {}),
-            "expects": fm.get("expects", {}),
-            "validation_status": (fm.get("validation") or {}).get("status", "draft"),
-        })
+        entries.append(skill_index_entry(sub, fm))
     header = (
         "# index.yaml — expert_skills selector (AUTO-MAINTAINED by _contribute/scaffold.py + "
         "validate_skill.py).\n"

perf_knowledge/expert_skills/_contribute/test_flydsl_skill_portability.py

@@ -1,14 +1,29 @@
 from pathlib import Path
+import importlib.util
+import subprocess
+import sys
 
 import pytest
 import yaml
 
 
 SKILLS_ROOT = Path(__file__).resolve().parents[1] / "skills"
+EXPERT_SKILLS_ROOT = SKILLS_ROOT.parent
+GEAK_ROOT = EXPERT_SKILLS_ROOT.parents[1]
 MOE_SKILLS = [
     "flydsl_decode_moe_stage1_blkmap",
     "flydsl_prefill_moe_stage2_fp8partial",
 ]
+DENSE_SKILL = "flydsl_fp8_blockscale_gemm"
+SUBMITTED_SKILLS = [*MOE_SKILLS, DENSE_SKILL]
+
+
+def load_scaffold_module():
+    path = EXPERT_SKILLS_ROOT / "_contribute" / "scaffold.py"
+    spec = importlib.util.spec_from_file_location("expert_skill_scaffold", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
 
 
 def load_skill(skill_id):
@@ -35,3 +50,142 @@ def test_flydsl_skill_uses_portable_minimum_version_guidance(skill_id):
     evidence = yaml.safe_load(artifact.read_text())
     assert evidence["skill_id"] == skill_id
     assert evidence["flydsl_version"] == "0.2.2"
+
+
+@pytest.mark.parametrize("skill_id", MOE_SKILLS)
+def test_moe_skills_match_both_taxonomy_ids_on_existing_flydsl(skill_id):
+    frontmatter, _ = load_skill(skill_id)
+    match = frontmatter["match"]
+
+    assert match["operator"] == ["grouped_gemm_moe", "fused_moe_grouped_gemm"]
+    assert match["from_backend"] == "flydsl"
+    assert match["to_backend"] == "flydsl"
+
+
+def test_dense_skill_matches_dense_and_scaled_quant_taxonomy():
+    frontmatter, _ = load_skill(DENSE_SKILL)
+
+    assert frontmatter["match"]["operator"] == ["dense_gemm", "scaled_quant_gemm"]
+
+
+@pytest.mark.parametrize(
+    "fragment",
+    [
+        GEAK_ROOT / "kernel_workflow" / "roles" / "_fragments" / "expert_skills.md",
+        GEAK_ROOT / "e2e_workflow" / "roles" / "_fragments" / "expert_skills.md",
+    ],
+)
+def test_selector_prompt_defines_list_operator_membership(fragment):
+    text = fragment.read_text()
+
+    assert "a list containing" in text
+
+
+@pytest.mark.parametrize("skill_id", MOE_SKILLS)
+def test_emit_plan_uses_in_place_optimize_for_moe_skills(skill_id):
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(EXPERT_SKILLS_ROOT / "_contribute" / "validate_skill.py"),
+            skill_id,
+            "--emit-plan",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+    assert "mode=optimize" in result.stdout
+    assert "target_language=triton" not in result.stdout
+    assert "mode=author" not in result.stdout
+
+
+def test_emit_plan_lists_author_and_optimize_for_mixed_source_skill():
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(EXPERT_SKILLS_ROOT / "_contribute" / "validate_skill.py"),
+            "gluon_authoring",
+            "--emit-plan",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+    assert "mode=author target_language=gluon" in result.stdout
+    assert "mode=optimize" in result.stdout
+
+
+@pytest.mark.parametrize("skill_id", SUBMITTED_SKILLS)
+def test_kernel_skills_do_not_claim_an_unmeasured_e2e_gate(skill_id):
+    frontmatter, _ = load_skill(skill_id)
+
+    assert "e2e_delta_min_pct" not in frontmatter["expects"]
+    assert frontmatter["validation"]["measured"]["e2e_pct"] == ""
+
+
+@pytest.mark.parametrize(
+    "document",
+    [
+        EXPERT_SKILLS_ROOT / "README.md",
+        GEAK_ROOT / "kernel_workflow" / "roles" / "_fragments" / "expert_skills.md",
+        GEAK_ROOT / "e2e_workflow" / "roles" / "_fragments" / "expert_skills.md",
+    ],
+)
+def test_disabled_wording_is_limited_to_the_skill_fragment(document):
+    text = document.read_text()
+    normalized = " ".join(text.split())
+
+    assert "base role prompt remains active" in normalized
+
+
+def test_dependency_skill_reindexes_as_non_profile_matched():
+    scaffold = load_scaffold_module()
+    entry = scaffold.skill_index_entry(
+        "ensure_flydsl",
+        {
+            "id": "ensure_flydsl",
+            "scope": "dependency",
+            "match": {"needs": "flydsl"},
+        },
+    )
+
+    assert entry["scope"] == "dependency"
+    assert "expects" not in entry
+    assert entry["validation_status"] == "n/a"
+
+
+def test_cross_backend_examples_use_canonical_backend_ids():
+    text = (GEAK_ROOT / "kernel_workflow" / "roles" / "tech_lead.md").read_text()
+
+    assert "ck→ck_tile" not in text
+
+
+def test_dense_skill_frontmatter_matches_archived_down_proj_evidence():
+    frontmatter, body = load_skill(DENSE_SKILL)
+    artifact = EXPERT_SKILLS_ROOT / frontmatter["validation"]["artifact"]
+    evidence = yaml.safe_load(artifact.read_text())
+
+    assert "down_proj" in frontmatter["validation"]["measured"]["isolated"]
+    assert "tile_n=256/tile_k=128" in frontmatter["validation"]["measured"]["isolated"]
+    assert evidence["skill_id"] == DENSE_SKILL
+    assert evidence["provider"] == "standalone_flydsl"
+    assert evidence["method"] == (
+        "CUDA-event same-session paired A/B, 3 archived on-box GEAK measurements"
+    )
+    assert evidence["warmup_per_run"] == 10
+    assert evidence["iterations_per_run"] == 100
+    assert evidence["inner_repetitions"] == 3
+    assert len(evidence["baseline_latency_ms"]) == 3
+    assert len(evidence["skill_latency_ms"]) == 3
+    assert evidence["speedup"] == pytest.approx(
+        evidence["baseline_median_ms"] / evidence["skill_median_ms"], rel=5e-5
+    )
+    assert evidence["parity"]["pass"] is True
+    assert evidence["conclusions"]["decisive_lever"] == "tile_n=256/tile_k=128"
+    assert evidence["conclusions"]["xcd_swizzle"] == "non-load-bearing for down_proj"
+    assert evidence["conclusions"]["eight_wave"] == "measured dead-end for down_proj"
+    assert evidence["raw_logs"].startswith("external archived GEAK artifacts")
+    assert "artifacts" not in evidence
+    assert "8-wave-blockscale port is also an open" not in body

perf_knowledge/expert_skills/_contribute/validate_skill.py

@@ -93,24 +93,25 @@ def emit_plan(skill_id, fm, args):
         print(f"  model_path=<CONTROL_MODEL> use_expert_skills=true  # expect |e2e delta| < noise band")
     else:
         match = fm.get("match") or {}
-        dst = match.get("to_backend") or "triton"
+        dst = match.get("to_backend") or ""
         srcs = match.get("from_backend") or []
         srcs = [srcs] if isinstance(srcs, str) else list(srcs)
+        cross_backend_srcs = [src for src in srcs if src and dst and src != dst]
         print("# EFFICACY (kernel_workflow, isolated A/B vs the immutable oracle):")
         print(f"Workflow scriptPath={GEAK}/kernel_workflow/kernel_workflow.js args:")
         print(f"  kernel_path=<OP_TASK_DIR> workflow_dir={GEAK}/kernel_workflow use_expert_skills=true")
         # kernel_workflow reads target_language ONLY on the mode=author branch, so a port has to ask for
         # that branch or the run silently measures the untouched source language instead of the migration.
-        if [s for s in srcs if s != dst]:
+        if cross_backend_srcs:
+            print("  # Cross-backend author arm:")
             print(f"  mode=author target_language={dst}"
-                  f"   # port from {'|'.join(s for s in srcs if s != dst)};"
+                  f"   # port from {'|'.join(cross_backend_srcs)};"
                   " target_language is inert without mode=author")
-        else:
-            print(f"  target_language={dst}")
+        if not cross_backend_srcs or dst in srcs:
+            if cross_backend_srcs:
+                print("  # Existing target-backend arm (measure separately):")
+            print("  mode=optimize   # in-place optimization; no target_language needed")
         print(f"  task='reproduce expert_skill:{skill_id}; beat oracle, hold parity'")
-        if dst in srcs:
-            print(f"# The selector also matches an existing {dst} source ({dst}->{dst}); measure that entry")
-            print("# state separately in the default mode=optimize (no mode/target_language args).")
         print("# DO-NO-HARM (control op that does NOT match the selector must stay within noise band):")
         print("  kernel_path=<CONTROL_OP_TASK_DIR> use_expert_skills=true  # expect no regression")
     print("\nThen stamp the result with:  validate_skill.py", skill_id,