feat(v4): GDN Path D — Triton frontend seam via tilelang.poc.triton_frontend

Wires the Path D entry per the integration plan:

    from tilelang.poc.triton_frontend import from_triton_kernel
    prim = from_triton_kernel(fla_chunk_kernel, grid=..., constexprs=...,
                              target='metal')
    kernel = tilelang.compile(prim, ...)

Two-stage status probe:
  1. triton importable + tilelang.poc.triton_frontend importable
  2. fla.ops.gated_delta_rule.chunk importable

When both hold, the status reports the *actionable* blocker — the
triton_frontend is currently Tier-1 (elementwise only, see
poc/triton_frontend/__init__.py:898), while FLA chunk_gated_delta_rule
uses tl.dot + multi-stage pipelines. Status reason names the exact next
step: "extend op_mapping.OP_TABLE with matmul & exp emitters".

Dispatch falls back to Path A on any failure (missing deps, unsupported
op, runtime error) so the rest of the v4 stack keeps working
unaffected. The lowering seam (`_try_lower_fla_chunk_kernel`) is kept
as the integration point for when the frontend grows tl.dot support —
no changes to either the host TileLang or FLA codebases.

Tests (8 new): module import, status text precision, runtime↔dispatch
status round-trip, both probe helpers, env-forced fallback, fallback
parity vs Path A bit-exact, lowering seam returns (None, msg) until
op_mapping extends.

v4 suite: 137 passed / 11 skipped.

Author: apstenku123

cppmega_v4/_tilelang/linear_attention_path_d.py

@@ -0,0 +1,129 @@
+"""GDN Path D — Triton kernel → TileLang via tilelang.poc.triton_frontend.
+
+Strategy (per integration plan):
+
+    from tilelang.poc.triton_frontend import from_triton_kernel
+    prim = from_triton_kernel(
+        fla_chunk_gated_delta_rule_inner_triton_kernel,
+        grid=...,
+        constexprs={...},
+        target='metal',
+    )
+    kernel = tilelang.compile(prim, target='metal', execution_backend='tvm_ffi', out_idx=...)
+
+The triton_frontend is currently Tier-1 (elementwise only — see
+``poc/triton_frontend/__init__.py:898``). FLA's chunk_gated_delta_rule uses
+``tl.dot``, ``tl.exp``, masked load/store, multi-stage pipelines — well
+outside Tier-1. The expected outcome on first attempt is
+``NotImplementedError`` from ``op_mapping``: that's the actionable signal
+telling frontend devs which op to add next (matrix-multiply lowering, etc.).
+
+Until the frontend covers those ops, Path D is wired but unavailable, and
+the dispatch falls back to Path A. The status reason names the precise
+blocker so the next person knows what to fix.
+
+This module never modifies the host TileLang frontend or FLA — both are
+read-only imports.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import Any
+
+import mlx.core as mx
+
+
+def _triton_frontend_importable() -> tuple[bool, str]:
+    """Probe whether triton + tilelang.poc.triton_frontend are both reachable."""
+    try:
+        import triton  # noqa: F401
+    except Exception as exc:
+        return False, f"triton not importable: {exc.__class__.__name__}: {exc}"
+    try:
+        from tilelang.poc.triton_frontend import from_triton_kernel  # noqa: F401
+    except Exception as exc:
+        return False, f"tilelang.poc.triton_frontend not importable: {exc}"
+    return True, "triton + tilelang.poc.triton_frontend importable"
+
+
+def _fla_chunk_kernel_importable() -> tuple[bool, str]:
+    """Probe whether FLA's chunk_gated_delta_rule entrypoint is reachable."""
+    try:
+        from fla.ops.gated_delta_rule.chunk import chunk_gated_delta_rule  # noqa: F401
+    except Exception as exc:
+        return False, f"fla.ops.gated_delta_rule.chunk not importable: {exc}"
+    return True, "fla.ops.gated_delta_rule.chunk importable"
+
+
+def _path_d_runtime_status() -> tuple[bool, str]:
+    """Two-stage gate: frontend + source kernel both reachable, then
+    op_mapping must cover the kernel's ops (the latter is only known after
+    a real compile attempt — we surface that on first call).
+    """
+    ok_fe, reason_fe = _triton_frontend_importable()
+    if not ok_fe:
+        return False, reason_fe
+    ok_src, reason_src = _fla_chunk_kernel_importable()
+    if not ok_src:
+        return False, reason_src
+    # Triton frontend is currently Tier-1 (elementwise only). FLA chunk
+    # gated-delta uses tl.dot / tl.exp / masked-pipeline — outside Tier-1.
+    # Mark unavailable with the actionable blocker until op_mapping extends.
+    return False, (
+        "triton frontend + FLA importable, but frontend is Tier-1 (elementwise) "
+        "only; FLA chunk_gated_delta_rule uses tl.dot and multi-stage pipelines "
+        "— extend tilelang.poc.triton_frontend.op_mapping.OP_TABLE with matmul "
+        "& exp emitters to enable Path D"
+    )
+
+
+@lru_cache(maxsize=8)
+def _try_lower_fla_chunk_kernel(target: str = "metal") -> tuple[Any | None, str]:
+    """Attempt the real frontend lowering. Returns (prim_func, message).
+
+    Only invoked when ``_path_d_runtime_status()[0]`` is True. Currently a
+    placeholder that always returns ``(None, "...")`` because Tier-1 doesn't
+    cover the kernel's ops; left here as the integration seam for when the
+    frontend grows tl.dot support.
+    """
+    try:
+        from tilelang.poc.triton_frontend import from_triton_kernel
+        from fla.ops.common.chunk_o import chunk_fwd_o  # noqa: F401
+
+        # The actual entry would inspect chunk_gated_delta_rule_fwd's inner
+        # @triton.jit kernels and pass each to from_triton_kernel. FLA wraps
+        # them under chunk_gated_delta_rule_fwd_h / chunk_fwd_o; each has its
+        # own grid + constexprs. Left as the integration seam:
+        #
+        #   from fla.ops.common.chunk_delta_h import (
+        #       chunk_gated_delta_rule_fwd_h_kernel as _fla_kernel,
+        #   )
+        #   prim = from_triton_kernel(_fla_kernel, grid=..., constexprs=...,
+        #                             target=target)
+        #
+        # That call currently raises NotImplementedError on tl.dot — exactly
+        # the signal we want the next contributor to act on.
+        return None, "Path D lowering seam present but op_mapping coverage missing"
+    except Exception as exc:
+        return None, f"unexpected lowering error: {exc.__class__.__name__}: {exc}"
+
+
+def _gdn_fwd_path_d_call(*args, **kwargs):
+    """Path D entry — currently always raises so the dispatch falls back.
+
+    Kept callable so the dispatch table is symmetric across paths.
+    """
+    raise RuntimeError(
+        "GDN Path D not yet runnable — triton_frontend op_mapping needs "
+        "matmul/exp emitters before FLA chunk_gated_delta_rule can lower"
+    )
+
+
+__all__ = [
+    "_fla_chunk_kernel_importable",
+    "_gdn_fwd_path_d_call",
+    "_path_d_runtime_status",
+    "_triton_frontend_importable",
+    "_try_lower_fla_chunk_kernel",
+]

