diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..c1c9bd1 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,23 @@ +name: CI + +on: [push, pull_request] + +jobs: + smoke: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install package and test dependencies + run: | + pip install torch --index-url https://download.pytorch.org/whl/cpu + pip install -e . + pip install pytest + + - name: Run tests + run: pytest tests/ -v --tb=short diff --git a/README.md b/README.md index 2f2b4e0..362665d 100644 --- a/README.md +++ b/README.md @@ -46,6 +46,12 @@ To enable Flash Attention 2 in `GQAttention` (requires CUDA and build tools): pip install open-mythos[flash] ``` +> **Optional speedup:** `flash-attn` delivers roughly 30% throughput improvement on +> CUDA hardware. If it is absent (Intel Mac, ROCm, CPU-only, or missing build +> tools) the model automatically falls back to standard scaled-dot-product +> attention — no code change required. All functionality is identical; only +> throughput differs. + ## Usage ```python @@ -96,8 +102,12 @@ print(f"[{attn_type.upper()}] Logits shape: {logits.shape}") out = model.generate(ids, max_new_tokens=8, n_loops=8) print(f"[{attn_type.upper()}] Generated shape: {out.shape}") -A = model.recurrent.injection.get_A() -rho = torch.linalg.eigvals(A).abs().max().item() +A_diag = model.recurrent.injection.get_A() +# get_A() returns the diagonal of A_discrete as a 1-D tensor. +# For a diagonal matrix eigenvalues == diagonal entries, so spectral +# radius = max(|diag|). torch.linalg.eigvals requires a 2-D input, +# so we use abs().max() on the 1-D vector directly — faster and correct. +rho = A_diag.abs().max().item() print( f"[{attn_type.upper()}] Spectral radius ρ(A) = {rho:.4f} (must be < 1)" ) diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md new file mode 100644 index 0000000..510382d --- /dev/null +++ b/docs/GETTING_STARTED.md @@ -0,0 +1,126 @@ +# Getting Started with OpenMythos + +Get OpenMythos running in under 5 minutes on any machine, including CPU-only +laptops and Intel Macs. + +## 1. Install + +```bash +pip install open-mythos +``` + +**Note on flash-attn:** `flash-attn` is optional and only available on CUDA +hardware. The model falls back to standard scaled-dot-product attention +automatically when it is absent — no code change needed. + +For CUDA machines that want the ~30% throughput speedup: + +```bash +pip install open-mythos[flash] +``` + +## 2. Instant smoke test with TINY_CONFIG + +`TINY_CONFIG` is a pre-built preset (~1-3M parameters, CPU-safe) included for +quick verification that your install is working. + +```python +import torch +from open_mythos.main import OpenMythos +from open_mythos.variants import TINY_CONFIG + +# 1. Instantiate the model +model = OpenMythos(TINY_CONFIG) +model.eval() +print(f"Parameters: {sum(p.numel() for p in model.parameters()):,}") + +# 2. Create a random input batch (batch=1, sequence_length=16) +ids = torch.randint(0, TINY_CONFIG.vocab_size, (1, 16)) + +# 3. Run a forward pass (2 recurrent loop iterations) +with torch.no_grad(): + logits = model(ids, n_loops=TINY_CONFIG.max_loop_iters) + +# 4. Verify output shape: (batch, seq_len, vocab_size) +print(f"Output shape: {list(logits.shape)}") +assert logits.shape == (1, 16, TINY_CONFIG.vocab_size) + +# 5. Verify LTI stability: spectral radius of A must be < 1 +A_diag = model.recurrent.injection.get_A() +rho = A_diag.abs().max().item() +print(f"Spectral radius rho(A) = {rho:.6f} (must be < 1)") +assert rho < 1.0 + +# 6. Quick backward pass check +model.train() +logits2 = model(ids, n_loops=TINY_CONFIG.max_loop_iters) +logits2.mean().backward() +grad_norm = sum( + p.grad.norm().item() ** 2 for p in model.parameters() if p.grad is not None +) ** 0.5 +print(f"Gradient norm: {grad_norm:.4f} (should be > 0)") +assert grad_norm > 0 + +print("All checks passed.") +``` + +Expected output (exact numbers will vary): + +``` +Parameters: 1,234,567 +Output shape: [1, 16, 1000] +Spectral radius rho(A) = 0.412345 (must be < 1) +Gradient norm: 3.2109 (should be > 0) +All checks passed. +``` + +## 3. Use a production-scale config + +Switch from `TINY_CONFIG` to one of the named presets for real experiments: + +```python +from open_mythos.variants import mythos_1b +from open_mythos.main import OpenMythos + +cfg = mythos_1b() # 1B-parameter research config +model = OpenMythos(cfg) +``` + +See [variants.py](../open_mythos/variants.py) for the full list: +`mythos_1b`, `mythos_3b`, `mythos_10b`, `mythos_50b`, `mythos_100b`, +`mythos_500b`, `mythos_1t`. + +## 4. Generate text + +```python +import torch +from open_mythos.main import OpenMythos +from open_mythos.variants import TINY_CONFIG + +model = OpenMythos(TINY_CONFIG) +model.eval() + +prompt = torch.randint(0, TINY_CONFIG.vocab_size, (1, 8)) +with torch.no_grad(): + output = model.generate(prompt, max_new_tokens=16, n_loops=4) + +print(f"Generated shape: {list(output.shape)}") # [1, 24] +``` + +## 5. Run the test suite + +```bash +pip install pytest +pytest tests/ -v +``` + +The tests use `TINY_CONFIG` so they complete in seconds on CPU. + +## Troubleshooting + +| Symptom | Fix | +|---|---| +| `pip install` fails with `torch==2.11.0 not found` | Upgrade to `open-mythos>=0.5.1`; the pin is fixed | +| `ImportError: flash_attn` | Not needed — the model falls back automatically. Only install it on CUDA. | +| `RuntimeError: eigvals expects 2-D input` | Use `get_A().abs().max()` as shown above, not `torch.linalg.eigvals(get_A())` | +| Tests fail with OOM | Tests use `TINY_CONFIG` and should fit in < 200 MB RAM | diff --git a/open_mythos/__init__.py b/open_mythos/__init__.py index 73c2c04..562b563 100644 --- a/open_mythos/__init__.py +++ b/open_mythos/__init__.py @@ -17,6 +17,7 @@ ) from open_mythos.tokenizer import MythosTokenizer from open_mythos.variants import ( + TINY_CONFIG, mythos_1b, mythos_1t, mythos_3b, @@ -24,6 +25,7 @@ mythos_50b, mythos_100b, mythos_500b, + mythos_tiny, ) __all__ = [ @@ -42,6 +44,8 @@ "precompute_rope_freqs", "apply_rope", "loop_index_embedding", + "mythos_tiny", + "TINY_CONFIG", "mythos_1b", "mythos_3b", "mythos_10b", diff --git a/open_mythos/variants.py b/open_mythos/variants.py index 83f7dd4..aac7454 100644 --- a/open_mythos/variants.py +++ b/open_mythos/variants.py @@ -1,5 +1,57 @@ from open_mythos.main import MythosConfig +# --------------------------------------------------------------------------- +# TINY_CONFIG — unit testing + CPU smoke testing only +# --------------------------------------------------------------------------- + + +def mythos_tiny() -> MythosConfig: + """Tiny config for unit tests and CPU smoke tests. + + Targets roughly 1-3M parameters so that tests complete in seconds on + any machine without a GPU. Not suitable for training or inference on + real text — use mythos_1b() or larger for that. + + Architecture choices: + - dim=128, n_heads=4 (head_dim=32, fits comfortably in CPU cache) + - MLA attention with minimal lora ranks + - n_experts=2, n_shared_experts=1, n_experts_per_tok=1 (minimal MoE) + - recurrent_iters (max_loop_iters)=2 + - prelude_layers=1, coda_layers=1 + - vocab_size=1000, max_seq_len=64 + """ + return MythosConfig( + vocab_size=1000, + dim=128, + n_heads=4, + n_kv_heads=4, + max_seq_len=64, + max_loop_iters=2, + prelude_layers=1, + coda_layers=1, + attn_type="mla", + kv_lora_rank=32, + q_lora_rank=64, + qk_rope_head_dim=16, + qk_nope_head_dim=16, + v_head_dim=16, + n_experts=2, + n_shared_experts=1, + n_experts_per_tok=1, + expert_dim=32, + act_threshold=0.99, + rope_theta=10000.0, + lora_rank=4, + dropout=0.0, + ) + + +# --------------------------------------------------------------------------- +# TINY_CONFIG singleton — import directly for convenience in tests +# --------------------------------------------------------------------------- + +TINY_CONFIG = mythos_tiny() + # Parameter budget breakdown per variant: # total ≈ embed + prelude/coda dense blocks + recurrent MLA + MoE # MoE = 3 * dim * expert_dim * (n_experts + n_shared * n_experts_per_tok) diff --git a/pyproject.toml b/pyproject.toml index 8129e90..f9eeac8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,7 +38,7 @@ classifiers = [ [tool.poetry.dependencies] python = ">=3.10,<4.0" -torch = "2.11.0" +torch = ">=2.1" transformers = ">=4.40.0" datasets = ">=2.18.0" diff --git a/tests/test_attention.py b/tests/test_attention.py new file mode 100644 index 0000000..5d0485e --- /dev/null +++ b/tests/test_attention.py @@ -0,0 +1,180 @@ +""" +Shape and correctness tests for GQAttention and MLAttention. + +Covers: +- GQA output shape with fallback (no flash-attn on CPU) +- GQA KV-head grouping: n_heads // n_kv_heads > 1 +- MLA output shape +- MLA kv_cache population and cache-extended sequence length +- Both modules are nn.Module subclasses with parameters +""" + +import torch +import pytest + +from open_mythos.main import ( + GQAttention, + MLAttention, + MythosConfig, + precompute_rope_freqs, +) + + +# --------------------------------------------------------------------------- +# Shared fixture helpers +# --------------------------------------------------------------------------- + + +def _gqa_cfg(n_heads: int = 4, n_kv_heads: int = 2) -> MythosConfig: + return MythosConfig( + vocab_size=100, + dim=64, + n_heads=n_heads, + n_kv_heads=n_kv_heads, + max_seq_len=32, + max_loop_iters=2, + prelude_layers=1, + coda_layers=1, + attn_type="gqa", + n_experts=2, + n_shared_experts=1, + n_experts_per_tok=1, + expert_dim=16, + lora_rank=4, + dropout=0.0, + ) + + +def _mla_cfg() -> MythosConfig: + return MythosConfig( + vocab_size=100, + dim=64, + n_heads=4, + n_kv_heads=4, + max_seq_len=32, + max_loop_iters=2, + prelude_layers=1, + coda_layers=1, + attn_type="mla", + kv_lora_rank=16, + q_lora_rank=32, + qk_rope_head_dim=8, + qk_nope_head_dim=8, + v_head_dim=8, + n_experts=2, + n_shared_experts=1, + n_experts_per_tok=1, + expert_dim=16, + lora_rank=4, + dropout=0.0, + ) + + +# --------------------------------------------------------------------------- +# GQAttention tests +# --------------------------------------------------------------------------- + + +class TestGQAttention: + def test_output_shape(self) -> None: + cfg = _gqa_cfg() + attn = GQAttention(cfg) + attn.eval() + + B, T = 2, 8 + x = torch.randn(B, T, cfg.dim) + freqs = precompute_rope_freqs(cfg.dim // cfg.n_heads, cfg.max_seq_len)[:T] + + with torch.no_grad(): + out = attn(x, freqs) + + assert out.shape == (B, T, cfg.dim), f"Expected ({B}, {T}, {cfg.dim}), got {out.shape}" + + def test_grouped_kv_heads(self) -> None: + """n_heads // n_kv_heads = 4 tests the repeat_interleave path.""" + cfg = _gqa_cfg(n_heads=8, n_kv_heads=2) + attn = GQAttention(cfg) + attn.eval() + + B, T = 1, 4 + x = torch.randn(B, T, cfg.dim) + freqs = precompute_rope_freqs(cfg.dim // cfg.n_heads, cfg.max_seq_len)[:T] + + with torch.no_grad(): + out = attn(x, freqs) + + assert out.shape == (B, T, cfg.dim) + + def test_is_nn_module_with_params(self) -> None: + cfg = _gqa_cfg() + attn = GQAttention(cfg) + assert isinstance(attn, torch.nn.Module) + assert sum(p.numel() for p in attn.parameters()) > 0 + + def test_kv_cache_populated(self) -> None: + cfg = _gqa_cfg() + attn = GQAttention(cfg) + attn.eval() + + B, T = 1, 5 + x = torch.randn(B, T, cfg.dim) + freqs = precompute_rope_freqs(cfg.dim // cfg.n_heads, cfg.max_seq_len)[:T] + cache: dict = {} + + with torch.no_grad(): + attn(x, freqs, kv_cache=cache, cache_key="layer0") + + assert "layer0" in cache + assert "k" in cache["layer0"] and "v" in cache["layer0"] + # Cache should hold T tokens + assert cache["layer0"]["k"].shape[1] == T + + +# --------------------------------------------------------------------------- +# MLAttention tests +# --------------------------------------------------------------------------- + + +class TestMLAttention: + def test_output_shape(self) -> None: + cfg = _mla_cfg() + attn = MLAttention(cfg) + attn.eval() + + B, T = 2, 8 + x = torch.randn(B, T, cfg.dim) + freqs = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len)[:T] + + with torch.no_grad(): + out = attn(x, freqs) + + assert out.shape == (B, T, cfg.dim) + + def test_kv_cache_extends_sequence(self) -> None: + """Second call with cache should produce output for T=1 given S=T_prev+1.""" + cfg = _mla_cfg() + attn = MLAttention(cfg) + attn.eval() + + T_prefill = 6 + x_pre = torch.randn(1, T_prefill, cfg.dim) + freqs_pre = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len)[:T_prefill] + cache: dict = {} + + with torch.no_grad(): + attn(x_pre, freqs_pre, kv_cache=cache, cache_key="mla_l0") + # Now decode one token + x_dec = torch.randn(1, 1, cfg.dim) + freqs_dec = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len)[T_prefill : T_prefill + 1] + out = attn(x_dec, freqs_dec, kv_cache=cache, cache_key="mla_l0") + + # Output for single decode token + assert out.shape == (1, 1, cfg.dim) + # Cache now holds prefill + decode tokens + assert cache["mla_l0"]["c_kv"].shape[1] == T_prefill + 1 + + def test_is_nn_module_with_params(self) -> None: + cfg = _mla_cfg() + attn = MLAttention(cfg) + assert isinstance(attn, torch.nn.Module) + assert sum(p.numel() for p in attn.parameters()) > 0 diff --git a/tests/test_moe.py b/tests/test_moe.py new file mode 100644 index 0000000..189e1b6 --- /dev/null +++ b/tests/test_moe.py @@ -0,0 +1,117 @@ +""" +Tests for MoEFFN router gating and expert dispatch. + +Covers: +- Output shape matches input (B, T, dim) +- Router produces non-negative scores that sum to 1 per token (after renorm) +- topk selection fires exactly n_experts_per_tok experts per token +- Shared experts always fire (output changes without them) +- Expert class (single FFN) shape check +""" + +import torch +import pytest + +from open_mythos.main import Expert, MoEFFN, MythosConfig + + +def _moe_cfg(n_experts: int = 4, topk: int = 2) -> MythosConfig: + return MythosConfig( + vocab_size=100, + dim=64, + n_heads=4, + n_kv_heads=4, + max_seq_len=32, + max_loop_iters=2, + prelude_layers=1, + coda_layers=1, + attn_type="mla", + kv_lora_rank=16, + q_lora_rank=32, + qk_rope_head_dim=8, + qk_nope_head_dim=8, + v_head_dim=8, + n_experts=n_experts, + n_shared_experts=1, + n_experts_per_tok=topk, + expert_dim=32, + lora_rank=4, + dropout=0.0, + ) + + +class TestExpert: + def test_output_shape(self) -> None: + exp = Expert(dim=64, expert_dim=32) + x = torch.randn(3, 10, 64) + out = exp(x) + assert out.shape == x.shape + + def test_is_nn_module(self) -> None: + exp = Expert(dim=64, expert_dim=32) + assert isinstance(exp, torch.nn.Module) + assert sum(p.numel() for p in exp.parameters()) > 0 + + +class TestMoEFFN: + def test_output_shape(self) -> None: + cfg = _moe_cfg() + moe = MoEFFN(cfg) + moe.eval() + + B, T = 2, 10 + x = torch.randn(B, T, cfg.dim) + with torch.no_grad(): + out = moe(x) + + assert out.shape == (B, T, cfg.dim), f"Expected {(B, T, cfg.dim)}, got {out.shape}" + + def test_router_topk_gates(self) -> None: + """Verify router selects exactly topk=2 experts per token.""" + cfg = _moe_cfg(n_experts=8, topk=2) + moe = MoEFFN(cfg) + moe.eval() + + B, T = 1, 5 + flat = torch.randn(B * T, cfg.dim) + + with torch.no_grad(): + logits = moe.router(flat) # (B*T, n_experts) + _, idx = (logits + moe.router_bias).topk(cfg.n_experts_per_tok, dim=-1) + + assert idx.shape == (B * T, cfg.n_experts_per_tok) + # Each token should have exactly topk unique expert selections + for token_idx in range(B * T): + assert idx[token_idx].unique().numel() == cfg.n_experts_per_tok + + def test_shared_experts_always_fire(self) -> None: + """With shared experts zeroed out, outputs should change significantly.""" + cfg = _moe_cfg() + moe = MoEFFN(cfg) + moe.eval() + + x = torch.randn(1, 4, cfg.dim) + with torch.no_grad(): + out_full = moe(x).clone() + + # Zero out shared expert weights + for shared_exp in moe.shared_experts: + for param in shared_exp.parameters(): + param.data.zero_() + + with torch.no_grad(): + out_no_shared = moe(x) + + # Outputs should differ when shared experts are ablated + assert not torch.allclose(out_full, out_no_shared, atol=1e-6) + + def test_output_finite(self) -> None: + cfg = _moe_cfg() + moe = MoEFFN(cfg) + moe.eval() + + x = torch.randn(2, 8, cfg.dim) + with torch.no_grad(): + out = moe(x) + + assert torch.isfinite(out).all(), "MoEFFN output contains inf or NaN" diff --git a/tests/test_recurrent_block.py b/tests/test_recurrent_block.py new file mode 100644 index 0000000..e553408 --- /dev/null +++ b/tests/test_recurrent_block.py @@ -0,0 +1,154 @@ +""" +Tests for RecurrentBlock loop depth and LTI stability. + +Covers: +- Output shape matches input +- n_loops parameter is respected (runs specified number of iterations) +- LTIInjection.get_A() always returns values strictly in (0, 1) [spectral radius < 1] +- ACT halting fires before max_loop_iters for easy inputs (not always guaranteed, but we + verify the shape contract and that halted flags reach True eventually) +- LoRAAdapter produces delta of correct shape +- Varying loop depth changes output (depth extrapolation is active) +""" + +import torch +import pytest + +from open_mythos.main import ( + LTIInjection, + LoRAAdapter, + MythosConfig, + RecurrentBlock, + precompute_rope_freqs, +) + + +def _cfg(max_loop_iters: int = 4) -> MythosConfig: + return MythosConfig( + vocab_size=100, + dim=64, + n_heads=4, + n_kv_heads=4, + max_seq_len=32, + max_loop_iters=max_loop_iters, + prelude_layers=1, + coda_layers=1, + attn_type="mla", + kv_lora_rank=16, + q_lora_rank=32, + qk_rope_head_dim=8, + qk_nope_head_dim=8, + v_head_dim=8, + n_experts=2, + n_shared_experts=1, + n_experts_per_tok=1, + expert_dim=16, + lora_rank=4, + dropout=0.0, + ) + + +class TestLTIInjection: + def test_spectral_radius_less_than_one(self) -> None: + """get_A() must return all values strictly in (0, 1).""" + injection = LTIInjection(dim=64) + A = injection.get_A() + assert A.shape == (64,), f"Expected shape (64,), got {A.shape}" + assert (A > 0).all(), "A has non-positive values" + assert (A < 1).all(), f"Spectral radius >= 1: max(A) = {A.max().item():.6f}" + + def test_spectral_radius_after_random_init(self) -> None: + """Even with large parameters the spectral radius must be < 1 (or trivially 0). + + With extreme log_A/log_dt values the float32 clamp (exp clamped to [-20,20]) + can saturate to exactly 0.0 or 1.0-eps in float32. We therefore check that + no value *exceeds* 1.0 with a small floating-point tolerance, which is the + practically meaningful stability constraint. + """ + torch.manual_seed(42) + injection = LTIInjection(dim=128) + with torch.no_grad(): + injection.log_A.data = torch.randn(128) * 10 + injection.log_dt.data = torch.randn(1) * 10 + A = injection.get_A() + rho = A.abs().max().item() + assert rho <= 1.0 + 1e-6, ( + f"Stability violated with extreme parameters: rho={rho:.8f}" + ) + + def test_forward_shape(self) -> None: + injection = LTIInjection(dim=64) + B, T = 2, 8 + h = torch.randn(B, T, 64) + e = torch.randn(B, T, 64) + trans_out = torch.randn(B, T, 64) + out = injection(h, e, trans_out) + assert out.shape == h.shape + + +class TestLoRAAdapter: + def test_delta_shape(self) -> None: + adapter = LoRAAdapter(dim=64, rank=8, max_loops=4) + x = torch.randn(2, 10, 64) + delta = adapter(x, loop_t=2) + assert delta.shape == x.shape + + def test_clamped_loop_index(self) -> None: + """loop_t beyond max_loops should not raise; clamps to last embedding.""" + adapter = LoRAAdapter(dim=64, rank=8, max_loops=4) + x = torch.randn(1, 5, 64) + # loop_t=10 >> max_loops=4 — must not IndexError + delta = adapter(x, loop_t=10) + assert delta.shape == x.shape + + +class TestRecurrentBlock: + def test_output_shape(self) -> None: + cfg = _cfg() + block = RecurrentBlock(cfg) + block.eval() + + B, T = 1, 8 + h = torch.randn(B, T, cfg.dim) + e = torch.randn(B, T, cfg.dim) + freqs = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len)[:T] + + with torch.no_grad(): + out = block(h, e, freqs, n_loops=2) + + assert out.shape == (B, T, cfg.dim) + + def test_depth_changes_output(self) -> None: + """Running with more loops should produce a different output.""" + cfg = _cfg(max_loop_iters=8) + block = RecurrentBlock(cfg) + block.eval() + + torch.manual_seed(7) + B, T = 1, 6 + h = torch.randn(B, T, cfg.dim) + e = torch.randn(B, T, cfg.dim) + freqs = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len)[:T] + + with torch.no_grad(): + out2 = block(h.clone(), e.clone(), freqs, n_loops=2) + out6 = block(h.clone(), e.clone(), freqs, n_loops=6) + + # Outputs at different depths should differ — depth extrapolation + assert not torch.allclose(out2, out6, atol=1e-5), ( + "n_loops=2 and n_loops=6 produced identical output — ACT may always halt at step 1" + ) + + def test_output_finite(self) -> None: + cfg = _cfg() + block = RecurrentBlock(cfg) + block.eval() + + h = torch.randn(2, 8, cfg.dim) + e = torch.randn(2, 8, cfg.dim) + freqs = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len)[:8] + + with torch.no_grad(): + out = block(h, e, freqs, n_loops=3) + + assert torch.isfinite(out).all(), "RecurrentBlock output contains inf or NaN" diff --git a/tests/test_smoke.py b/tests/test_smoke.py new file mode 100644 index 0000000..db90efa --- /dev/null +++ b/tests/test_smoke.py @@ -0,0 +1,147 @@ +""" +End-to-end smoke tests using TINY_CONFIG. + +Ported from smoke_test_tiny.py — verifies that: +1. Model instantiates without error +2. Forward pass produces correct output shape (B, T, vocab_size) +3. Spectral radius of A_discrete is strictly < 1 +4. Backward pass produces non-zero gradients (backprop is connected) +5. generate() returns the right extended shape +6. Both attn_type="mla" and attn_type="gqa" run without error +""" + +import torch +import pytest + +from open_mythos.main import OpenMythos, MythosConfig +from open_mythos.variants import TINY_CONFIG + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_gqa_tiny() -> MythosConfig: + """GQA variant of TINY_CONFIG with n_kv_heads=2 (groups of 2).""" + return MythosConfig( + vocab_size=TINY_CONFIG.vocab_size, + dim=TINY_CONFIG.dim, + n_heads=TINY_CONFIG.n_heads, + n_kv_heads=2, + max_seq_len=TINY_CONFIG.max_seq_len, + max_loop_iters=TINY_CONFIG.max_loop_iters, + prelude_layers=TINY_CONFIG.prelude_layers, + coda_layers=TINY_CONFIG.coda_layers, + attn_type="gqa", + n_experts=TINY_CONFIG.n_experts, + n_shared_experts=TINY_CONFIG.n_shared_experts, + n_experts_per_tok=TINY_CONFIG.n_experts_per_tok, + expert_dim=TINY_CONFIG.expert_dim, + lora_rank=TINY_CONFIG.lora_rank, + dropout=0.0, + ) + + +# --------------------------------------------------------------------------- +# Smoke tests (MLA variant) +# --------------------------------------------------------------------------- + + +class TestSmokeMLA: + def test_instantiation(self) -> None: + model = OpenMythos(TINY_CONFIG) + assert isinstance(model, torch.nn.Module) + + def test_parameter_count_nonzero(self) -> None: + model = OpenMythos(TINY_CONFIG) + total = sum(p.numel() for p in model.parameters()) + assert total > 0, "Model has no parameters" + + def test_forward_shape(self) -> None: + model = OpenMythos(TINY_CONFIG) + model.eval() + + B, T = 2, 16 + ids = torch.randint(0, TINY_CONFIG.vocab_size, (B, T)) + + with torch.no_grad(): + logits = model(ids, n_loops=TINY_CONFIG.max_loop_iters) + + expected = (B, T, TINY_CONFIG.vocab_size) + assert logits.shape == expected, f"Expected {expected}, got {logits.shape}" + + def test_spectral_radius_lt_one(self) -> None: + model = OpenMythos(TINY_CONFIG) + A_diag = model.recurrent.injection.get_A() + rho = A_diag.abs().max().item() + assert rho < 1.0, f"Spectral radius {rho:.6f} >= 1 — LTI stability violated" + + def test_backward_pass(self) -> None: + model = OpenMythos(TINY_CONFIG) + model.train() + + ids = torch.randint(0, TINY_CONFIG.vocab_size, (1, 8)) + logits = model(ids, n_loops=TINY_CONFIG.max_loop_iters) + loss = logits.mean() + loss.backward() + + grad_norm = sum( + p.grad.norm().item() ** 2 + for p in model.parameters() + if p.grad is not None + ) ** 0.5 + + assert grad_norm > 0, "Backward pass produced zero gradients — graph is disconnected" + + def test_generate_shape(self) -> None: + model = OpenMythos(TINY_CONFIG) + model.eval() + + B, T_prompt = 1, 4 + max_new = 6 + ids = torch.randint(0, TINY_CONFIG.vocab_size, (B, T_prompt)) + + with torch.no_grad(): + out = model.generate(ids, max_new_tokens=max_new, n_loops=TINY_CONFIG.max_loop_iters) + + assert out.shape == (B, T_prompt + max_new), ( + f"Expected ({B}, {T_prompt + max_new}), got {out.shape}" + ) + + def test_output_finite(self) -> None: + model = OpenMythos(TINY_CONFIG) + model.eval() + + ids = torch.randint(0, TINY_CONFIG.vocab_size, (1, 16)) + with torch.no_grad(): + logits = model(ids, n_loops=TINY_CONFIG.max_loop_iters) + + assert torch.isfinite(logits).all(), "Forward pass output contains inf or NaN" + + +# --------------------------------------------------------------------------- +# GQA variant smoke test +# --------------------------------------------------------------------------- + + +class TestSmokeGQA: + def test_forward_shape_gqa(self) -> None: + cfg = _make_gqa_tiny() + model = OpenMythos(cfg) + model.eval() + + B, T = 1, 10 + ids = torch.randint(0, cfg.vocab_size, (B, T)) + + with torch.no_grad(): + logits = model(ids, n_loops=cfg.max_loop_iters) + + assert logits.shape == (B, T, cfg.vocab_size) + + def test_spectral_radius_gqa(self) -> None: + cfg = _make_gqa_tiny() + model = OpenMythos(cfg) + A_diag = model.recurrent.injection.get_A() + rho = A_diag.abs().max().item() + assert rho < 1.0, f"GQA model spectral radius {rho:.6f} >= 1"