File size: 24,560 Bytes
b025706
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
# SPDX-FileCopyrightText: © 2025 Tenstorrent USA, Inc.
# SPDX-License-Identifier: Apache-2.0

"""
These tests demonstrate how to use @validate_against to compare TTNN implementations
against reference PyTorch implementations with automatic metrics collection.
"""

import pytest
import torch

import ttnn
from models.common.auto_compose import to_torch_auto_compose
from models.common.metrics import compute_pcc_host
from models.common.validation_tools import (
    Metric,
    MetricSpec,
    clear_validation_results,
    compare_to_torch,
    compare_to_ttnn,
    enable_validation,
    get_validation_registry,
)

# [INFO] the purpose of this test is to validate the validation framework itself,
# which does not care about the mesh shape or tensor layout; we have other test files on those topics.
pytestmark = [
    pytest.mark.parametrize(
        "ttnn_mesh_device",
        [
            (1, 1),
        ],
        ids=[
            "1x1",
        ],
        indirect=True,
    ),
]

# ============================================================================
# Example 1: Validating RMSNorm against PyTorch reference
# ============================================================================


def torch_rms_norm(x, weight, eps=1e-6):
    """Reference PyTorch implementation of RMS normalization"""
    variance = x.pow(2).mean(-1, keepdim=True)
    x = x * torch.rsqrt(variance + eps)
    return weight * x


class HostValidatedRMSNorm:
    """RMS Normalization with validation decorator using old input_map pattern"""

    def __init__(self, weight: torch.Tensor, eps: float, device):
        self.eps = eps
        self.weight = ttnn.from_torch(
            weight.unsqueeze(0).unsqueeze(0), device=device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT
        )

    @compare_to_torch(
        reference_fn=torch_rms_norm,
        input_to_torch=lambda self, x: (
            # [INFO] produce input args to torch_rms_norm as a tuple
            (to_torch_auto_compose(x), to_torch_auto_compose(self.weight)),
            # [INFO] produce input kwargs to torch_rms_norm as a dict
            {"eps": self.eps},
        ),
        metric_tolerances={
            Metric.MAX_ABS_ERROR: 5e-2,
            Metric.MEAN_ABS_ERROR: 1e-3,
            "pcc": 0.99,  # can use enum or their string values
        },
        enabled=True,
    )
    def __call__(self, x):
        # x shape: [1, seq_len, hidden_size]
        x_squared = ttnn.mul(x, x)
        mean_x_squared = ttnn.mean(x_squared, dim=-1, keepdim=True)
        rms = ttnn.sqrt(ttnn.add(mean_x_squared, self.eps))
        x_normed = ttnn.mul(x, ttnn.reciprocal(rms))
        return ttnn.mul(x_normed, self.weight)


class DeviceValidatedRMSNorm:
    """RMS Normalization - ultra-clean pattern: NO conversions needed!"""

    def __init__(self, weight: torch.Tensor, eps: float, device):
        self.eps = eps
        self.weight_torch = weight  # Keep for reference
        self.weight = ttnn.from_torch(
            weight.unsqueeze(0).unsqueeze(0), device=device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT
        )
        self.device = device

    def _reference_impl(self, x):
        """Reference implementation - mocking a TTNN reference implementation for testing"""
        # Convert TTNN to torch for reference computation
        x_torch = ttnn.to_torch(x).squeeze(0)
        result_torch = torch_rms_norm(x_torch, self.weight_torch, self.eps)
        # Convert back to TTNN to match __call__ output type
        return ttnn.from_torch(
            result_torch.unsqueeze(0), device=self.device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT
        )

    # [INFO] this decorator is useful when the reference function is a TTNN-native function.
    #        currently, it is experimental and requires the reference function has same-ordered
    #        inputs as the decorated function.
    @compare_to_ttnn(
        reference_fn=lambda self, x: self._reference_impl(x),
        # [INFO] passing `metric_tolerances` is optional; if not provided, the default tolerances will be used:
        # metric_tolerances={
        #     Metric.MAX_ABS_ERROR: 1e-2,
        #     Metric.PCC: 0.99,
        # },
    )
    def __call__(self, x):
        # x shape: [1, seq_len, hidden_size]
        x_squared = ttnn.mul(x, x)
        mean_x_squared = ttnn.mean(x_squared, dim=-1, keepdim=True)
        rms = ttnn.sqrt(ttnn.add(mean_x_squared, self.eps))
        x_normed = ttnn.mul(x, ttnn.reciprocal(rms))
        return ttnn.mul(x_normed, self.weight)

    @compare_to_ttnn(
        reference_fn=lambda self, x: self._reference_impl(x),
    )
    def _call_torch__(self, x):
        # copied __call__ code below and converted to torch tensor to mock a function under test that returns a torch tensor
        # x shape: [1, seq_len, hidden_size]
        x_squared = ttnn.mul(x, x)
        mean_x_squared = ttnn.mean(x_squared, dim=-1, keepdim=True)
        rms = ttnn.sqrt(ttnn.add(mean_x_squared, self.eps))
        x_normed = ttnn.mul(x, ttnn.reciprocal(rms))
        return to_torch_auto_compose(ttnn.mul(x_normed, self.weight))