cppmega_v4/_tilelang/linear_attention_paths.py

@@ -133,25 +133,34 @@ def _path_c_call(*args, **kwargs):
 
 def _path_d_status() -> PathStatus:
     try:
-        importlib.import_module("triton")
-    except Exception:
+        from cppmega_v4._tilelang.linear_attention_path_d import (
+            _path_d_runtime_status,
+        )
+    except Exception as exc:
         return PathStatus(
-            path="path_d",
-            available=False,
-            reason="triton not importable (CPU/Apple Silicon)",
+            path="path_d", available=False,
+            reason=f"path_d module not importable: {exc}",
         )
+    ok, reason = _path_d_runtime_status()
     return PathStatus(
-        path="path_d",
-        available=False,
+        path="path_d", available=ok,
         reason=(
-            "Triton frontend wiring pending: tilelang/poc/triton_frontend/"
-            "from_triton_kernel on FLA chunk_gated_delta_rule"
+            "GDN Path D: Triton kernel → tilelang.poc.triton_frontend."
+            f"from_triton_kernel → tilelang.compile. {reason}"
         ),
     )
 
 
 def _path_d_call(*args, **kwargs):
-    return _path_a_call(*args, **kwargs)  # fallback
+    if not _path_d_status().available:
+        return _path_a_call(*args, **kwargs)
+    try:
+        from cppmega_v4._tilelang.linear_attention_path_d import (
+            _gdn_fwd_path_d_call,
+        )
+        return _gdn_fwd_path_d_call(*args, **kwargs)
+    except Exception:
+        return _path_a_call(*args, **kwargs)
 
 
 # --- Path E (vendored mlx-lm PR #1217) ------------------------------------

tests/v4/test_linear_attention_path_d.py

