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
|
#!/usr/bin/env python3
"""Deterministic completeness and failure-retention tests for X1 joiner."""
import copy
from analyze_crossover_81 import aggregate
from crossover_registry import METHODS
PLAIN_ARCHITECTURES = ("minicnn", "vgglike", "vgg16")
RESNET_DEPTHS = (20, 32, 56)
TRANSFORMER_DEPTHS = (4, 8, 12)
def plain_report():
records = []
for architecture in PLAIN_ARCHITECTURES:
for method in METHODS:
records.append({
"architecture": architecture,
"method": method,
"status": "completed",
"outcome": "finite",
"best_validation_accuracy": 75.0,
"final_validation_accuracy": 74.0,
"final_validation_loss": 1.0,
"driver_wall_seconds": 10.0,
"physical_gpu": {
"peak_process_gpu_memory_mib": 1000.0,
},
"work": {"synthetic": True},
})
return {
"gate": "pass",
"stage": "plain_cnn_p2",
"num_expected_records": 27,
"num_audited_records": 27,
"missing_experiments": [],
"test_policy": "none",
"records": records,
}
def native_report(family):
if family == "resnet":
stage = "resnet_crossover_r2"
cell_ids = [
f"resnet{depth}::{method}"
for depth in RESNET_DEPTHS for method in METHODS
]
else:
stage = "transformer_crossover_t2"
cell_ids = [
f"transformer{depth}::{method}"
for depth in TRANSFORMER_DEPTHS for method in METHODS
]
records = []
for cell_id in cell_ids:
common = {
"cell_id": cell_id,
"status": "completed",
"finite": True,
"driver_wall_seconds": 12.0,
"peak_memory_allocated_bytes": 1024 ** 3,
"work": {"synthetic": True},
}
if family == "resnet":
common.update({
"best_validation_accuracy": 0.8,
"final_validation_accuracy": 0.79,
"final_validation_loss": 0.9,
})
else:
common.update({
"final_validation_nll": 2.0,
"final_validation_perplexity": 7.389,
"final_validation_accuracy": 0.3,
})
records.append(common)
return {
"audit_status": "passed",
"stage": stage,
"complete_grid": True,
"failure_retaining": True,
"num_expected_cells": 27,
"num_audited_cells": 27,
"test_policy": "none",
"records": records,
}
def must_fail(plain, resnet, transformer):
try:
aggregate(plain, resnet, transformer)
except (AssertionError, KeyError):
return
raise AssertionError("invalid 81-cell panel was accepted")
def main():
plain = plain_report()
resnet = native_report("resnet")
transformer = native_report("transformer")
report = aggregate(plain, resnet, transformer)
assert report["num_audited_cells"] == 81
assert report["num_finite_cells"] == 81
assert report["failed_or_incomplete_cells"] == []
missing = copy.deepcopy(plain)
missing["records"].pop()
must_fail(missing, resnet, transformer)
duplicate = copy.deepcopy(transformer)
duplicate["records"][-1] = copy.deepcopy(duplicate["records"][0])
must_fail(plain, resnet, duplicate)
failed = copy.deepcopy(resnet)
cell = failed["records"][0]
cell.update({
"status": "timeout",
"finite": False,
"best_validation_accuracy": None,
"final_validation_accuracy": None,
"final_validation_loss": None,
})
report = aggregate(plain, failed, transformer)
assert report["num_audited_cells"] == 81
assert report["num_finite_cells"] == 80
assert report["failed_or_incomplete_cells"] == ["resnet20::bp"]
print("crossover 81-cell smoke: all checks passed")
if __name__ == "__main__":
main()
|