def test_validation_rmsnorm_host_and_device(ttnn_mesh_device: ttnn.MeshDevice):
    registry = get_validation_registry()

    hidden_size = 64
    batch_size = 1
    seq_len = 8

    weight = torch.randn(hidden_size, dtype=torch.bfloat16)

    # Device-validated RMSNorm
    rms_device = DeviceValidatedRMSNorm(weight, eps=1e-6, device=ttnn_mesh_device)
    x = torch.randn(batch_size, seq_len, hidden_size, dtype=torch.bfloat16)
    x_tt = ttnn.from_torch(x.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    _ = rms_device(x_tt)

    _ = rms_device._call_torch__(x_tt)

    # Host-validated RMSNorm
    rms_host = HostValidatedRMSNorm(weight, eps=1e-6, device=ttnn_mesh_device)
    x2 = torch.randn(batch_size, seq_len, hidden_size, dtype=torch.bfloat16)
    x2_tt = ttnn.from_torch(x2.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    _ = rms_host(x2_tt)

    assert len(registry.results) >= 3
    # Expect all validations to pass
    assert registry.results[0].metrics[Metric.MAX_ABS_ERROR].passed
    assert registry.results[1].metrics[Metric.MAX_ABS_ERROR].passed
    assert registry.results[2].metrics[Metric.MAX_ABS_ERROR].passed
    assert registry.results[2].metrics[Metric.MEAN_ABS_ERROR].passed
    assert registry.results[0].metrics[Metric.PCC].passed
    assert registry.results[1].metrics[Metric.PCC].passed
    assert registry.results[2].metrics[Metric.PCC].passed


# ============================================================================
# Example 2: Validating matrix multiplication
# ============================================================================


@compare_to_torch(
    reference_fn=torch.matmul,
    # [INFO] when reference function accepts inputs in the same order as the decorated function,
    #        we can omit input_to_torch; the mapping will be inferred automatically.
    metric_tolerances={
        Metric.MAX_ABS_ERROR: 1.5e-1,
        Metric.PCC: 0.99,
    },
)
def ttnn_matmul(a, b):
    """TTNN matrix multiplication with validation"""
    return ttnn.matmul(a, b)


# make a test case to show how to directly use auto_compose to convert ttnn to torch
@compare_to_torch(
    reference_fn=torch.matmul,
    # [INFO] this is a simple example of input remapping.
    input_to_torch=lambda a, b: (to_torch_auto_compose(b), to_torch_auto_compose(a)),
    metric_tolerances={
        Metric.MAX_ABS_ERROR: 1.5e-1,
        Metric.PCC: 0.99,
    },
)
def ttnn_matmul_reverse(a, b):
    """TTNN matrix multiplication with validation"""
    return ttnn.matmul(b, a)


def test_validation_matmul(ttnn_mesh_device: ttnn.MeshDevice):
    registry = get_validation_registry()

    m, n, k = 16, 24, 12
    a = torch.randn(1, m, k, dtype=torch.bfloat16)
    b = torch.randn(1, k, n, dtype=torch.bfloat16)

    a_tt = ttnn.from_torch(a.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    b_tt = ttnn.from_torch(b.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    _ = ttnn_matmul(a_tt, b_tt)
    _ = ttnn_matmul_reverse(b_tt, a_tt)

    # Expect two validations recorded and both passed
    assert len(registry.results) >= 2
    assert registry.results[-1].passed
    assert registry.results[-2].passed


# ============================================================================
# Example 3: Custom metrics and complex mappings
# ============================================================================


def custom_attention_reference(q, k, v, scale):
    """Reference attention computation"""
    scores = torch.matmul(q, k.transpose(-2, -1)) * scale
    attn_weights = torch.nn.functional.softmax(scores, dim=-1)
    return torch.matmul(attn_weights, v)


@compare_to_torch(
    reference_fn=custom_attention_reference,
    # [INFO]{ when reference function accepts inputs in the same order as the decorated function,
    # we can omit input_to_torch; it will be inferred automatically as if the following code were written:
    # input_to_torch=lambda q, k, v, scale: (
    #     to_torch_auto_compose(q),
    #     to_torch_auto_compose(k),
    #     to_torch_auto_compose(v),
    #     scale,
    # ),
    # [INFO]}
    metric_tolerances={
        Metric.MAX_ABS_ERROR: 0.1,
        Metric.MEAN_ABS_ERROR: 0.02,
        Metric.PCC: 0.99,
    },
)
def ttnn_attention(q, k, v, scale):
    """Simplified attention with validation"""
    scores = ttnn.matmul(q, ttnn.transpose(k, -2, -1))
    scores = ttnn.mul(scores, scale)
    attn_weights = ttnn.softmax(scores, dim=-1)
    return ttnn.matmul(attn_weights, v)


def test_validation_attention(ttnn_mesh_device: ttnn.MeshDevice):
    m, n, dk, dv = 8, 8, 16, 16
    q = torch.randn(1, m, dk, dtype=torch.bfloat16)
    k = torch.randn(1, n, dk, dtype=torch.bfloat16)
    v = torch.randn(1, n, dv, dtype=torch.bfloat16)

    q_tt = ttnn.from_torch(q.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    k_tt = ttnn.from_torch(k.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    v_tt = ttnn.from_torch(v.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    registry = get_validation_registry()
    before = len(registry.results)
    scale = 1.0 / (dk**0.5)
    _ = ttnn_attention(q_tt, k_tt, v_tt, scale)
    assert len(registry.results) == before + 1
    test_result = registry.results[-1]
    # expect the test to pass max_abs_error and pcc checks
    assert test_result.metrics[Metric.MAX_ABS_ERROR].passed
    assert test_result.metrics[Metric.PCC].passed
    assert test_result.metrics[Metric.MEAN_ABS_ERROR].passed


# ============================================================================
# Example 4: Validating from_torch checkpoint
# ============================================================================


@compare_to_torch(
    reference_fn=lambda tensor, device: tensor,
    output_to_torch=to_torch_auto_compose,
    metric_tolerances={
        Metric.MAX_ABS_ERROR: 0.015,
        Metric.MEAN_ABS_ERROR: 0.01,
        Metric.PCC: 0.99,
    },
)
def from_torch_checkpoint(tensor: torch.Tensor, device: ttnn.MeshDevice):
    """Return TTNN tensor created via from_torch from a checkpoint tensor."""
    return ttnn.from_torch(tensor, device=device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)


def test_validation_checkpoint_from_torch(ttnn_mesh_device: ttnn.MeshDevice):
    registry = get_validation_registry()
    before = len(registry.results)

    # Simulated checkpoint tensor (e.g., a weight matrix)
    rows, cols = 32, 128
    weight = torch.randn(rows, cols, dtype=torch.float32)

    # Validate a direct from_torch call via the decorated function
    _ = from_torch_checkpoint(weight, ttnn_mesh_device)

    # Ensure a result was recorded and it passed
    assert len(registry.results) == before + 1
    assert registry.results[-1].passed


# ============================================================================
# Example 5: Validating with MetricSpec
# ============================================================================


@compare_to_torch(
    reference_fn=torch.matmul,
    metric_tolerances={
        "pcc_host": MetricSpec(tolerance=0.99, higher_is_better=True, compute_fn=compute_pcc_host),
    },
)
def ttnn_matmul_metric_spec(a, b):
    return ttnn.matmul(a, b)


def test_validation_matmul_metric_spec(ttnn_mesh_device: ttnn.MeshDevice):
    registry = get_validation_registry()
    before = len(registry.results)

    m, n, k = 8, 10, 6
    a = torch.randn(1, m, k, dtype=torch.bfloat16)
    b = torch.randn(1, k, n, dtype=torch.bfloat16)

    a_tt = ttnn.from_torch(a.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    b_tt = ttnn.from_torch(b.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    _ = ttnn_matmul_metric_spec(a_tt, b_tt)

    assert len(registry.results) == before + 1
    assert registry.results[-1].passed


# ============================================================================
# Example 6: Validating with non-decorator use of compare_to_torch
#            between a class instance and a reference class instance!
# ============================================================================


def test_validation_non_decorator_class_vs_class_torch(ttnn_mesh_device: ttnn.MeshDevice):
    """Validate a callable class against a reference class using non-decorator style."""
    registry = get_validation_registry()
    before = len(registry.results)

    # Simple linear layer implemented in TTNN (__call__) vs Torch reference (forward)
    m, n, k = 8, 10, 6
    x = torch.randn(1, m, k, dtype=torch.bfloat16)
    w = torch.randn(1, k, n, dtype=torch.bfloat16)

    class TTLinear:
        def __init__(self, weight: torch.Tensor, device: ttnn.MeshDevice):
            # Weight expected as [1, k, n]; add device batch dim for TTNN tensor
            self.weight = ttnn.from_torch(
                weight.unsqueeze(0), device=device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT
            )

        def __call__(self, inp):
            return ttnn.matmul(inp, self.weight)

    class TorchLinearRef:
        def __init__(self, weight: torch.Tensor):
            self.weight = weight

        def forward(self, inp: torch.Tensor):
            return torch.matmul(inp, self.weight)

    # Instantiate both implementations
    layer = TTLinear(w, ttnn_mesh_device)
    ref_layer = TorchLinearRef(w)

    # Convert input to TTNN tensor (add device batch dim)
    x_tt = ttnn.from_torch(x.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    # Non-decorator usage: wrap the unbound __call__ so we can pass (self, x)
    validated_call = compare_to_torch(
        reference_fn=lambda self, inp: ref_layer.forward(inp),
        metric_tolerances={
            Metric.MAX_ABS_ERROR: 1.5e-1,
            Metric.PCC: 0.99,
        },
    )(TTLinear.__call__)

    _ = validated_call(layer, x_tt)

    assert len(registry.results) == before + 1
    assert registry.results[-1].passed


# ============================================================================
# Example 7: Validating with non-decorator use of compare_to_ttnn
#            between a class instance (return torch tensor) and a reference class instance
#            (return TTNN tensor)
# NOTE: This use of compare_to_ttnn could come in handy in situations where a module instance
#       within torch implementation is being replaced by a TTNN module instance and
#       we want to check the output of the TTNN module instance against the output of
#       the torch module instance during end2end testing.
# ============================================================================


def test_validation_non_decorator_class_vs_class_ttnn(ttnn_mesh_device: ttnn.MeshDevice):
    """Validate a callable TTNN class against a TTNN reference class using non-decorator style."""
    registry = get_validation_registry()
    before = len(registry.results)

    # Simple linear layer implemented in TTNN (__call__) vs TTNN reference (forward)
    m, n, k = 8, 10, 6
    x = torch.randn(1, m, k, dtype=torch.bfloat16)
    w = torch.randn(1, k, n, dtype=torch.bfloat16)

    class TTLinear:
        def __init__(self, weight: torch.Tensor, device: ttnn.MeshDevice):
            # Weight expected as [1, k, n]; add device batch dim for TTNN tensor
            self.weight = ttnn.from_torch(
                weight.unsqueeze(0), device=device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT
            )

        def __call__(self, inp):
            return ttnn.matmul(inp, self.weight)

    class TorchLinearRef:
        def __init__(self, weight: torch.Tensor, device: ttnn.MeshDevice):
            self.weight = weight

        def forward(self, inp):
            return torch.matmul(inp, self.weight)

    # Instantiate both implementations
    layer = TTLinear(w, ttnn_mesh_device)
    ref_layer = TorchLinearRef(w, ttnn_mesh_device)

    # # Convert input to TTNN tensor (add device batch dim)
    # x_tt = ttnn.from_torch(x.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    # Non-decorator usage: wrap the unbound __call__ so we can pass (self, x)
    validated_call = compare_to_ttnn(
        reference_fn=lambda inp: layer(inp),
        input_to_ttnn=lambda self, inp: (
            ttnn.from_torch(inp, device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT),
        ),
        metric_tolerances={
            Metric.MAX_ABS_ERROR: 1.5e-1,
            Metric.PCC: 0.99,
        },
    )(TorchLinearRef.forward)

    _ = validated_call(ref_layer, x.unsqueeze(0))

    assert len(registry.results) == before + 1
    assert registry.results[-1].passed


def test_return_reference_output_torch(ttnn_mesh_device: ttnn.MeshDevice):
    """Demonstrate return_reference_output=True returns the reference (torch) output.

    The decorator computes torch.matmul on host for reference, then returns that
    reference result converted back to a TTNN tensor distributed like the impl output.
    """
    registry = get_validation_registry()
    before = len(registry.results)

    m, n, k = 8, 10, 6
    a = torch.randn(1, m, k, dtype=torch.bfloat16)
    b = torch.randn(1, k, n, dtype=torch.bfloat16)
    a_tt = ttnn.from_torch(a.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    b_tt = ttnn.from_torch(b.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    ref_host = torch.ones(m, n, dtype=torch.bfloat16)

    @compare_to_torch(
        # mock a reference function that returns a torch tensor with the same shape as the decorated function output
        reference_fn=lambda a, b: ref_host,
        metric_tolerances={
            Metric.MAX_ABS_ERROR: 1,  # outrageous tolerance to confirm the mock
            Metric.PCC: 0.99,
        },
        return_reference_output=True,
    )
    def _impl_matmul(a, b):
        return ttnn.matmul(a, b)

    # Call impl; returned value should be the reference result (distributed as impl output)
    out_tt = _impl_matmul(a_tt, b_tt)

    # Registry records one validation
    assert len(registry.results) == before + 1
    assert not registry.results[-1].metrics[Metric.MAX_ABS_ERROR].passed
    # Mock ref is constant (ones); impl is matmul
    assert not registry.results[-1].metrics[Metric.PCC].passed

    # Convert both outputs to host and verify numerical equivalence
    out_host = to_torch_auto_compose(out_tt)
    assert torch.allclose(out_host, ref_host)


# ============================================================================
# Additional test functions
# ============================================================================


def test_validation_enable_disable(ttnn_mesh_device: ttnn.MeshDevice):
    a = torch.randn(1, 8, 8, dtype=torch.bfloat16)
    b = torch.randn(1, 8, 8, dtype=torch.bfloat16)
    a_tt = ttnn.from_torch(a.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    b_tt = ttnn.from_torch(b.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    registry = get_validation_registry()
    recorded_after_enable = len(registry.results)
    # Disable validation: should not record
    enable_validation(False)
    _ = ttnn_matmul(a_tt, b_tt)
    assert len(registry.results) == recorded_after_enable

    # Re-enable for subsequent tests
    enable_validation(True)


def test_validation_non_decorator_host(ttnn_mesh_device: ttnn.MeshDevice):
    registry = get_validation_registry()
    before = len(registry.results)

    m, n, k = 8, 10, 6
    a = torch.randn(1, m, k, dtype=torch.bfloat16)
    b = torch.randn(1, k, n, dtype=torch.bfloat16)
    a_tt = ttnn.from_torch(a.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    b_tt = ttnn.from_torch(b.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    def _matmul(a, b):
        return ttnn.matmul(a, b)

    validated_matmul = compare_to_torch(
        reference_fn=torch.matmul,
        metric_tolerances={
            Metric.MAX_ABS_ERROR: 1.5e-1,
            Metric.PCC: 0.99,
        },
    )(_matmul)

    _ = validated_matmul(a_tt, b_tt)

    assert len(registry.results) == before + 1
    assert registry.results[-1].passed


def test_validation_raises_on_reference_exception(ttnn_mesh_device: ttnn.MeshDevice, expect_error):
    """When raise_exceptions=True, reference exceptions should propagate and not record results."""
    registry = get_validation_registry()
    before = len(registry.results)

    a = torch.randn(1, 8, 8, dtype=torch.bfloat16)
    b = torch.randn(1, 8, 8, dtype=torch.bfloat16)
    a_tt = ttnn.from_torch(a.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)
    b_tt = ttnn.from_torch(b.unsqueeze(0), device=ttnn_mesh_device, dtype=ttnn.bfloat16, layout=ttnn.TILE_LAYOUT)

    def _ref_raises(a, b):
        pass

    # [INFO] make a mismatched signature on reference function to force the reference function to raise an exception!
    @compare_to_torch(reference_fn=lambda a, b, c: _ref_raises(a, b), raise_exceptions=True)
    def _matmul(a, b):
        return ttnn.matmul(a, b)

    with expect_error(TypeError, "missing 1 required positional argument: 'c'") as e:
        _ = _matmul(a_tt, b_tt)
    assert "missing 1 required positional argument: 'c'" in str(e.value)
    assert len(registry.results) == before + 1

    # [INFO] make a mismatched signature on output_to_torch to force the reference function to raise an exception!
    @compare_to_torch(reference_fn=lambda a, b: ..., output_to_torch=lambda x, y: ..., raise_exceptions=True)
    def _matmul_too(a, b):
        return ttnn.matmul(a, b)

    with expect_error(TypeError, "missing 1 required positional argument: 'y'") as e:
        _ = _matmul_too(a_tt, b_tt)
    assert "missing 1 required positional argument: 'y'" in str(e.value)
    assert len(registry.results) == before + 2

    # [INFO] make a mismatched signature on input_to_torch to force the reference function to raise an exception!
    @compare_to_torch(reference_fn=lambda a, b: ..., input_to_torch=lambda x: ..., raise_exceptions=True)
    def _matmul_three(a, b):
        return ttnn.matmul(a, b)

    with expect_error(TypeError, "takes 1 positional argument but 2 were given") as e:
        _ = _matmul_three(a_tt, b_tt)
    assert "takes 1 positional argument but 2 were given" in str(e.value)
    assert len(registry.results) == before + 3


@pytest.fixture(scope="module", autouse=True)
def _print_validation_report_after_module(request):
    # Runs once after all tests in this module finish
    yield
    registry = get_validation_registry()
    reporter = request.config.pluginmanager.get_plugin("terminalreporter")
    reporter.write_line("Printing validation report after yield")
    registry.print_report(verbose=True)


@pytest.fixture(scope="module", autouse=True)
def _clear_validation_results_before_module():
    clear_validation_results()


# todo)) add an example where the reference is a file name that we can load output tensors from!
# idea: validating against tensors -- give us a file name and we can compare our tensor under test against that!
# - a filename for the tensor