diff --git a/docs/design/verified-scope-solver.md b/docs/design/verified-scope-solver.md index a8b73681..45b4620b 100644 --- a/docs/design/verified-scope-solver.md +++ b/docs/design/verified-scope-solver.md @@ -484,6 +484,30 @@ the closure. certificate. - Semantic fields never take the autoregressive path. +## Implemented constrained autoregressive surface realizer (VSS3-05 / SLM-73) + +[`models/surface_autoregressor.py`](../../src/slm_training/models/surface_autoregressor.py) +adds a small standalone causal byte-decoder with printable-ASCII byte tokens, +grammar-aware `IdentifierConstraint`, permissive `DecorativeConstraint`, prompt +context encoding, and constrained greedy/top-k generation. It is kept separate +from the main TwoTower/diffusion paths so it cannot become a second program +generator. + +[`dsl/neural_surface_realizer.py`](../../src/slm_training/dsl/neural_surface_realizer.py) +wraps the model as a `SurfaceRealizer`. It accepts only `SURFACE_ONLY` slots, +only `INTERNAL_IDENTIFIER` and `DECORATIVE_TEXT` kinds, generates each slot +under its own constraint mask, and falls back to the deterministic baseline +per slot on dead end, invalid proposal, model error, or missing model. +`realize_surface_and_verify` still canonicalizes and re-verifies the final +program with the pack oracle. The neural realizer is a caller-supplied +`SurfaceRealizer`, not a new pack slot; the default pipeline keeps +`DeterministicSurfaceRealizer`. + +This is fixture wiring only (`tests/test_models/test_surface_autoregressor.py`, +`tests/test_dsl/test_neural_surface_realizer.py`, and +`tests/test_data/test_surface_rows.py`). The deterministic baseline remains the +default; no full train/eval run, checkpoint, or ship gate is claimed. + ## Relationship to existing implementation | Symbol (file:line) | Existing role | Disposition under this contract | diff --git a/docs/design/vss3-05-autoregressive-surface-realizer.md b/docs/design/vss3-05-autoregressive-surface-realizer.md new file mode 100644 index 00000000..61fa088b --- /dev/null +++ b/docs/design/vss3-05-autoregressive-surface-realizer.md @@ -0,0 +1,140 @@ +# VSS3-05: Constrained autoregressive surface realizer with deterministic fallback + +**Issue:** SLM-73 + +**Status:** wiring / fixture evidence. No trained production checkpoint, no +full-corpus train/eval run, and no `--ship-gates` claim. Deterministic surface +realization remains the default and baseline. + +## What was added + +A small, standalone causal byte-autoregressor for surface-only slots, kept +outside the main TwoTower/diffusion program generator so it cannot accidentally +become a second generator path. + +`src/slm_training/models/surface_autoregressor.py`: + +- `SurfaceByteVocab` — printable-ASCII byte vocabulary (``, ``, + ``, ``, `B:xx`). Any lowercase identifier or decorative ASCII text + can be spelled token-by-token without a corpus-bound tokenizer. +- `IdentifierConstraint` and `DecorativeConstraint` — constrained next-token + sets for OpenUI-style internal identifiers and decorative text. Enforce + grammar, reserved-word avoidance, peer uniqueness, and byte budgets at decode + time. +- Small causal decoder (`CausalSelfAttention`, `SurfaceTransformerBlock`, + `SurfaceContextEncoder`, `SurfaceAutoregressor`) with prompt context encoding, + cross-attention, weight tying, and constrained greedy/top-k generation. +- `train_surface_autoregressor` — tiny fixture trainer for (prompt, target) + pairs, used by tests and future harnesses. +- Save/load and `from_records` for API symmetry. + +`src/slm_training/dsl/neural_surface_realizer.py`: + +- `NeuralSurfaceRealizerConfig` — runtime knobs (`model`, `model_path`, + `device`, `max_bytes`, `temperature`, `top_k`, `seed`, `fallback_to_deterministic`). +- `NeuralSurfaceRealizer` — a `SurfaceRealizer` backed by the AR model. + - Only accepts slots already classified as `SURFACE_ONLY`. + - Only supports `INTERNAL_IDENTIFIER` and `DECORATIVE_TEXT` kinds. + - Generates each slot independently under its own constraint mask. + - Falls back to `DeterministicSurfaceRealizer` per slot on dead end, invalid + proposal, model error, or when no model is configured. + - Rejects non-`SURFACE_ONLY` slots before the model is invoked. + +`src/slm_training/harnesses/model_build/config.py`: + +- Default-off AR fields for API symmetry and future harness wiring: + `surface_realizer`, `surface_ar_enabled`, `surface_ar_d_model`, + `surface_ar_n_layers`, `surface_ar_n_heads`, `surface_ar_max_bytes`, + `surface_ar_temperature`, `surface_ar_top_k`, `surface_ar_fallback`, + `surface_ar_verify_retry`. Deterministic realization remains the default. + +`src/slm_training/dsl/schema.py`: + +- Added `"surface_realization"` to `TASK_TOKENS` so derived training records can + carry the correct task label. + +`src/slm_training/data/progspec/surface_rows.py`: + +- `derive_surface_realization_records(spec, *, realizer, opaque_bindings, + include_authorities)` — derives one `ExampleRecord` per realized surface slot + from a verified `ProgramSpec`, preserving `split` and `split_group_id`. + Defaults to the deterministic baseline when `realizer` is `None` and only + emits authorities in `include_authorities` (`SURFACE_ONLY` and + `OPAQUE_USER_VALUE` by default). + +Regression tests: + +- `tests/test_models/test_surface_autoregressor.py` — byte vocab round-trip, + identifier/decorative constraints, fixture overfit, save/load, `from_records` + (torch-only; skipped where torch is unavailable). +- `tests/test_dsl/test_neural_surface_realizer.py` — neural realizer plugged into + `realize_surface_and_verify`, trained-identifier verification, no-model + deterministic fallback, dead-end fallback, disabled fallback error, + authority/kind rejection. The no-model and guard paths are torch-free; the + model-backed paths are torch-only. +- `tests/test_data/test_surface_rows.py` — `derive_surface_realization_records` + emits one record per surface slot, respects `include_authorities`, inherits + split/group (torch-free), and can use a trained neural realizer (torch-only). + +## What is intentionally not added + +- The AR realizer is **not** registered as a full `ModelPlugin` in + `factory.py`. That would create a second program generator, which this issue + forbids. +- The `ModelBuildConfig` fields are wiring placeholders; no train/eval harness + automatically instantiates or trains the AR realizer yet. +- No production checkpoint, no full-corpus run, and no ship-gate claim. + +## Design boundaries preserved + +- `DeterministicSurfaceRealizer` remains the default. +- Autoregression is allowed only for slots the pack classifier marks as + `SURFACE_ONLY`. +- Each slot is generated under a hard constraint mask; the final program is + still canonicalized and re-verified by `pack.oracle`. +- Fallback is per-slot and transparent in assignment provenance. +- Fixture-only; meaningful-parse and full ship gates remain the production bar. + +## End-to-end example + +Input (solved semantic IR, placeholders preserved): + +```text +root = Stack([title], "column") +title = TextContent(":hero.title") +``` + +A tiny fixture-trained model maps the binder slot prompt +`kind=internal_identifier authority=surface_only slot_id=openui:binder:title +symbol=title max=64` to the identifier `title`. `realize_surface_and_verify` +substitutes `title` for the semantic symbol, splices the opaque content +`:user.title`, canonicalizes, and verifies: + +```text +root = Stack([v0], "column") +v0 = TextContent(":user.title") +``` + +The assignment carries the model-chosen value and provenance: + +```text +SurfaceAssignment( + slot_id='openui:binder:title', + value='title', + provenance='autoregressive', +) +``` + +If the model is absent or fails, the same slot falls back to deterministic +`v0` with provenance `autoregressive_fallback::deterministic:canonical_name`. + +## Caveats + +- The model is a scratch causal decoder; real capacity and corpus training are + future matrix work. +- `DecorativeConstraint` accepts all printable ASCII, but OpenUI V1 decorative + text slots are currently classified as `OPAQUE_USER_VALUE`, so the AR path for + decorative text is wired but not exercised end-to-end by the default pack. +- Constrained generation is greedy/top-k; sampling temperature > 0 is supported + but not ship-gated. +- No checkpoint, no full `--ship-gates` run, and no production claim. diff --git a/src/slm_training/data/progspec/surface_rows.py b/src/slm_training/data/progspec/surface_rows.py new file mode 100644 index 00000000..a6a034be --- /dev/null +++ b/src/slm_training/data/progspec/surface_rows.py @@ -0,0 +1,104 @@ +"""Surface-realization training rows for VSS3-05 (SLM-73). + +Derives ``ExampleRecord`` rows from verified ``ProgramSpec`` roots after split +assignment. Each row represents one surface slot realization target. Split and +group identity are inherited from the parent spec so leakage checks remain +intact. +""" + +from __future__ import annotations + +import hashlib +from typing import Any + +from slm_training.data.progspec.schema import ProgramSpec, emit_record +from slm_training.dsl.surface import ( + SurfaceAuthority, + SurfaceRealizer, + realize_surface_and_verify, + resolve_surface_slot_extractor, +) + + +def _semantic_fingerprint(spec: ProgramSpec) -> str: + """Stable fingerprint of the solved semantic IR.""" + payload = f"{spec.program_family_id}:{spec.lineage_id}:{spec.canonical_openui}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] + + +def _surface_prompt(spec: ProgramSpec, slot: Any) -> str: + """Short prompt describing the slot for training/conditioning.""" + parts = [ + f"kind={slot.kind.value}", + f"symbol={slot.semantic_symbol_id or ''}", + f"slot={slot.slot_id}", + ] + return " ".join(parts) + + +def derive_surface_realization_records( + spec: ProgramSpec, + *, + realizer: SurfaceRealizer | None = None, + opaque_bindings: dict[str, Any] | None = None, + include_authorities: frozenset[str] | None = None, +) -> tuple[Any, ...]: + """Derive one ``ExampleRecord`` per realized surface slot. + + The deterministic baseline is used when ``realizer`` is ``None``. Only slots + whose authority is in ``include_authorities`` are emitted (default: + ``SURFACE_ONLY`` and ``OPAQUE_USER_VALUE``). + """ + from slm_training.dsl.pack import get_pack + + include_authorities = include_authorities or frozenset( + {SurfaceAuthority.SURFACE_ONLY.value, SurfaceAuthority.OPAQUE_USER_VALUE.value} + ) + pack = get_pack("openui") + result = realize_surface_and_verify( + spec, + pack=pack, + realizer=realizer, + opaque_bindings=opaque_bindings or {}, + semantic_ir_fingerprint=_semantic_fingerprint(spec), + prior_status="verified", + ) + if result.status not in {"solved", "verified"}: + return () + + assignment_by_slot = {a.slot_id: a for a in result.assignments} + # Main resolves the OpenUI extractor via the pack_id registry; the pack does + # not expose a ``surface_slot_extractor`` attribute directly. + extractor = resolve_surface_slot_extractor(pack) + if extractor is None: + return () + slots = extractor(spec.canonical_openui) + records: list[Any] = [] + for slot in slots: + if slot.authority.value not in include_authorities: + continue + assignment = assignment_by_slot.get(slot.slot_id) + if assignment is None: + continue + prompt = _surface_prompt(spec, slot) + record_id = f"{spec.id}_surface_{slot.slot_id}_{hashlib.sha256(assignment.value.encode()).hexdigest()[:8]}" + record = emit_record( + spec, + prompt=prompt, + task="surface_realization", + openui=result.source, + record_id=record_id, + source="surface_realization", + determinacy="deterministic", + tier="Silver", + provenance={ + "surface_slot_id": slot.slot_id, + "surface_assignment": assignment.to_dict(), + "surface_result_status": result.status, + }, + ) + records.append(record) + return tuple(records) + + +__all__ = ["derive_surface_realization_records"] diff --git a/src/slm_training/dsl/neural_surface_realizer.py b/src/slm_training/dsl/neural_surface_realizer.py new file mode 100644 index 00000000..b8ea81e1 --- /dev/null +++ b/src/slm_training/dsl/neural_surface_realizer.py @@ -0,0 +1,210 @@ +"""Neural SurfaceRealizer with deterministic fallback for VSS3-05 (SLM-73). + +The autoregressive realizer only handles slots already classified as +``SURFACE_ONLY`` by the pack classifier. Unsupported or invalid proposals fall +back to ``DeterministicSurfaceRealizer`` per slot, and the final assembled +program is still globally re-verified by ``realize_surface_and_verify``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from slm_training.dsl.surface import ( + DeterministicSurfaceRealizer, + SurfaceAssignment, + SurfaceAuthority, + SurfaceRealizationRequest, + SurfaceSlotKind, +) + +# torch and the AR model module are imported lazily inside the methods that need +# them (mirroring main's ``models/grammar.py`` / ``models/causal_lm_openui.py``). +# This keeps ``NeuralSurfaceRealizerConfig`` and the no-model deterministic +# fallback path importable and exercisable without torch installed. +if TYPE_CHECKING: + import torch + + from slm_training.models.surface_autoregressor import ( + DecorativeConstraint, + IdentifierConstraint, + SurfaceAutoregressor, + ) + + +@dataclass(frozen=True) +class NeuralSurfaceRealizerConfig: + """Runtime knobs for the neural surface realizer.""" + + model: SurfaceAutoregressor | None = None + model_path: str | None = None + device: str = "cpu" + max_bytes: int = 64 + temperature: float = 0.0 + top_k: int = 1 + seed: int | None = None + fallback_to_deterministic: bool = True + + +class NeuralSurfaceRealizer: + """SurfaceRealizer backed by a small causal byte-autoregressor. + + Unsupported slot kinds are rejected before the model is invoked. Each slot + is generated independently under its own constraint mask. If the model + proposes an invalid value, hits a dead end, or raises an exception, the + realizer optionally falls back to ``DeterministicSurfaceRealizer`` for that + single slot and records the reason. + """ + + def __init__(self, config: NeuralSurfaceRealizerConfig) -> None: + self.config = config + self._fallback = DeterministicSurfaceRealizer() + self._model = self._load_model() + # The vocab lives on the (torch) model; the no-model fallback path never + # touches it, so leave it ``None`` to stay torch-free without a model. + self._vocab = self._model.vocab if self._model is not None else None + + def _load_model(self) -> SurfaceAutoregressor | None: + if self.config.model is not None: + return self.config.model.to(self.config.device) + if self.config.model_path is not None: + from slm_training.models.surface_autoregressor import SurfaceAutoregressor + + return SurfaceAutoregressor.load(self.config.model_path, device=self.config.device) + # No model: every supported slot will fall back to deterministic. + return None + + def _build_prompt(self, slot: Any, peers: set[str]) -> str: + """Encode slot metadata and namespace context as a short prompt.""" + parts = [ + f"kind={slot.kind.value}", + f"authority={slot.authority.value}", + f"slot_id={slot.slot_id}", + ] + if slot.semantic_symbol_id is not None: + parts.append(f"symbol={slot.semantic_symbol_id}") + max_bytes = slot.constraints.max_bytes or self.config.max_bytes + parts.append(f"max={max_bytes}") + if slot.constraints.reserved: + parts.append(f"reserved={','.join(slot.constraints.reserved)}") + if peers: + parts.append(f"peers={','.join(sorted(peers))}") + return " ".join(parts) + + def _encode_prompt(self, prompt: str) -> torch.Tensor: + import torch + + ids = self._vocab.encode(prompt, add_special=True) + return torch.tensor(ids, dtype=torch.long) + + def _make_constraint(self, slot: Any, peers: set[str]) -> IdentifierConstraint | DecorativeConstraint: + from slm_training.models.surface_autoregressor import ( + DecorativeConstraint, + IdentifierConstraint, + ) + + max_bytes = slot.constraints.max_bytes or self.config.max_bytes + reserved = set(slot.constraints.reserved) + if slot.kind is SurfaceSlotKind.INTERNAL_IDENTIFIER: + return IdentifierConstraint( + self._vocab, + max_bytes=max_bytes, + reserved=reserved, + peers=peers, + ) + if slot.kind is SurfaceSlotKind.DECORATIVE_TEXT: + return DecorativeConstraint(self._vocab, max_bytes=max_bytes) + raise ValueError(f"unsupported slot kind {slot.kind.value}") + + def realize(self, request: SurfaceRealizationRequest) -> tuple[SurfaceAssignment, ...]: + """Generate surface assignments with per-slot deterministic fallback.""" + assignments: list[SurfaceAssignment] = [] + peers: set[str] = set() + + for slot in request.slots: + # Reject anything not explicitly approved for AR surface generation. + if slot.authority is not SurfaceAuthority.SURFACE_ONLY: + raise ValueError( + f"slot {slot.slot_id!r} has authority {slot.authority.value}; " + "only SURFACE_ONLY slots may be realized by the AR model" + ) + if slot.kind not in { + SurfaceSlotKind.INTERNAL_IDENTIFIER, + SurfaceSlotKind.DECORATIVE_TEXT, + }: + raise ValueError( + f"slot {slot.slot_id!r} ({slot.kind.value}) is not supported by " + "the autoregressive surface realizer" + ) + + assignment = self._realize_one(slot, peers, request) + assignments.append(assignment) + peers.add(assignment.value) + + return tuple(assignments) + + def _realize_one( + self, slot: Any, peers: set[str], request: SurfaceRealizationRequest + ) -> SurfaceAssignment: + """Generate one slot, falling back on invalid/dead-end/model error.""" + fallback_reason: str | None = None + value: str | None = None + provenance = "autoregressive" + + if self._model is not None: + try: + prompt = self._build_prompt(slot, peers) + prompt_ids = self._encode_prompt(prompt) + constraint = self._make_constraint(slot, peers) + value = self._model.generate( + prompt_ids, + constraint, + max_bytes=slot.constraints.max_bytes or self.config.max_bytes, + temperature=self.config.temperature, + top_k=self.config.top_k, + seed=self.config.seed, + ) + if value is None: + fallback_reason = "dead_end" + elif not constraint.is_complete(value): + fallback_reason = "invalid_proposal" + value = None + else: + provenance = "autoregressive" + except Exception as exc: # noqa: BLE001 + fallback_reason = f"model_error:{type(exc).__name__}" + value = None + else: + fallback_reason = "no_model" + + if value is None: + if not self.config.fallback_to_deterministic: + raise ValueError( + f"slot {slot.slot_id!r} could not be realized and fallback is disabled" + ) + # Fall back to deterministic for this single slot. + sub_request = SurfaceRealizationRequest( + pack_id=request.pack_id, + constraint_version=request.constraint_version, + semantic_ir_fingerprint=request.semantic_ir_fingerprint, + slots=(slot,), + context=request.context, + ) + fb_assignments = self._fallback.realize(sub_request) + if not fb_assignments: + raise ValueError( + f"slot {slot.slot_id!r}: AR failed ({fallback_reason}) and " + "deterministic fallback produced no assignment" + ) + assignment = fb_assignments[0] + return SurfaceAssignment( + slot_id=assignment.slot_id, + value=assignment.value, + provenance=f"autoregressive_fallback:{fallback_reason}:{assignment.provenance}", + ) + + return SurfaceAssignment(slot_id=slot.slot_id, value=value, provenance=provenance) + + +__all__ = ["NeuralSurfaceRealizer", "NeuralSurfaceRealizerConfig"] diff --git a/src/slm_training/dsl/schema.py b/src/slm_training/dsl/schema.py index afd26a74..82867354 100644 --- a/src/slm_training/dsl/schema.py +++ b/src/slm_training/dsl/schema.py @@ -26,6 +26,7 @@ "noop", "adversarial", "identity", + "surface_realization", } ) diff --git a/src/slm_training/harnesses/model_build/config.py b/src/slm_training/harnesses/model_build/config.py index 93e96524..94b1a29d 100644 --- a/src/slm_training/harnesses/model_build/config.py +++ b/src/slm_training/harnesses/model_build/config.py @@ -274,6 +274,18 @@ class ModelBuildConfig: sync_checkpoints: bool | None = False # Plan-only sync (no upload) — for wiring tests / agents without write auth. checkpoint_bucket_dry_run: bool = False + # VSS3-05: optional constrained autoregressive surface realizer. + # All fields are default-off; deterministic realization remains the baseline. + surface_realizer: str = "deterministic" # deterministic | autoregressive + surface_ar_enabled: bool = False + surface_ar_d_model: int = 64 + surface_ar_n_layers: int = 2 + surface_ar_n_heads: int = 2 + surface_ar_max_bytes: int = 64 + surface_ar_temperature: float = 0.0 + surface_ar_top_k: int = 1 + surface_ar_fallback: str = "deterministic" + surface_ar_verify_retry: bool = True @property def run_dir(self) -> Path: diff --git a/src/slm_training/models/surface_autoregressor.py b/src/slm_training/models/surface_autoregressor.py new file mode 100644 index 00000000..9ae95de5 --- /dev/null +++ b/src/slm_training/models/surface_autoregressor.py @@ -0,0 +1,531 @@ +"""Small causal byte-autoregressor for VSS3-05 surface realization. + +Generates short surface strings (internal identifiers, decorative text) under +per-slot constraints. Kept separate from the main TwoTower/diffusion paths so +it can be trained independently and fallback to the deterministic baseline is +always available. +""" + +from __future__ import annotations + +import random +import re +import string +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Sequence + +import torch +import torch.nn as nn +import torch.nn.functional as F + + +PRINTABLE_ASCII = string.printable[:-5] # exclude \t\n\r\x0b\x0c + + +@dataclass(frozen=True) +class SurfaceAutoregressorConfig: + """Hyperparameters for the surface autoregressor.""" + + d_model: int = 64 + n_layers: int = 2 + n_heads: int = 2 + max_len: int = 64 + dropout: float = 0.0 + max_bytes: int = 64 + vocab_kind: str = "byte" # byte only for V1 + + def to_dict(self) -> dict[str, Any]: + return { + "d_model": self.d_model, + "n_layers": self.n_layers, + "n_heads": self.n_heads, + "max_len": self.max_len, + "dropout": self.dropout, + "max_bytes": self.max_bytes, + "vocab_kind": self.vocab_kind, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SurfaceAutoregressorConfig: + return cls( + d_model=int(data.get("d_model", 64)), + n_layers=int(data.get("n_layers", 2)), + n_heads=int(data.get("n_heads", 2)), + max_len=int(data.get("max_len", 64)), + dropout=float(data.get("dropout", 0.0)), + max_bytes=int(data.get("max_bytes", 64)), + vocab_kind=str(data.get("vocab_kind", "byte")), + ) + + +class SurfaceByteVocab: + """Tiny byte vocabulary for surface strings. + + Tokens are ````, ````, ````, ````, and ``B:xx`` for + printable ASCII codepoints. This makes the model corpus-independent: any + lowercase identifier or decorative ASCII text can be spelled token-by-token. + """ + + PAD = "" + BOS = "" + EOS = "" + UNK = "" + SPECIAL = (PAD, BOS, EOS, UNK) + + def __init__(self) -> None: + self.tokens = list(self.SPECIAL) + self.tokens.extend(f"B:{ord(ch):02x}" for ch in PRINTABLE_ASCII) + self.token_to_id = {tok: i for i, tok in enumerate(self.tokens)} + self.id_to_token = {i: tok for tok, i in self.token_to_id.items()} + self.pad_id = self.token_to_id[self.PAD] + self.bos_id = self.token_to_id[self.BOS] + self.eos_id = self.token_to_id[self.EOS] + self.unk_id = self.token_to_id[self.UNK] + self.vocab_size = len(self.tokens) + self.byte_to_id = { + ch: self.token_to_id[f"B:{ord(ch):02x}"] for ch in PRINTABLE_ASCII + } + + def encode(self, text: str, *, add_special: bool = True) -> list[int]: + ids = [self.byte_to_id.get(ch, self.unk_id) for ch in text] + if add_special: + return [self.bos_id] + ids + [self.eos_id] + return ids + + def decode(self, ids: Iterable[int], *, skip_special: bool = True) -> str: + chars: list[str] = [] + for i in ids: + tok = self.id_to_token.get(i, self.UNK) + if skip_special and tok in self.SPECIAL: + continue + if tok.startswith("B:"): + chars.append(chr(int(tok[2:], 16))) + else: + chars.append("�") + return "".join(chars) + + def __len__(self) -> int: + return self.vocab_size + + +# Generic identifier grammar: lowercase start, alphanumeric/underscore. +_IDENTIFIER_RE = re.compile(r"^[a-z_][a-z0-9_]*$") + + +class IdentifierConstraint: + """Constrained next-token set for OpenUI-style internal identifiers.""" + + FIRST = set("abcdefghijklmnopqrstuvwxyz_") + REST = set("abcdefghijklmnopqrstuvwxyz0123456789_") + + def __init__( + self, + vocab: SurfaceByteVocab, + *, + max_bytes: int = 64, + reserved: set[str] | frozenset[str] | None = None, + peers: set[str] | frozenset[str] | None = None, + ) -> None: + self.vocab = vocab + self.max_bytes = max_bytes + self.reserved: frozenset[str] = frozenset(reserved or ()) + self.peers: frozenset[str] = frozenset(peers or ()) + self.first_ids = {vocab.byte_to_id[ch] for ch in self.FIRST if ch in vocab.byte_to_id} + self.rest_ids = {vocab.byte_to_id[ch] for ch in self.REST if ch in vocab.byte_to_id} + + def allowed_next(self, prefix: str) -> set[int]: + """Return token ids that may legally follow ``prefix``.""" + if len(prefix.encode("utf-8")) >= self.max_bytes: + return set() + if not prefix: + return self.first_ids.copy() + allowed = self.rest_ids.copy() + candidate = prefix + if ( + _IDENTIFIER_RE.match(candidate) + and candidate not in self.reserved + and candidate not in self.peers + and len(candidate.encode("utf-8")) <= self.max_bytes + ): + allowed.add(self.vocab.eos_id) + return allowed + + def is_complete(self, value: str) -> bool: + return ( + bool(value) + and _IDENTIFIER_RE.match(value) is not None + and value not in self.reserved + and value not in self.peers + and len(value.encode("utf-8")) <= self.max_bytes + ) + + +class DecorativeConstraint: + """Constrained next-token set for decorative ASCII text.""" + + def __init__(self, vocab: SurfaceByteVocab, *, max_bytes: int = 256) -> None: + self.vocab = vocab + self.max_bytes = max_bytes + self.byte_ids = set(vocab.byte_to_id.values()) + + def allowed_next(self, prefix: str) -> set[int]: + if len(prefix.encode("utf-8")) >= self.max_bytes: + return {self.vocab.eos_id} + allowed = self.byte_ids.copy() + allowed.add(self.vocab.eos_id) + return allowed + + def is_complete(self, value: str) -> bool: + return len(value.encode("utf-8")) <= self.max_bytes + + +class CausalSelfAttention(nn.Module): + """Causal scaled dot-product attention.""" + + def __init__(self, d_model: int, n_heads: int, dropout: float = 0.0) -> None: + super().__init__() + if d_model % n_heads != 0: + raise ValueError("d_model must be divisible by n_heads") + self.n_heads = n_heads + self.head_dim = d_model // n_heads + self.q_proj = nn.Linear(d_model, d_model) + self.k_proj = nn.Linear(d_model, d_model) + self.v_proj = nn.Linear(d_model, d_model) + self.out_proj = nn.Linear(d_model, d_model) + self.dropout = dropout + + def forward( + self, + x: torch.Tensor, + key_padding_mask: torch.Tensor | None = None, + ) -> torch.Tensor: + bsz, tlen, _ = x.shape + q = self.q_proj(x).view(bsz, tlen, self.n_heads, self.head_dim).transpose(1, 2) + k = self.k_proj(x).view(bsz, tlen, self.n_heads, self.head_dim).transpose(1, 2) + v = self.v_proj(x).view(bsz, tlen, self.n_heads, self.head_dim).transpose(1, 2) + + # Causal mask: positions j > i are masked. + causal = torch.triu(torch.ones(tlen, tlen, device=x.device), diagonal=1).bool() + attn_mask = torch.zeros(bsz, self.n_heads, tlen, tlen, device=x.device, dtype=q.dtype) + attn_mask = attn_mask.masked_fill(causal[None, None, :, :], float("-inf")) + if key_padding_mask is not None: + pad = key_padding_mask[:, None, None, :] # [B,1,1,S] + attn_mask = attn_mask.masked_fill(pad, float("-inf")) + + out = F.scaled_dot_product_attention( + q, + k, + v, + attn_mask=attn_mask, + dropout_p=self.dropout if self.training else 0.0, + ) + out = out.transpose(1, 2).contiguous().view(bsz, tlen, -1) + return self.out_proj(out) + + +class SurfaceTransformerBlock(nn.Module): + """Causal decoder block with optional cross-attention to context.""" + + def __init__( + self, + d_model: int, + n_heads: int, + mlp_ratio: float = 4.0, + dropout: float = 0.0, + cross_attn: bool = False, + ) -> None: + super().__init__() + self.self_attn = CausalSelfAttention(d_model, n_heads, dropout) + self.norm1 = nn.LayerNorm(d_model) + self.cross_attn = ( + nn.MultiheadAttention(d_model, n_heads, dropout=dropout, batch_first=True) + if cross_attn + else None + ) + self.norm_cross = nn.LayerNorm(d_model) if cross_attn else None + hidden = int(d_model * mlp_ratio) + self.mlp = nn.Sequential( + nn.Linear(d_model, hidden), + nn.GELU(), + nn.Linear(hidden, d_model), + ) + self.norm2 = nn.LayerNorm(d_model) + self.dropout = nn.Dropout(dropout) + + def forward( + self, + x: torch.Tensor, + self_pad_mask: torch.Tensor | None = None, + ctx: torch.Tensor | None = None, + ctx_pad_mask: torch.Tensor | None = None, + ) -> torch.Tensor: + x = x + self.dropout(self.self_attn(self.norm1(x), key_padding_mask=self_pad_mask)) + if self.cross_attn is not None and ctx is not None: + assert self.norm_cross is not None + cross_out, _ = self.cross_attn( + self.norm_cross(x), ctx, ctx, key_padding_mask=ctx_pad_mask + ) + x = x + self.dropout(cross_out) + x = x + self.dropout(self.mlp(self.norm2(x))) + return x + + +class SurfaceContextEncoder(nn.Module): + """Tiny encoder that builds a fixed-size context vector from a prompt.""" + + def __init__( + self, + vocab_size: int, + d_model: int = 64, + n_layers: int = 1, + n_heads: int = 2, + max_len: int = 128, + dropout: float = 0.0, + ) -> None: + super().__init__() + self.tok = nn.Embedding(vocab_size, d_model) + self.pos = nn.Embedding(max_len, d_model) + self.layers = nn.ModuleList( + [SurfaceTransformerBlock(d_model, n_heads, dropout=dropout) for _ in range(n_layers)] + ) + self.norm = nn.LayerNorm(d_model) + self.max_len = max_len + self.d_model = d_model + + def forward(self, input_ids: torch.Tensor, pad_id: int) -> torch.Tensor: + bsz, seq = input_ids.shape + if seq > self.max_len: + input_ids = input_ids[:, : self.max_len] + seq = self.max_len + pos = torch.arange(seq, device=input_ids.device).unsqueeze(0).expand(bsz, -1) + x = self.tok(input_ids) + self.pos(pos) + pad_mask = input_ids.eq(pad_id) + for layer in self.layers: + x = layer(x, self_pad_mask=pad_mask) + # Mean pool over non-pad positions. + x = self.norm(x) + mask = (~pad_mask).unsqueeze(-1).float() + pooled = (x * mask).sum(dim=1) / mask.sum(dim=1).clamp_min(1.0) + return pooled # [B, D] + + +class SurfaceAutoregressor(nn.Module): + """Causal byte-decoder for surface slot realization.""" + + def __init__(self, config: SurfaceAutoregressorConfig) -> None: + super().__init__() + self.config = config + self.vocab = SurfaceByteVocab() + if config.vocab_kind != "byte": + raise ValueError(f"unsupported vocab_kind {config.vocab_kind!r}") + if config.d_model % config.n_heads != 0: + raise ValueError("d_model must be divisible by n_heads") + self.tok = nn.Embedding(self.vocab.vocab_size, config.d_model) + self.pos = nn.Embedding(config.max_len, config.d_model) + self.ctx_encoder = SurfaceContextEncoder( + vocab_size=self.vocab.vocab_size, + d_model=config.d_model, + n_layers=max(1, config.n_layers // 2), + n_heads=config.n_heads, + max_len=config.max_len, + dropout=config.dropout, + ) + self.layers = nn.ModuleList( + [ + SurfaceTransformerBlock( + config.d_model, + config.n_heads, + dropout=config.dropout, + cross_attn=True, + ) + for _ in range(config.n_layers) + ] + ) + self.norm = nn.LayerNorm(config.d_model) + self.lm_head = nn.Linear(config.d_model, self.vocab.vocab_size, bias=False) + self.lm_head.weight = self.tok.weight # tie weights + self.dropout = nn.Dropout(config.dropout) + + def forward(self, input_ids: torch.Tensor, prompt_ids: torch.Tensor) -> torch.Tensor: + """Teacher-forced forward. Returns logits [B, T, V].""" + bsz, tlen = input_ids.shape + ctx = self.ctx_encoder(prompt_ids, pad_id=self.vocab.pad_id) # [B, D] + ctx = ctx.unsqueeze(1) # [B, 1, D] for cross-attn + pos = torch.arange(tlen, device=input_ids.device).unsqueeze(0).expand(bsz, -1) + x = self.tok(input_ids) + self.pos(pos) + pad_mask = input_ids.eq(self.vocab.pad_id) + for layer in self.layers: + x = layer(x, self_pad_mask=pad_mask, ctx=ctx) + x = self.norm(x) + return self.lm_head(x) + + def compute_loss( + self, + input_ids: torch.Tensor, + prompt_ids: torch.Tensor, + targets: torch.Tensor, + ) -> torch.Tensor: + logits = self.forward(input_ids, prompt_ids) + return F.cross_entropy( + logits.view(-1, logits.size(-1)), + targets.view(-1), + ignore_index=self.vocab.pad_id, + ) + + @torch.no_grad() + def generate( + self, + prompt_ids: torch.Tensor, + constraint: IdentifierConstraint | DecorativeConstraint, + *, + max_bytes: int | None = None, + temperature: float = 0.0, + top_k: int = 1, + seed: int | None = None, + ) -> str | None: + """Constrained greedy/top-k generation for one slot. + + Returns the generated string, or ``None`` if the decoder reaches a dead + end before a legal EOS. + """ + if seed is not None: + torch.manual_seed(seed) + random.seed(seed) + device = next(self.parameters()).device + prompt_ids = prompt_ids.to(device) + if prompt_ids.dim() == 1: + prompt_ids = prompt_ids.unsqueeze(0) + ctx = self.ctx_encoder(prompt_ids, pad_id=self.vocab.pad_id).unsqueeze(1) + + max_steps = max_bytes or self.config.max_bytes + generated: list[int] = [self.vocab.bos_id] + prefix = "" + for _ in range(max_steps + 2): + input_ids = torch.tensor([generated], dtype=torch.long, device=device) + pos = torch.arange(len(generated), device=device).unsqueeze(0) + x = self.tok(input_ids) + self.pos(pos) + pad_mask = input_ids.eq(self.vocab.pad_id) + for layer in self.layers: + x = layer(x, self_pad_mask=pad_mask, ctx=ctx) + logits = self.lm_head(self.norm(x))[:, -1, :] # [1, V] + + allowed = constraint.allowed_next(prefix) + if not allowed: + return None + mask = torch.full_like(logits, float("-inf")) + mask[0, list(allowed)] = 0.0 + logits = logits + mask + + if temperature <= 0.0 or top_k == 1: + next_id = int(logits.argmax(dim=-1).item()) + else: + probs = F.softmax(logits / temperature, dim=-1) + topk = torch.topk(probs, min(top_k, probs.size(-1))) + next_id = int(torch.multinomial(topk.values, num_samples=1).item()) + next_id = int(topk.indices[0, next_id].item()) + + if next_id == self.vocab.eos_id: + if constraint.is_complete(prefix): + return prefix + return None + generated.append(next_id) + prefix = self.vocab.decode(generated, skip_special=True) + if len(prefix.encode("utf-8")) > max_steps: + return None + return None + + def save(self, path: Path | str) -> None: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + torch.save( + { + "config": self.config.to_dict(), + "state_dict": self.state_dict(), + }, + path, + ) + + @classmethod + def load(cls, path: Path | str, device: str = "cpu") -> SurfaceAutoregressor: + path = Path(path) + payload = torch.load(path, map_location=device, weights_only=False) + config = SurfaceAutoregressorConfig.from_dict(payload["config"]) + model = cls(config) + model.load_state_dict(payload["state_dict"]) + model.to(device) + return model + + @classmethod + def from_records( + cls, + records: Sequence[Any], + *, + config: SurfaceAutoregressorConfig | None = None, + device: str = "cpu", + ) -> SurfaceAutoregressor: + """Build an untrained model; records are accepted for API symmetry.""" + del records + model = cls(config or SurfaceAutoregressorConfig()) + model.to(device) + return model + + +def train_surface_autoregressor( + model: SurfaceAutoregressor, + examples: list[tuple[str, str]], + *, + steps: int = 100, + lr: float = 3e-3, + device: str = "cpu", + seed: int = 0, +) -> dict[str, float]: + """Tiny fixture trainer: overfit the model on (prompt, target) pairs. + + Each example is a short prompt string and a target surface value. The model + is trained teacher-forced with cross-entropy. Returns final loss. + """ + import random + + torch.manual_seed(seed) + random.seed(seed) + model.to(device) + model.train() + optimizer = torch.optim.Adam(model.parameters(), lr=lr) + vocab = model.vocab + + def make_batch() -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]: + prompt, target = random.choice(examples) + prompt_ids = torch.tensor(vocab.encode(prompt, add_special=True), dtype=torch.long) + target_ids = vocab.encode(target, add_special=False) + input_ids = [vocab.bos_id] + target_ids + targets = target_ids + [vocab.eos_id] + return ( + torch.tensor(input_ids, dtype=torch.long), + prompt_ids, + torch.tensor(targets, dtype=torch.long), + ) + + losses: list[float] = [] + for _ in range(steps): + input_ids, prompt_ids, targets = make_batch() + input_ids = input_ids.unsqueeze(0).to(device) + prompt_ids = prompt_ids.unsqueeze(0).to(device) + targets = targets.unsqueeze(0).to(device) + optimizer.zero_grad() + loss = model.compute_loss(input_ids, prompt_ids, targets) + loss.backward() + optimizer.step() + losses.append(float(loss.item())) + + return {"final_loss": losses[-1], "initial_loss": losses[0]} + + +__all__ = [ + "DecorativeConstraint", + "IdentifierConstraint", + "SurfaceAutoregressor", + "SurfaceAutoregressorConfig", + "SurfaceByteVocab", + "train_surface_autoregressor", +] diff --git a/tests/test_data/test_surface_rows.py b/tests/test_data/test_surface_rows.py new file mode 100644 index 00000000..bef807db --- /dev/null +++ b/tests/test_data/test_surface_rows.py @@ -0,0 +1,143 @@ +"""Regression tests for VSS3-05 surface-realization training rows (SLM-73). + +The deterministic-baseline derivation is torch-free and bridge-free (main's +``realize_surface_and_verify`` needs neither), so it runs here. The neural-model +derivation requires torch and is skipped where torch is unavailable +(``pytest.importorskip("torch")``, this repo's convention); it is CI-run. +""" + +from __future__ import annotations + +import pytest + +from slm_training.data.progspec.schema import ProgramSpec +from slm_training.data.progspec.surface_rows import derive_surface_realization_records +from slm_training.dsl.pack import get_pack +from slm_training.dsl.surface import ( + SurfaceAuthority, + SurfaceSlotKind, + resolve_surface_slot_extractor, +) + +NO_OPAQUE = 'root = Stack([title], "column")\ntitle = Stack([], "row")' + + +def _spec(spec_id: str, *, split: str = "train") -> ProgramSpec: + return ProgramSpec.from_openui( + id=spec_id, + openui=NO_OPAQUE, + facts={}, + program_family_id="surf_family", + lineage_id="surf_lineage", + split_group_id="surf_group", + split=split, + ) + + +# --------------------------------------------------------------------------- +# Deterministic baseline (torch-free, bridge-free) +# --------------------------------------------------------------------------- + + +def test_derive_surface_realization_records_emits_surface_only_slots() -> None: + spec = _spec("surface_det") + records = derive_surface_realization_records(spec) + assert len(records) == 1 + record = records[0] + assert record.meta["task"] == "surface_realization" + assert record.source == "surface_realization" + assert record.split == spec.split + assert record.meta["split_group_id"] == spec.split_group_id + assert record.meta["parent_id"] == spec.id + assert record.meta["determinacy"] == "deterministic" + assert record.meta["tier"] == "Silver" + provenance = record.meta["provenance"] + assert provenance["surface_slot_id"] == "openui:binder:title" + assert provenance["surface_assignment"]["value"] == "v0" + assert "deterministic" in provenance["surface_assignment"]["provenance"] + + +def test_derive_surface_realization_records_honors_include_authorities() -> None: + spec = _spec("surface_auth") + # An empty frozenset is falsy, so the function falls back to defaults; use a + # non-matching authority to prove the filter is honored. + assert ( + derive_surface_realization_records( + spec, + include_authorities=frozenset({"semantic"}), + ) + == () + ) + assert ( + len( + derive_surface_realization_records( + spec, + include_authorities=frozenset({SurfaceAuthority.SURFACE_ONLY.value}), + ) + ) + == 1 + ) + assert ( + derive_surface_realization_records( + spec, + include_authorities=frozenset({SurfaceAuthority.OPAQUE_USER_VALUE.value}), + ) + == () + ) + + +def test_derive_surface_realization_records_inherits_split_and_group() -> None: + spec = _spec("surface_split", split="held_out") + records = derive_surface_realization_records(spec) + assert len(records) == 1 + assert records[0].split == "held_out" + assert records[0].meta["split_group_id"] == "surf_group" + + +# --------------------------------------------------------------------------- +# Neural realizer derivation (torch-only; CI-run) +# --------------------------------------------------------------------------- + + +def _trained_neural_realizer_for_spec(spec: ProgramSpec): + """Train a tiny fixture model to emit a custom name for the spec's binder.""" + from slm_training.dsl.neural_surface_realizer import ( + NeuralSurfaceRealizer, + NeuralSurfaceRealizerConfig, + ) + from slm_training.models.surface_autoregressor import ( + SurfaceAutoregressor, + SurfaceAutoregressorConfig, + train_surface_autoregressor, + ) + + pack = get_pack("openui") + extractor = resolve_surface_slot_extractor(pack) + slots = extractor(spec.canonical_openui) + slot = next( + s + for s in slots + if s.authority is SurfaceAuthority.SURFACE_ONLY + and s.kind is SurfaceSlotKind.INTERNAL_IDENTIFIER + ) + realizer = NeuralSurfaceRealizer(NeuralSurfaceRealizerConfig()) + prompt = realizer._build_prompt(slot, set()) + model = SurfaceAutoregressor( + SurfaceAutoregressorConfig(d_model=32, n_layers=1, n_heads=2, max_len=64) + ) + train_surface_autoregressor(model, [(prompt, "mytitle")], steps=300, lr=5e-3, seed=0) + model.eval() + return NeuralSurfaceRealizer( + NeuralSurfaceRealizerConfig(model=model, fallback_to_deterministic=True) + ) + + +def test_derive_surface_realization_records_can_use_neural_realizer() -> None: + pytest.importorskip("torch") + spec = _spec("surface_ar") + realizer = _trained_neural_realizer_for_spec(spec) + records = derive_surface_realization_records(spec, realizer=realizer) + assert len(records) == 1 + provenance = records[0].meta["provenance"] + assert provenance["surface_assignment"]["value"] == "mytitle" + assert provenance["surface_assignment"]["provenance"] == "autoregressive" diff --git a/tests/test_dsl/test_neural_surface_realizer.py b/tests/test_dsl/test_neural_surface_realizer.py new file mode 100644 index 00000000..d1f95fb3 --- /dev/null +++ b/tests/test_dsl/test_neural_surface_realizer.py @@ -0,0 +1,214 @@ +"""Regression tests for the VSS3-05 neural surface realizer (SLM-73). + +The no-model deterministic-fallback path and the authority/kind guards are +torch-free and run here. The model-backed paths (trained generation, constrained +dead-end fallback) require torch and are skipped where torch is unavailable +(``pytest.importorskip("torch")``, this repo's convention); they are CI-run. +""" + +from __future__ import annotations + +import pytest + +from slm_training.dsl.neural_surface_realizer import ( + NeuralSurfaceRealizer, + NeuralSurfaceRealizerConfig, +) +from slm_training.dsl.opaque_regions import OpaqueRegionBinding +from slm_training.dsl.pack import get_pack +from slm_training.dsl.surface import ( + SurfaceAuthority, + SurfaceConstraint, + SurfaceRealizationRequest, + SurfaceSlot, + SurfaceSlotKind, + realize_surface_and_verify, +) + +HERO = 'root = Stack([title], "column")\ntitle = TextContent(":hero.title")' + + +def _openui_binding(region_id: str, value: str) -> OpaqueRegionBinding: + return OpaqueRegionBinding(region_id=region_id, scalar_value=value) + + +def _identifier_request( + slots: tuple[SurfaceSlot, ...], +) -> SurfaceRealizationRequest: + return SurfaceRealizationRequest( + pack_id="openui", + constraint_version="v1", + semantic_ir_fingerprint="fp", + slots=slots, + context={}, + ) + + +def _identifier_slot( + slot_id: str = "s1", + symbol: str = "title", + authority: SurfaceAuthority = SurfaceAuthority.SURFACE_ONLY, + max_bytes: int = 64, + reserved: tuple[str, ...] = (), +) -> SurfaceSlot: + return SurfaceSlot( + slot_id=slot_id, + kind=SurfaceSlotKind.INTERNAL_IDENTIFIER, + authority=authority, + ast_path=(), + semantic_symbol_id=symbol, + opaque_region_id=None, + constraints=SurfaceConstraint(max_bytes=max_bytes, reserved=reserved), + current_value_digest=None, + ) + + +# --------------------------------------------------------------------------- +# Torch-free logic: no-model fallback and authority/kind guards +# --------------------------------------------------------------------------- + + +def test_neural_realizer_without_model_falls_back_to_deterministic() -> None: + """With no model, every supported slot falls back to the deterministic baseline.""" + pack = get_pack("openui") + realizer = NeuralSurfaceRealizer(NeuralSurfaceRealizerConfig(model=None)) + result = realize_surface_and_verify( + HERO, + pack=pack, + realizer=realizer, + opaque_bindings={ + "openui:content::hero.title": _openui_binding( + "openui:content::hero.title", ":user.title" + ) + }, + semantic_ir_fingerprint="fp", + prior_status="solved", + ) + assert result.status == "solved" + assert result.source is not None + assert "v0 = TextContent" in result.source + assignment = next( + a for a in result.assignments if a.slot_id == "openui:binder:title" + ) + assert assignment.value == "v0" + assert assignment.provenance.startswith( + "autoregressive_fallback:no_model:deterministic:canonical_name" + ) + + +def test_neural_realizer_disabled_fallback_raises() -> None: + """When fallback is disabled, a realization failure is a hard error.""" + slot = _identifier_slot() + request = _identifier_request((slot,)) + realizer = NeuralSurfaceRealizer( + NeuralSurfaceRealizerConfig(model=None, fallback_to_deterministic=False) + ) + with pytest.raises(ValueError, match="fallback is disabled"): + realizer.realize(request) + + +def test_neural_realizer_rejects_semantic_authority() -> None: + """Only SURFACE_ONLY slots may be handed to the AR model.""" + slot = _identifier_slot(authority=SurfaceAuthority.SEMANTIC) + request = _identifier_request((slot,)) + realizer = NeuralSurfaceRealizer(NeuralSurfaceRealizerConfig(model=None)) + with pytest.raises(ValueError, match="only SURFACE_ONLY"): + realizer.realize(request) + + +def test_neural_realizer_rejects_unsupported_kind() -> None: + """STRUCTURED_STRING and other non-surface kinds are rejected before generation.""" + slot = SurfaceSlot( + slot_id="s1", + kind=SurfaceSlotKind.STRUCTURED_STRING, + authority=SurfaceAuthority.SURFACE_ONLY, + ast_path=(), + semantic_symbol_id=None, + opaque_region_id=None, + constraints=SurfaceConstraint(), + current_value_digest=None, + ) + request = _identifier_request((slot,)) + realizer = NeuralSurfaceRealizer(NeuralSurfaceRealizerConfig(model=None)) + with pytest.raises(ValueError, match="not supported by the autoregressive"): + realizer.realize(request) + + +# --------------------------------------------------------------------------- +# Model-backed paths (torch-only; CI-run) +# --------------------------------------------------------------------------- + + +def _tiny_ar_config(): + from slm_training.models.surface_autoregressor import SurfaceAutoregressorConfig + + return SurfaceAutoregressorConfig(d_model=32, n_layers=1, n_heads=2, max_len=64) + + +def _trained_neural_realizer( + prompt_target_pairs: list[tuple[str, str]], +) -> NeuralSurfaceRealizer: + """Train a tiny fixture model and wrap it in the neural realizer.""" + from slm_training.models.surface_autoregressor import ( + SurfaceAutoregressor, + train_surface_autoregressor, + ) + + model = SurfaceAutoregressor(_tiny_ar_config()) + train_surface_autoregressor( + model, prompt_target_pairs, steps=500, lr=5e-3, seed=0 + ) + model.eval() + return NeuralSurfaceRealizer( + NeuralSurfaceRealizerConfig(model=model, fallback_to_deterministic=True) + ) + + +def test_neural_realizer_trained_identifier_is_verified() -> None: + """A trained AR model can realize a binder name through realize_surface_and_verify.""" + pytest.importorskip("torch") + pack = get_pack("openui") + prompt = ( + "kind=internal_identifier authority=surface_only " + "slot_id=openui:binder:title symbol=title max=64" + ) + realizer = _trained_neural_realizer([(prompt, "title")]) + result = realize_surface_and_verify( + HERO, + pack=pack, + realizer=realizer, + opaque_bindings={ + "openui:content::hero.title": _openui_binding( + "openui:content::hero.title", ":user.title" + ) + }, + semantic_ir_fingerprint="fp", + prior_status="solved", + ) + assert result.status == "solved" + assert result.source is not None + # The pack canonicalizer renames binders to canonical names, so the final + # source uses v0; the assignment itself records the model-chosen value. + assert "v0 = TextContent" in result.source + assignment = next( + a for a in result.assignments if a.slot_id == "openui:binder:title" + ) + assert assignment.value == "title" + assert assignment.provenance == "autoregressive" + + +def test_neural_realizer_dead_end_falls_back_to_deterministic() -> None: + """A constrained dead end triggers per-slot deterministic fallback.""" + pytest.importorskip("torch") + from slm_training.models.surface_autoregressor import SurfaceAutoregressor + + slot = _identifier_slot(max_bytes=2) + request = _identifier_request((slot,)) + model = SurfaceAutoregressor(_tiny_ar_config()) + realizer = NeuralSurfaceRealizer( + NeuralSurfaceRealizerConfig(model=model, fallback_to_deterministic=True) + ) + assignments = realizer.realize(request) + assert len(assignments) == 1 + assert assignments[0].value == "v0" + assert "dead_end" in assignments[0].provenance diff --git a/tests/test_models/test_surface_autoregressor.py b/tests/test_models/test_surface_autoregressor.py new file mode 100644 index 00000000..440c7f4d --- /dev/null +++ b/tests/test_models/test_surface_autoregressor.py @@ -0,0 +1,120 @@ +"""Regression tests for VSS3-05 surface autoregressor (SLM-73). + +Torch-only. Skipped where torch is unavailable, per this repo's convention +(``pytest.importorskip("torch")``); the assertions are CI-run. +""" + +from __future__ import annotations + +import pytest + +torch = pytest.importorskip("torch") + +from slm_training.models.surface_autoregressor import ( + DecorativeConstraint, + IdentifierConstraint, + SurfaceAutoregressor, + SurfaceAutoregressorConfig, + SurfaceByteVocab, + train_surface_autoregressor, +) + + +@pytest.fixture +def vocab() -> SurfaceByteVocab: + return SurfaceByteVocab() + + +@pytest.fixture +def tiny_config() -> SurfaceAutoregressorConfig: + return SurfaceAutoregressorConfig(d_model=32, n_layers=1, n_heads=2, max_len=32) + + +def test_byte_vocab_encode_decode_roundtrip(vocab: SurfaceByteVocab) -> None: + text = "hello_world" + ids = vocab.encode(text) + assert ids[0] == vocab.bos_id + assert ids[-1] == vocab.eos_id + assert vocab.decode(ids) == text + + +def test_byte_vocab_decode_ignores_specials(vocab: SurfaceByteVocab) -> None: + text = "abc" + ids = vocab.encode(text) + assert vocab.decode([vocab.pad_id, *ids, vocab.unk_id]) == text + + +def test_identifier_constraint_first_position(vocab: SurfaceByteVocab) -> None: + constraint = IdentifierConstraint(vocab, max_bytes=16) + allowed = constraint.allowed_next("") + assert vocab.eos_id not in allowed + # Digits are not allowed at position 0. + assert vocab.byte_to_id["0"] not in allowed + assert vocab.byte_to_id["a"] in allowed + assert vocab.byte_to_id["_"] in allowed + + +def test_identifier_constraint_later_positions_allow_eos(vocab: SurfaceByteVocab) -> None: + constraint = IdentifierConstraint(vocab, max_bytes=16) + allowed = constraint.allowed_next("h") + assert vocab.byte_to_id["0"] in allowed + assert vocab.eos_id in allowed + + +def test_identifier_constraint_rejects_reserved(vocab: SurfaceByteVocab) -> None: + constraint = IdentifierConstraint(vocab, max_bytes=16, reserved={"hello"}) + assert not constraint.is_complete("hello") + allowed = constraint.allowed_next("hello") + assert vocab.eos_id not in allowed + + +def test_identifier_constraint_rejects_collision(vocab: SurfaceByteVocab) -> None: + constraint = IdentifierConstraint(vocab, max_bytes=16, peers={"foo"}) + assert not constraint.is_complete("foo") + + +def test_decorative_constraint_allows_printable_and_eos(vocab: SurfaceByteVocab) -> None: + constraint = DecorativeConstraint(vocab, max_bytes=8) + allowed = constraint.allowed_next("hi") + assert vocab.byte_to_id["!"] in allowed + assert vocab.eos_id in allowed + + +def test_untrained_model_generate_returns_none_or_fallback(tiny_config: SurfaceAutoregressorConfig, vocab: SurfaceByteVocab) -> None: + model = SurfaceAutoregressor(tiny_config) + constraint = IdentifierConstraint(vocab, max_bytes=16) + prompt_ids = torch.tensor(vocab.encode("kind=internal_identifier symbol=title"), dtype=torch.long) + value = model.generate(prompt_ids, constraint, max_bytes=16) + # Untrained model is extremely unlikely to produce a legal identifier. + assert value is None or constraint.is_complete(value) + + +def test_tiny_fixture_overfit(tiny_config: SurfaceAutoregressorConfig, vocab: SurfaceByteVocab) -> None: + model = SurfaceAutoregressor(tiny_config) + # Use a single example so the tiny fixture model reliably overfits. + examples = [("kind=internal_identifier symbol=title", "title")] + metrics = train_surface_autoregressor(model, examples, steps=200, lr=5e-3, seed=0) + assert metrics["final_loss"] < metrics["initial_loss"] + + # Greedy generation should recover the trained target. + model.eval() + constraint = IdentifierConstraint(vocab, max_bytes=16) + prompt_ids = torch.tensor(vocab.encode("kind=internal_identifier symbol=title"), dtype=torch.long) + value = model.generate(prompt_ids, constraint, max_bytes=16, temperature=0.0, top_k=1) + assert value == "title" + + +def test_model_save_load_roundtrip(tmp_path: pytest.TempPathFactory, tiny_config: SurfaceAutoregressorConfig) -> None: + model = SurfaceAutoregressor(tiny_config) + path = tmp_path / "surface_ar.pt" # type: ignore[attr-defined] + model.save(path) + loaded = SurfaceAutoregressor.load(path, device="cpu") + assert loaded.config.d_model == model.config.d_model + assert sum(p.numel() for p in loaded.parameters()) == sum( + p.numel() for p in model.parameters() + ) + + +def test_from_records_ignores_records(tiny_config: SurfaceAutoregressorConfig) -> None: + model = SurfaceAutoregressor.from_records([], config=tiny_config) + assert model.config == tiny_config