@@ -0,0 +1,101 @@
+"""GDN Path D (Triton frontend → TileLang) tests.
+
+Path D wraps ``tilelang.poc.triton_frontend.from_triton_kernel`` over FLA's
+``chunk_gated_delta_rule`` kernels. The frontend is currently Tier-1
+(elementwise only) — FLA's kernel uses ``tl.dot`` and multi-stage
+pipelines, so the lowering raises ``NotImplementedError`` until
+``op_mapping.OP_TABLE`` gains matmul/exp emitters. Path D therefore
+reports unavailable on every box right now, but the wiring + fallback
+must be correct, the blocker must be precisely named in the status, and
+the dispatch must fall back to Path A without raising.
+"""
+
+from __future__ import annotations
+
+import mlx.core as mx
+import numpy as np
+import pytest
+
+from cppmega_v4._tilelang.linear_attention_path_d import (
+    _fla_chunk_kernel_importable,
+    _path_d_runtime_status,
+    _triton_frontend_importable,
+    _try_lower_fla_chunk_kernel,
+)
+from cppmega_v4._tilelang.linear_attention_paths import (
+    ENV_VAR as GDN_ENV,
+    _path_d_status,
+    gated_delta_recurrent_dispatch,
+)
+from cppmega_v4.nn._external.fla_naive_gated_delta_rule import (
+    naive_recurrent_gated_delta_rule,
+)
+
+
+def test_path_d_module_imports():
+    from cppmega_v4._tilelang import linear_attention_path_d  # noqa: F401
+
+
+def test_path_d_status_names_concrete_blocker():
+    st = _path_d_status()
+    # Status must mention either the missing dep or the op_mapping coverage
+    # blocker, so the next contributor knows what to fix.
+    assert "triton" in st.reason.lower() or "fla" in st.reason.lower() \
+        or "op_mapping" in st.reason.lower()
+
+
+def test_path_d_runtime_status_matches_dispatch_status():
+    ok, reason = _path_d_runtime_status()
+    st = _path_d_status()
+    assert st.available == ok
+    assert reason in st.reason
+
+
+def test_triton_frontend_probe_returns_tuple():
+    ok, reason = _triton_frontend_importable()
+    assert isinstance(ok, bool)
+    assert isinstance(reason, str) and reason
+
+
+def test_fla_chunk_kernel_probe_returns_tuple():
+    ok, reason = _fla_chunk_kernel_importable()
+    assert isinstance(ok, bool)
+    assert isinstance(reason, str) and reason
+
+
+def test_path_d_forced_via_env_falls_back_cleanly(monkeypatch):
+    """Forcing path_d must always return valid output (via Path A fallback)."""
+    monkeypatch.setenv(GDN_ENV, "path_d")
+    B, T, H, K, V = 1, 4, 2, 32, 32
+    q = mx.random.normal((B, T, H, K))
+    k = mx.random.normal((B, T, H, K))
+    v = mx.random.normal((B, T, H, V))
+    beta = mx.sigmoid(mx.random.normal((B, T, H)))
+    g = -mx.abs(mx.random.normal((B, T, H)) * 0.1)
+    o, _ = gated_delta_recurrent_dispatch(q, k, v, beta, g)
+    assert o.shape == (B, T, H, V)
+    assert not bool(mx.any(mx.isnan(o)).item())
+
+
+def test_path_d_fallback_matches_path_a_when_unavailable(monkeypatch):
+    ok, _ = _path_d_runtime_status()
+    if ok:
+        pytest.skip("Path D actually available — fallback path not exercised")
+    monkeypatch.setenv(GDN_ENV, "path_d")
+    B, T, H, K, V = 1, 5, 2, 8, 8
+    rng = np.random.default_rng(13)
+    q = mx.array(rng.standard_normal((B, T, H, K)).astype(np.float32))
+    k = mx.array(rng.standard_normal((B, T, H, K)).astype(np.float32))
+    v = mx.array(rng.standard_normal((B, T, H, V)).astype(np.float32))
+    beta = mx.array(rng.uniform(0.1, 0.9, (B, T, H)).astype(np.float32))
+    g = mx.array(-rng.uniform(0.01, 0.5, (B, T, H)).astype(np.float32))
+    o_disp, _ = gated_delta_recurrent_dispatch(q, k, v, beta, g)
+    o_ref, _ = naive_recurrent_gated_delta_rule(q, k, v, beta, g)
+    np.testing.assert_array_equal(np.array(o_disp), np.array(o_ref))
+
+
+def test_try_lower_returns_none_until_op_mapping_extended():
+    """Until op_mapping covers tl.dot, the lowering seam returns (None, msg)."""
+    result, msg = _try_lower_fla_chunk_kernel(target="metal")
+    assert result is None
+    assert isinstance(msg, str) and msg