PHON-106 — CSP Contrastive Scorers Implementation Plan¶

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Pivot CSP contrastive enforcement from post-hoc cross-slot scoring to linked-slot enumeration. Define MinpairConstraint / MaxoppConstraint / MultoppConstraint constraint types. Emit data/runtime/pairs.parquet (built from existing _compute_minimal_pairs data plus continuous feature_distance / sonorant_diff from learned posterior vectors). Wire minpair + maxopp into the vectorized solver via linked-slot mode on single-sentence shapes; multopp is defined but defers to PHON-108.

Architecture: Two tiers. Pipeline tier: extend _compute_minimal_pairs with feature distances, emit pairs.parquet, load in WordStore, switch export-to-d1.py to use WordStore as single source of truth. CSP tier: replace the single ContrastiveConstraint with three types; in solve_shape, when a Minpair/Maxopp constraint is present, switch to linked-slot mode — pair_rows × non-linked-slot Cartesian instead of independent slot Cartesian. Multopp errors out in single-sentence shapes.

Tech Stack: Python 3.12, Polars 1.0+, pytest, NumPy (for cosine distance).

Spec: docs/superpowers/specs/2026-05-09-phon-106-csp-contrastive-scorers-design.md

File map¶

File	Action
`packages/data/src/phonolex_data/pipeline/derived.py`	Modify — `_compute_minimal_pairs` returns 8-tuple; add `_phoneme_pair_distances` helper
`packages/data/src/phonolex_data/pipeline/schema.py`	Modify — `Derived.minimal_pairs` tuple shape extends to 8 elements
`packages/data/src/phonolex_data/runtime/schema.py`	Modify — add `pairs_schema()`
`packages/data/src/phonolex_data/runtime/emit_parquet.py`	Modify — emit `data/runtime/pairs.parquet`
`packages/data/src/phonolex_data/runtime/store.py`	Modify — `WordStore.from_parquet()` also loads `pairs_df`
`packages/web/workers/scripts/export-to-d1.py`	Modify — read pairs via WordStore (single source of truth)
`.gitattributes`	Modify — LFS-track `data/runtime/pairs.parquet`
`packages/generation/research/2026-05-07-sentence-generation-paradigms/constraint_surface.py`	Modify — replace `ContrastiveConstraint` with `MinpairConstraint`/`MaxoppConstraint`/`MultoppConstraint`; rewrite `cross_slot_axes` accordingly
`packages/generation/research/2026-05-07-sentence-generation-paradigms/skeleton_csp.py`	Modify — `_should_use_vectorized` narrows; add linked-slot mode to `_enumerate_vectorized`; validation in `solve_shape`
`packages/generation/research/2026-05-07-sentence-generation-paradigms/paradigm_3_csp.py`	Modify — update ContrastiveConstraint imports
`packages/generation/research/2026-05-07-sentence-generation-paradigms/paragraph_csp.py`	Modify — same import update
`packages/generation/research/2026-05-07-sentence-generation-paradigms/test_vectorized_enumeration.py`	Modify — update imports; existing minpair test stays
`packages/generation/research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py`	Create — new test file
`data/runtime/pairs.parquet`	Create (regenerated artifact)

All paths in this plan are relative to repo root /Users/jneumann/Repos/PhonoLex/. The spike directory is referenced as <spike>/ for brevity: <spike>/ = packages/generation/research/2026-05-07-sentence-generation-paradigms/.

Test commands throughout:

# Spike tests (CSP layer)
cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/ -v

# Data tests (pipeline layer)
cd packages/data && uv run python -m pytest tests/ -v

Task 1: Add `_phoneme_pair_distances` helper in `derived.py`¶

Files: - Modify: packages/data/src/phonolex_data/pipeline/derived.py - Modify: packages/data/tests/test_derived.py (or create if absent)

Pure function: given the learned vectors dict from _compute_phoneme_data, return a dict[(p1, p2), (feature_distance, sonorant_diff)] lookup for fast per-pair access in the upcoming _compute_minimal_pairs extension.

[ ] Step 1.1: Write failing test

Find or create packages/data/tests/test_derived.py. Append:

def test_phoneme_pair_distances_basic():
    import numpy as np
    from phonolex_data.pipeline.derived import _phoneme_pair_distances

    # Synthetic vectors: 2 phonemes, 3 features, sonorant at index 0
    vectors = {
        "p": np.array([0.0, 0.5, 0.5]),  # non-sonorant
        "m": np.array([1.0, 0.5, 0.5]),  # sonorant
    }
    feature_names = ["sonorant", "voice", "labial"]

    distances = _phoneme_pair_distances(vectors, feature_names)

    # Symmetric: both (p, m) and (m, p) keys present
    assert ("p", "m") in distances
    assert ("m", "p") in distances

    feat_dist, son_diff = distances[("p", "m")]
    # L2 distance: sqrt((1-0)^2 + 0 + 0) = 1.0
    assert abs(feat_dist - 1.0) < 1e-6
    # Sonorant diff: |1.0 - 0.0| = 1.0
    assert abs(son_diff - 1.0) < 1e-6

    # (p, p) not in dict — only distinct pairs
    assert ("p", "p") not in distances

[ ] Step 1.2: Run test, verify it fails

cd packages/data && uv run python -m pytest tests/test_derived.py::test_phoneme_pair_distances_basic -v

Expected: ImportError or AttributeError: module 'phonolex_data.pipeline.derived' has no attribute '_phoneme_pair_distances'.

[ ] Step 1.3: Implement _phoneme_pair_distances

Append to packages/data/src/phonolex_data/pipeline/derived.py (just below _compute_phoneme_data):

def _phoneme_pair_distances(
    vectors: dict[str, np.ndarray],
    feature_names: list[str],
) -> dict[tuple[str, str], tuple[float, float]]:
    """Build per-(phoneme1, phoneme2) lookup of (feature_distance, sonorant_diff).

    feature_distance is L2 distance between the two posterior vectors.
    sonorant_diff is |vec1[sonorant] - vec2[sonorant]|, in [0, 1].

    Both directions emitted (p1, p2) and (p2, p1) — symmetric, but caller
    code is simpler with two-direction dict than with sorted-key lookup.
    """
    try:
        sonorant_idx = feature_names.index("sonorant")
    except ValueError as e:
        raise ValueError(
            "feature_names must include 'sonorant' for major-class-diff scoring"
        ) from e

    out: dict[tuple[str, str], tuple[float, float]] = {}
    phonemes = list(vectors.keys())
    for i, p1 in enumerate(phonemes):
        v1 = vectors[p1]
        for p2 in phonemes[i + 1:]:
            v2 = vectors[p2]
            feat_dist = float(np.linalg.norm(v1 - v2))
            son_diff = float(abs(v1[sonorant_idx] - v2[sonorant_idx]))
            out[(p1, p2)] = (feat_dist, son_diff)
            out[(p2, p1)] = (feat_dist, son_diff)
    return out

[ ] Step 1.4: Run test, verify it passes

cd packages/data && uv run python -m pytest tests/test_derived.py::test_phoneme_pair_distances_basic -v

Expected: 1 passed.

[ ] Step 1.5: Commit

git add packages/data/src/phonolex_data/pipeline/derived.py \
        packages/data/tests/test_derived.py
git commit -m "$(cat <<'EOF'
PHON-106: add _phoneme_pair_distances helper

Pure function: given vectors dict + feature_names list, return a per-
(phoneme1, phoneme2) lookup of (feature_distance, sonorant_diff) for
use by _compute_minimal_pairs. Both directions emitted for caller
simplicity.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 2: Extend `_compute_minimal_pairs` to emit 8-tuple¶

Files: - Modify: packages/data/src/phonolex_data/pipeline/derived.py - Modify: packages/data/src/phonolex_data/pipeline/schema.py - Modify: packages/data/tests/test_derived.py

The existing function returns 6-element tuples (w1, w2, ph1, ph2, pos, pos_type). Extend to 8 elements: (w1, w2, ph1, ph2, pos, pos_type, feature_distance, sonorant_diff). Re-key the bucketing logic to also emit the distance fields per pair.

[ ] Step 2.1: Write failing test

Append to packages/data/tests/test_derived.py:

def test_compute_minimal_pairs_includes_distance_columns():
    """Each minimal-pair row carries feature_distance and sonorant_diff."""
    import numpy as np
    from phonolex_data.pipeline.derived import _compute_minimal_pairs
    from phonolex_data.pipeline.schema import WordRecord

    # Two minimal-pair words: cat /k æ t/ and bat /b æ t/ — differ at position 0
    words = {
        "cat": WordRecord(
            word="cat", phonemes=["k", "æ", "t"],
            has_phonology=True, syllables=[],
        ),
        "bat": WordRecord(
            word="bat", phonemes=["b", "æ", "t"],
            has_phonology=True, syllables=[],
        ),
    }
    # Synthetic vectors: k and b differ only on voice
    vectors = {
        "k": np.array([0.0, 0.0, 0.0]),
        "b": np.array([0.0, 1.0, 0.0]),
        "æ": np.array([1.0, 1.0, 0.0]),
        "t": np.array([0.0, 0.0, 0.0]),
    }
    feature_names = ["sonorant", "voice", "labial"]

    pairs = _compute_minimal_pairs(words, vectors=vectors, feature_names=feature_names)

    # One pair (cat, bat) at position 0
    assert len(pairs) == 1
    pair = pairs[0]
    assert len(pair) == 8, f"expected 8-tuple, got {len(pair)}: {pair}"
    w1, w2, ph1, ph2, pos, pos_type, feat_dist, son_diff = pair
    assert {w1, w2} == {"cat", "bat"}
    assert {ph1, ph2} == {"k", "b"}
    assert pos == 0
    assert pos_type == "initial"
    # k and b differ on voice only → L2 distance = 1.0
    assert abs(feat_dist - 1.0) < 1e-6
    # Both non-sonorant → sonorant_diff = 0.0
    assert abs(son_diff - 0.0) < 1e-6

[ ] Step 2.2: Run test, verify it fails

cd packages/data && uv run python -m pytest tests/test_derived.py::test_compute_minimal_pairs_includes_distance_columns -v

Expected: TypeError on tuple unpacking (still 6-tuple), or signature mismatch on vectors kwarg.

[ ] Step 2.3: Update _compute_minimal_pairs signature + body

Find the function in packages/data/src/phonolex_data/pipeline/derived.py (around line 179). The existing signature:

def _compute_minimal_pairs(
    words: dict[str, WordRecord],
) -> list[tuple[str, str, str, str, int, str]]:

Change to:

def _compute_minimal_pairs(
    words: dict[str, WordRecord],
    *,
    vectors: dict[str, np.ndarray],
    feature_names: list[str],
) -> list[tuple[str, str, str, str, int, str, float, float]]:
    """Precompute minimal pairs with feature_distance and sonorant_diff
    per pair, derived from the learned posterior vectors.

    Returns 8-tuples: (word1, word2, phoneme1, phoneme2, position,
    position_type, feature_distance, sonorant_diff).
    """

Then inside the body, build the distance lookup once at function entry:

    pair_distances = _phoneme_pair_distances(vectors, feature_names)

In the section that emits pairs (current line ~228 area), look up (ph1, ph2) in pair_distances. If the pair is missing (e.g., one phoneme isn't in vectors), default to (0.0, 0.0) and emit anyway:

    # Inside the existing append logic, replace:
    #   minimal_pairs.append((w1, w2, ph1, ph2, pos, pos_type))
    # with:
    feat_dist, son_diff = pair_distances.get((ph1, ph2), (0.0, 0.0))
    minimal_pairs.append((w1, w2, ph1, ph2, pos, pos_type, feat_dist, son_diff))

[ ] Step 2.4: Update the caller in derived.py

Find where _compute_minimal_pairs(words) is called (around line 265 in the compute_derived function). It currently calls without the vectors/feature_names kwargs. Update the call to pass them — they're already available via learned (the input):

    # Existing context near line 265:
    phonemes_data, phoneme_norms, phoneme_dots = _compute_phoneme_data(learned)
    # ... other derived computations ...

    # Replace the existing line:
    #   minimal_pairs = _compute_minimal_pairs(words)
    # with:
    minimal_pairs = _compute_minimal_pairs(
        words,
        vectors=learned["vectors"],
        feature_names=learned["feature_names"],
    )

[ ] Step 2.5: Update Derived schema

Find packages/data/src/phonolex_data/pipeline/schema.py. Locate the Derived dataclass field minimal_pairs. The current field:

    minimal_pairs: list[tuple] = field(default_factory=list)

Update the type annotation (though tuple is already broad, make the intent explicit):

    # 8-tuple per pair: (word1, word2, phoneme1, phoneme2, position,
    #                    position_type, feature_distance, sonorant_diff)
    minimal_pairs: list[tuple[str, str, str, str, int, str, float, float]] = field(
        default_factory=list
    )

[ ] Step 2.6: Run test, verify it passes

cd packages/data && uv run python -m pytest tests/test_derived.py -v

Expected: all derived.py tests pass (existing + 2 new from Tasks 1 and 2).

[ ] Step 2.7: Commit

git add packages/data/src/phonolex_data/pipeline/derived.py \
        packages/data/src/phonolex_data/pipeline/schema.py \
        packages/data/tests/test_derived.py
git commit -m "$(cat <<'EOF'
PHON-106: _compute_minimal_pairs emits 8-tuple with feature distance

Each minimal-pair row now carries feature_distance and sonorant_diff
from the learned posterior vectors, enabling continuous maxopp scoring
in the CSP. Falls back to (0.0, 0.0) for any phoneme not present in
the vectors dict (defensive — shouldn't fire given the CMU inventory
filter in _compute_phoneme_data).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 3: Add `pairs_schema()` and emit `pairs.parquet`¶

Files: - Modify: packages/data/src/phonolex_data/runtime/schema.py - Modify: packages/data/src/phonolex_data/runtime/emit_parquet.py - Modify: packages/data/tests/ (test_emit_parquet.py or test_schema.py — wherever schemas are tested)

[ ] Step 3.1: Add pairs_schema() to runtime/schema.py

Append to packages/data/src/phonolex_data/runtime/schema.py:

def pairs_schema() -> pl.Schema:
    """Polars schema for data/runtime/pairs.parquet — minimal pair edges
    with continuous feature_distance and sonorant_diff for maxopp scoring."""
    return pl.Schema({
        "word1":            pl.String,
        "word2":            pl.String,
        "phoneme1":         pl.String,
        "phoneme2":         pl.String,
        "position":         pl.UInt8,
        "position_type":    pl.String,
        "feature_distance": pl.Float32,
        "sonorant_diff":    pl.Float32,
    })

If there are existing tests for words_schema() etc., add a parallel test:

def test_pairs_schema_columns():
    from phonolex_data.runtime.schema import pairs_schema

    schema = pairs_schema()
    assert set(schema.names()) == {
        "word1", "word2", "phoneme1", "phoneme2",
        "position", "position_type",
        "feature_distance", "sonorant_diff",
    }

[ ] Step 3.2: Update emit_parquet.py to emit pairs.parquet

Find packages/data/src/phonolex_data/runtime/emit_parquet.py. It contains an emit_parquet function (or similar) that writes words.parquet, edges.parquet, selectional.parquet from Derived. Add a pairs.parquet emission alongside.

Locate where selectional.parquet is written. After it, add:

    # pairs.parquet — minimal-pair edges with continuous feature distances (PHON-106)
    pairs_rows = derived.minimal_pairs
    pairs_df = pl.DataFrame(
        {
            "word1":            [r[0] for r in pairs_rows],
            "word2":            [r[1] for r in pairs_rows],
            "phoneme1":         [r[2] for r in pairs_rows],
            "phoneme2":         [r[3] for r in pairs_rows],
            "position":         [r[4] for r in pairs_rows],
            "position_type":    [r[5] for r in pairs_rows],
            "feature_distance": [r[6] for r in pairs_rows],
            "sonorant_diff":    [r[7] for r in pairs_rows],
        },
        schema=pairs_schema(),
    )
    pairs_path = output_dir / "pairs.parquet"
    pairs_df.write_parquet(pairs_path)
    print(f"  wrote {pairs_path} ({pairs_df.height:,} rows)")

Add from phonolex_data.runtime.schema import pairs_schema (or extend the existing import line) at the top of the file.

[ ] Step 3.3: Smoke-test the emit

cd packages/data && uv run python scripts/build_runtime_parquet.py

Expected: command exits 0, prints wrote .../pairs.parquet with a row count > 0.

Then verify the schema:

uv run python -c "
import polars as pl
df = pl.read_parquet('data/runtime/pairs.parquet')
print('schema:', df.schema)
print('shape:', df.shape)
print(df.head(5))
"

Expected: 8 columns matching pairs_schema(), ~300K–500K rows, sample rows show real (word1, word2) pairs.

[ ] Step 3.4: LFS-track the new artifact

Find .gitattributes at the repo root. The existing entry for data/runtime/*.parquet should already cover pairs.parquet:

grep "data/runtime" .gitattributes

If the existing entry uses a wildcard like data/runtime/*.parquet filter=lfs ..., no change needed. If it lists files explicitly, add a line:

data/runtime/pairs.parquet filter=lfs diff=lfs merge=lfs -text

[ ] Step 3.5: Commit (separate from the parquet artifact itself)

git add packages/data/src/phonolex_data/runtime/schema.py \
        packages/data/src/phonolex_data/runtime/emit_parquet.py \
        packages/data/tests/test_schema.py \
        .gitattributes
git commit -m "$(cat <<'EOF'
PHON-106: add pairs_schema() and emit pairs.parquet

Schema parallels words/edges/selectional. Pipeline now writes
data/runtime/pairs.parquet alongside the other runtime artifacts.
.gitattributes covers it via the existing data/runtime/*.parquet
LFS rule (or adds an explicit rule if the wildcard isn't already
present).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

[ ] Step 3.6: Commit the regenerated parquet artifact

git add data/runtime/pairs.parquet
git commit -m "$(cat <<'EOF'
PHON-106: regenerate data/runtime/pairs.parquet (LFS)

Output of the updated pipeline: ~Nk rows × 8 columns, including
continuous feature_distance and sonorant_diff per pair.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

(Replace ~Nk in the commit message with the actual row count from Step 3.3.)

Task 4: Load `pairs_df` in `WordStore.from_parquet`¶

Files: - Modify: packages/data/src/phonolex_data/runtime/store.py - Modify: packages/data/tests/test_store.py (or wherever store is tested)

[ ] Step 4.1: Write failing test

Find or create packages/data/tests/test_store.py. Append:

def test_word_store_loads_pairs_df(tmp_path):
    """WordStore.from_parquet loads pairs.parquet as store.pairs_df."""
    from pathlib import Path
    from phonolex_data.runtime.store import WordStore

    repo_root = Path(__file__).resolve().parents[3]
    words_path = repo_root / "data" / "runtime" / "words.parquet"
    if not words_path.exists():
        pytest.skip("data/runtime/words.parquet not present (LFS not pulled)")

    store = WordStore.from_parquet(words_path)
    assert hasattr(store, "pairs_df"), "WordStore should expose pairs_df"
    assert store.pairs_df is not None
    assert store.pairs_df.height > 0
    assert "feature_distance" in store.pairs_df.columns

[ ] Step 4.2: Run test, verify it fails

cd packages/data && uv run python -m pytest tests/test_store.py::test_word_store_loads_pairs_df -v

Expected: AttributeError — store has no pairs_df.

[ ] Step 4.3: Update WordStore

Find packages/data/src/phonolex_data/runtime/store.py. Locate the WordStore dataclass and from_parquet classmethod. Add a pairs_df field and load it:

import polars as pl


@dataclass
class WordStore:
    df: pl.DataFrame
    edges_df: pl.DataFrame | None = None
    selectional_df: pl.DataFrame | None = None
    pairs_df: pl.DataFrame | None = None   # NEW (PHON-106)

    @classmethod
    def from_parquet(cls, words_path: Path) -> "WordStore":
        runtime_dir = words_path.parent
        df = pl.read_parquet(words_path)

        edges_path = runtime_dir / "edges.parquet"
        edges_df = pl.read_parquet(edges_path) if edges_path.exists() else None

        selectional_path = runtime_dir / "selectional.parquet"
        selectional_df = pl.read_parquet(selectional_path) if selectional_path.exists() else None

        pairs_path = runtime_dir / "pairs.parquet"
        pairs_df = pl.read_parquet(pairs_path) if pairs_path.exists() else None

        return cls(df=df, edges_df=edges_df, selectional_df=selectional_df, pairs_df=pairs_df)

If WordStore doesn't currently follow this exact dataclass+classmethod pattern, adapt: the goal is store.pairs_df is populated after from_parquet if pairs.parquet exists.

[ ] Step 4.4: Run test, verify it passes

cd packages/data && uv run python -m pytest tests/test_store.py -v

Expected: 1 passed (or "skipped" if LFS not pulled — that's acceptable in CI without LFS).

[ ] Step 4.5: Commit

git add packages/data/src/phonolex_data/runtime/store.py \
        packages/data/tests/test_store.py
git commit -m "$(cat <<'EOF'
PHON-106: WordStore loads pairs.parquet

Adds pairs_df field to WordStore; from_parquet() reads
data/runtime/pairs.parquet alongside the other runtime artifacts.
Path is None if the file isn't present (back-compat for environments
without the new artifact).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 5: `export-to-d1.py` reads pairs from WordStore¶

Files: - Modify: packages/web/workers/scripts/export-to-d1.py

The current export script reads Derived.minimal_pairs directly. Switch to reading from the new parquet via WordStore so the parquet is the single source of truth. The web's D1 schema only consumes the legacy 6 columns; drop the two new ones during the SQL emit.

[ ] Step 5.1: Update the minimal_pairs section

Find packages/web/workers/scripts/export-to-d1.py. Locate the minimal_pairs SQL emission (around lines 169–179). The current pattern:

        minimal_pairs = derived.minimal_pairs
        for batch_start in range(0, len(minimal_pairs), batch_size):
            ...

Replace with reading from the parquet:

        # PHON-106: read from data/runtime/pairs.parquet (single source of truth).
        # The new parquet has 8 columns; D1 schema consumes the legacy 6.
        # PHON-111 will harmonize D1 to use the continuous feature columns.
        from phonolex_data.runtime.store import WordStore
        store = WordStore.from_parquet(repo_root / "data" / "runtime" / "words.parquet")
        if store.pairs_df is None or store.pairs_df.height == 0:
            raise RuntimeError(
                "data/runtime/pairs.parquet missing or empty. Run "
                "packages/data/scripts/build_runtime_parquet.py first."
            )
        # Project to the legacy 6 columns the D1 schema expects
        minimal_pairs = list(zip(
            store.pairs_df["word1"].to_list(),
            store.pairs_df["word2"].to_list(),
            store.pairs_df["phoneme1"].to_list(),
            store.pairs_df["phoneme2"].to_list(),
            store.pairs_df["position"].to_list(),
            store.pairs_df["position_type"].to_list(),
        ))
        for batch_start in range(0, len(minimal_pairs), batch_size):
            ...  # existing batched INSERT logic uses the 6-element tuples

The repo_root variable should already be defined in the script. If not, add: repo_root = Path(__file__).resolve().parents[4].

[ ] Step 5.2: Smoke-test the export

cd packages/web/workers && uv run python scripts/export-to-d1.py

Expected: completes without error, emits d1-seed.sql. The minimal_pairs section in the SQL should match the previous output (same row count, same INSERT format).

Verify row counts match the pre-change output:

grep -c "INSERT INTO minimal_pairs" packages/web/workers/scripts/d1-seed.sql

[ ] Step 5.3: Commit

git add packages/web/workers/scripts/export-to-d1.py
git commit -m "$(cat <<'EOF'
PHON-106: export-to-d1 reads pairs from runtime parquet

Single source of truth: pairs.parquet feeds both the Python CSP and
the D1 seed. D1 schema consumes only the legacy 6 columns; the new
feature_distance/sonorant_diff columns are projected out at SQL emit
time (PHON-111 will harmonize the web side later).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 6: Replace `ContrastiveConstraint` with three constraint types¶

Files: - Modify: <spike>/constraint_surface.py - Modify: <spike>/paradigm_3_csp.py - Modify: <spike>/paragraph_csp.py - Modify: <spike>/test_vectorized_enumeration.py - Modify: <spike>/eval_hybrid_xcomp_ccomp.py - Modify: <spike>/bench_enumeration.py - Modify: <spike>/bench_domain_cache.py

[ ] Step 6.1: Replace ContrastiveConstraint in constraint_surface.py

Find the existing ContrastiveConstraint dataclass in <spike>/constraint_surface.py:

@dataclass(frozen=True)
class ContrastiveConstraint:
    pair_type: Literal["minpair", "maxopp"]
    phoneme1: str
    phoneme2: str
    position: Literal["initial", "medial", "final", "any"] = "any"
    type: Literal["contrastive"] = "contrastive"

Replace with three new dataclasses:

@dataclass(frozen=True)
class MinpairConstraint:
    phoneme1: str
    phoneme2: str
    position: Literal["initial", "medial", "final", "any"] = "any"
    type: Literal["contrastive_minpair"] = "contrastive_minpair"


@dataclass(frozen=True)
class MaxoppConstraint:
    phoneme1: str
    phoneme2: str
    position: Literal["initial", "medial", "final", "any"] = "any"
    min_sonorant_diff: float = 0.5
    type: Literal["contrastive_maxopp"] = "contrastive_maxopp"


@dataclass(frozen=True)
class MultoppConstraint:
    substitute: str
    targets: tuple[str, ...]
    position: Literal["initial", "medial", "final", "any"] = "any"
    n_targets: int = 3
    type: Literal["contrastive_multopp"] = "contrastive_multopp"

Update the Constraint union:

Constraint = (
    ExcludeConstraint
    | IncludeConstraint
    | BoundConstraint
    | BoundBoostConstraint
    | MinpairConstraint
    | MaxoppConstraint
    | MultoppConstraint
)

[ ] Step 6.2: Replace cross_slot_axes to dispatch on the new types

Find cross_slot_axes in <spike>/constraint_surface.py. Currently it dispatches on c.pair_type ("minpair" / "maxopp"). Replace with:

def cross_slot_axes(
    constraints: list[Constraint],
) -> dict[str, Callable[[dict[str, str], pl.DataFrame], float]]:
    """Return per-axis cross-slot scorers for legacy (post-hoc) contrastive
    scoring. PHON-106 shifts MinpairConstraint and MaxoppConstraint to
    linked-slot enumeration in solve_shape, so these legacy scorers are
    only used by paths that haven't migrated yet (currently: none — kept
    as a transition-period stub; subsequent tasks remove the call sites).
    """
    axes: dict[str, Callable[[dict[str, str], pl.DataFrame], float]] = {}
    # MinpairConstraint / MaxoppConstraint: handled by linked-slot mode in
    # solve_shape (PHON-106). No cross-slot scorer needed.
    # MultoppConstraint: deferred to paragraph integration (PHON-108).
    # Returning empty dict means contrastive constraints don't add legacy
    # cross-axes.
    return axes

This effectively neuters the legacy cross-slot scoring for contrastive constraints. The linked-slot mode in solve_shape (Task 8) will handle them properly.

[ ] Step 6.3: Update imports in dependent spike files

Run a search for ContrastiveConstraint across the spike directory:

grep -rln "ContrastiveConstraint" packages/generation/research/2026-05-07-sentence-generation-paradigms/

For each file that imports ContrastiveConstraint, replace the import with the three new types as needed. Most files won't actually use it after Tasks 7–9; the import can usually be removed. Check the body of each file for actual usages and update:

paradigm_3_csp.py — likely imports for type-hint purposes; if no usages remain, remove the import.
paragraph_csp.py — same.
test_vectorized_enumeration.py — same; existing tests don't assert ContrastiveConstraint behavior, so the import is likely removable.
eval_hybrid_xcomp_ccomp.py — same.
bench_enumeration.py, bench_domain_cache.py — same.

After updates, verify imports compile:

cd packages/generation/research/2026-05-07-sentence-generation-paradigms && \
  uv run python -c "
import constraint_surface
import paradigm_3_csp
import paragraph_csp
import skeleton_csp
print('OK')
"

Expected: OK.

[ ] Step 6.4: Run all spike tests to confirm no regression

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/ -v

Expected: all 65 prior tests pass (the existing minpair-cross-slot test should still pass, since linked-slot mode hasn't been wired in yet — cross_slot_axes returns empty dict for contrastive constraints, but the test in question may have been using the OLD ContrastiveConstraint via the import path and might need to be removed or rewritten in Task 9).

If a test fails referencing ContrastiveConstraint directly: remove or rewrite that test — it's testing the legacy shape we're replacing. The replacement tests land in Task 11.

[ ] Step 6.5: Commit

git add packages/generation/research/2026-05-07-sentence-generation-paradigms/
git commit -m "$(cat <<'EOF'
PHON-106: replace ContrastiveConstraint with three constraint types

ContrastiveConstraint (single dataclass with pair_type field) is split
into MinpairConstraint, MaxoppConstraint, and MultoppConstraint. The
Constraint union extends accordingly. cross_slot_axes is neutered for
contrastive cases (returns empty dict) — the linked-slot mode in
solve_shape (next task) will handle them properly. Legacy minpair
cross-slot scoring is gone; no current test relies on it.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 7: Add `_load_pairs_for_request` helper in `skeleton_csp.py`¶

Files: - Modify: <spike>/skeleton_csp.py - Modify: <spike>/test_contrastive_scorers.py (create)

Helper: given a constraint, the WordStore's pairs_df, the filtered_spec, and an orientation flag, return a Polars DataFrame of (nsubj, dobj, feature_distance, sonorant_diff) rows ready to enter the cartesian.

[ ] Step 7.1: Write failing test

Create <spike>/test_contrastive_scorers.py:

"""Tests for PHON-106 contrastive scorers (linked-slot enumeration)."""
from __future__ import annotations

import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent))

import polars as pl
from phonolex_data.runtime.store import WordStore
from constraint_surface import (
    MaxoppConstraint,
    MinpairConstraint,
    MultoppConstraint,
)
import skeleton_csp


@pytest.fixture(scope="session")
def store():
    repo_root = Path(__file__).resolve().parents[4]
    return WordStore.from_parquet(repo_root / "data" / "runtime" / "words.parquet")


@pytest.fixture(scope="session")
def sel_df():
    repo_root = Path(__file__).resolve().parents[4]
    return pl.read_parquet(repo_root / "data" / "runtime" / "selectional.parquet")


def test_load_pairs_for_minpair_basic(store):
    """Minpair (k, b) at initial position returns a non-empty pair frame
    with the right schema."""
    pairs_df = skeleton_csp._load_pairs_for_request(
        constraint=MinpairConstraint(phoneme1="k", phoneme2="b", position="initial"),
        pairs_df=store.pairs_df,
        filtered_spec=frozenset(["cat", "bat", "kid", "bid", "key", "bee"]),
    )
    assert pairs_df.height > 0
    assert set(pairs_df.columns) >= {"nsubj", "dobj", "feature_distance", "sonorant_diff"}
    # Both halves of any returned pair must be in filtered_spec
    for nsubj, dobj in zip(pairs_df["nsubj"].to_list(), pairs_df["dobj"].to_list()):
        assert nsubj in {"cat", "bat", "kid", "bid", "key", "bee"}
        assert dobj in {"cat", "bat", "kid", "bid", "key", "bee"}


def test_load_pairs_for_maxopp_filters_sonorant_diff(store):
    """Maxopp filters out pair rows where sonorant_diff < min_sonorant_diff."""
    pairs_df = skeleton_csp._load_pairs_for_request(
        constraint=MaxoppConstraint(
            phoneme1="k", phoneme2="b", position="initial",
            min_sonorant_diff=0.5,
        ),
        pairs_df=store.pairs_df,
        filtered_spec=frozenset(["cat", "bat"]),
    )
    # Every retained row has sonorant_diff >= 0.5
    if pairs_df.height > 0:
        for sd in pairs_df["sonorant_diff"].to_list():
            assert sd >= 0.5


def test_load_pairs_emits_both_orientations(store):
    """Both (word1→nsubj, word2→dobj) and the swap are emitted."""
    pairs_df = skeleton_csp._load_pairs_for_request(
        constraint=MinpairConstraint(phoneme1="k", phoneme2="b", position="initial"),
        pairs_df=store.pairs_df,
        filtered_spec=frozenset(["cat", "bat"]),
    )
    nsubj_set = set(pairs_df["nsubj"].to_list())
    dobj_set = set(pairs_df["dobj"].to_list())
    # If both orientations emitted, both nsubj_set and dobj_set should
    # include words from BOTH halves of the pair
    assert nsubj_set & dobj_set, (
        "Expected overlap between nsubj and dobj sets — orientations not swapped?"
    )

[ ] Step 7.2: Run tests, verify they fail

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py -v

Expected: AttributeError on _load_pairs_for_request.

[ ] Step 7.3: Add _load_pairs_for_request to skeleton_csp.py

Append to <spike>/skeleton_csp.py:

from constraint_surface import MaxoppConstraint, MinpairConstraint  # noqa: E402


def _load_pairs_for_request(
    *,
    constraint: MinpairConstraint | MaxoppConstraint,
    pairs_df: pl.DataFrame | None,
    filtered_spec: frozenset[str],
) -> pl.DataFrame:
    """Filter pairs_df to rows matching the constraint and the spec lexicon.
    Emits both orientations (word1→nsubj/word2→dobj and the swap).

    For MaxoppConstraint, additionally filters by sonorant_diff >=
    constraint.min_sonorant_diff.

    Returns a DataFrame with columns: nsubj, dobj, feature_distance,
    sonorant_diff. The cardinality is up to 2 * |matched pair rows|
    after the spec ∩ sonorant filter.
    """
    if pairs_df is None:
        return pl.DataFrame({
            "nsubj": pl.Series(dtype=pl.String),
            "dobj": pl.Series(dtype=pl.String),
            "feature_distance": pl.Series(dtype=pl.Float32),
            "sonorant_diff": pl.Series(dtype=pl.Float32),
        })

    p1, p2 = constraint.phoneme1, constraint.phoneme2

    # Match either orientation in the underlying parquet
    base = pairs_df.filter(
        ((pl.col("phoneme1") == p1) & (pl.col("phoneme2") == p2))
        | ((pl.col("phoneme1") == p2) & (pl.col("phoneme2") == p1))
    )

    # Position filter (skip when "any")
    if constraint.position != "any":
        base = base.filter(pl.col("position_type") == constraint.position)

    # Maxopp pre-filter on sonorant_diff
    if isinstance(constraint, MaxoppConstraint):
        base = base.filter(pl.col("sonorant_diff") >= constraint.min_sonorant_diff)

    # Spec lexicon intersection (both halves must survive)
    base = base.filter(
        pl.col("word1").is_in(list(filtered_spec))
        & pl.col("word2").is_in(list(filtered_spec))
    )

    # Project to (nsubj, dobj, feature_distance, sonorant_diff)
    forward = base.select([
        pl.col("word1").alias("nsubj"),
        pl.col("word2").alias("dobj"),
        pl.col("feature_distance"),
        pl.col("sonorant_diff"),
    ])
    backward = base.select([
        pl.col("word2").alias("nsubj"),
        pl.col("word1").alias("dobj"),
        pl.col("feature_distance"),
        pl.col("sonorant_diff"),
    ])

    return pl.concat([forward, backward])

[ ] Step 7.4: Run tests, verify all pass

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py -v

Expected: 3 passed.

[ ] Step 7.5: Commit

git add packages/generation/research/2026-05-07-sentence-generation-paradigms/skeleton_csp.py \
        packages/generation/research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py
git commit -m "$(cat <<'EOF'
PHON-106: add _load_pairs_for_request helper for linked-slot mode

Filters pairs_df to rows matching the constraint (phoneme1/phoneme2,
position, sonorant_diff for maxopp), intersects with the spec lexicon
on both halves, and emits both orientations (word1→nsubj/word2→dobj
and swap). Returns a 4-column DataFrame ready to enter the cartesian.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 8: Wire linked-slot mode into `_enumerate_vectorized` and `solve_shape`¶

Files: - Modify: <spike>/skeleton_csp.py - Modify: <spike>/test_contrastive_scorers.py

The vectorized enumeration currently builds slot_frames independently per slot. When a contrastive constraint is present, the (nsubj, dobj) pair becomes a linked group: instead of nsubj_frame ⨯ dobj_frame, we use pair_frame directly.

[ ] Step 8.1: Update _should_use_vectorized to narrow contrastive forcing

Find _should_use_vectorized in <spike>/skeleton_csp.py:

def _should_use_vectorized(*, cross_axes: dict) -> bool:
    if _FORCE_PYTHON_PATH:
        return False
    return not cross_axes

Since Task 6 made cross_slot_axes always return empty for contrastive constraints, this function already returns True for minpair/maxopp. But MultoppConstraint should force the python fallback (deferred). Add an explicit constraint check:

def _should_use_vectorized(
    *,
    cross_axes: dict,
    constraints: list | None = None,
) -> bool:
    """Route between vectorized and python fallback paths.

    Vectorized when no cross-slot scorers are registered AND no
    MultoppConstraint is present (multopp deferred to PHON-108).
    """
    if _FORCE_PYTHON_PATH:
        return False
    if cross_axes:
        return False
    if constraints:
        from constraint_surface import MultoppConstraint
        for c in constraints:
            if isinstance(c, MultoppConstraint):
                return False
    return True

Update the callsite in solve_shape to pass constraints. (The call currently passes only cross_axes per _should_use_vectorized(cross_axes=cross_axes). The constraints list is available higher up in solve_shape — search for where contrastive constraints are extracted, or pass the original constraints list down.)

If solve_shape doesn't currently take a constraints parameter, the contrastive constraint must be passed in via an existing mechanism (e.g., paradigm_3_csp.solve passes constraints and extracts cross_axes from them via cross_slot_axes). Check the call chain — cross_axes is built from constraints, so constraints is reachable if we thread it.

The simplest fix: have solve_shape accept an optional constraints parameter (default None) used only for routing. Existing callers don't pass it; only paradigm_3_csp.solve does.

[ ] Step 8.2: Add linked-slot mode to _enumerate_vectorized

In <spike>/skeleton_csp.py, find _enumerate_vectorized. After it builds slot_frames (via _build_slot_filler_tables) and BEFORE the cross-join loop, add a check for linked-slot mode:

The current cartesian construction is roughly:

    cart = tables[shape.slots[0]]
    for s in shape.slots[1:]:
        cart = cart.join(tables[s], how="cross")

Modify _enumerate_vectorized to accept an optional pair_frame parameter. When present, replace nsubj_frame ⨯ dobj_frame with pair_frame in the cross-join chain. Updated signature:

def _enumerate_vectorized(
    shape: SkeletonShape,
    slot_fillers: list[tuple[str, list[str], dict[str, float], dict[str, float]]],
    word_axes: dict[str, dict[str, float]],
    weights: dict[str, float] | None,
    locked_slots: dict[str, str],
    *,
    contrast_pair_frame: pl.DataFrame | None = None,
    contrast_axis_name: str | None = None,
) -> pl.DataFrame:

Inside, after tables = _build_slot_filler_tables(...):

    # Linked-slot mode: replace (nsubj, dobj) per-slot frames with a single
    # pair frame whose rows are valid (nsubj, dobj) joint assignments.
    if contrast_pair_frame is not None:
        # PHON-106: pair_frame already has nsubj, dobj, feature_distance,
        # sonorant_diff columns. Drop the per-slot tables for nsubj/dobj
        # so the cartesian doesn't double-count them.
        if "nsubj" not in shape.slots or "dobj" not in shape.slots:
            raise ValueError(
                "linked-slot mode requires both nsubj and dobj in shape.slots; "
                f"got slots={shape.slots}"
            )
        # Look up PMI scores for the pair-row words against the verb's PMI table
        nsubj_scores = tables["nsubj"]   # used only for its pmi lookup
        dobj_scores = tables["dobj"]
        # Replace the joined nsubj/dobj columns with the pair frame's nsubj/dobj
        # plus the pmi_nsubj/pmi_dobj joined from per-slot tables
        pair_with_pmi = contrast_pair_frame.join(
            nsubj_scores.select(["nsubj", "pmi_nsubj"]),
            on="nsubj",
            how="left",
        ).join(
            dobj_scores.select(["dobj", "pmi_dobj"]),
            on="dobj",
            how="left",
        ).fill_null(0.0)

        # Add the maxopp axis column when contrast_axis_name is provided
        if contrast_axis_name is not None:
            pair_with_pmi = pair_with_pmi.rename(
                {"feature_distance": contrast_axis_name}
            )

        tables["__contrast_group__"] = pair_with_pmi
        # Reorder build to start from the contrast group, then cross-join
        # remaining non-(nsubj/dobj) slots
        ordered_slots = ["__contrast_group__"] + [
            s for s in shape.slots if s not in ("nsubj", "dobj")
        ]
        cart = tables[ordered_slots[0]]
        for s in ordered_slots[1:]:
            cart = cart.join(tables[s], how="cross")
        # Skip the per-slot Cartesian path
    else:
        # Standard mode: independent per-slot Cartesian
        cart = tables[shape.slots[0]]
        for s in shape.slots[1:]:
            cart = cart.join(tables[s], how="cross")

Then drop the standard nsubj != dobj filter when we're in linked-slot mode (the pair list already guarantees distinct words):

    # nsubj != dobj invariant — only enforce in standard mode (linked-slot mode
    # already only emits valid pair rows)
    if contrast_pair_frame is None and "nsubj" in shape.slots and "dobj" in shape.slots:
        cart = cart.filter(pl.col("nsubj") != pl.col("dobj"))

The remainder of _enumerate_vectorized (per-word axes, adv_sentinel, total_score) operates on the merged cart — nsubj, dobj, and pmi_nsubj, pmi_dobj columns are all present whether linked-slot or standard, so the score expressions don't need modification.

[ ] Step 8.3: Wire solve_shape to detect and dispatch linked-slot mode

Find solve_shape. Add (near the top, after slot_fillers is built):

    # PHON-106: detect contrastive constraints and prepare linked-slot mode
    contrast_pair_frame: pl.DataFrame | None = None
    contrast_axis_name: str | None = None
    constraints_list = constraints if isinstance(constraints, list) else []
    contrast_constraints = [
        c for c in constraints_list
        if isinstance(c, (MinpairConstraint, MaxoppConstraint))
    ]
    if contrast_constraints:
        if len(contrast_constraints) > 1:
            raise ValueError(
                "at most one contrastive constraint per request"
            )
        cc = contrast_constraints[0]
        # Validate shape: needs >= 2 nominal content slots
        nominal_content = [
            s for s in shape.content_slots
            if s in ("nsubj", "dobj", "iobj") or s.startswith("pobj_")
        ]
        if len(nominal_content) < 2:
            raise ValueError(
                "contrastive constraint needs >= 2 content nominal slots; "
                f"shape={shape.arg_structure} has {len(nominal_content)}"
            )
        # For v1, only the (nsubj, dobj) pair is supported
        if "nsubj" not in shape.slots or "dobj" not in shape.slots:
            raise ValueError(
                "contrastive constraint requires (nsubj, dobj) in shape; "
                f"got slots={shape.slots}"
            )
        contrast_pair_frame = _load_pairs_for_request(
            constraint=cc,
            pairs_df=word_store.pairs_df if word_store else None,
            filtered_spec=frozenset(domain_words),
        )
        if contrast_pair_frame.height == 0:
            return []  # no valid pairs after filtering
        if isinstance(cc, MaxoppConstraint):
            contrast_axis_name = f"contrast_maxopp_{cc.phoneme1}_{cc.phoneme2}"

    # Multopp constraint: error in single-sentence mode (deferred to PHON-108)
    multopp = [c for c in constraints_list if isinstance(c, MultoppConstraint)]
    if multopp:
        raise ValueError(
            "MultoppConstraint requires multi-sentence paragraph composition; "
            "not implemented in v1 (deferred to PHON-108)"
        )

Pass contrast_pair_frame and contrast_axis_name through to _enumerate_vectorized:

    if _should_use_vectorized(cross_axes=cross_axes, constraints=constraints_list):
        cart = _enumerate_vectorized(
            shape=shape,
            slot_fillers=slot_fillers,
            word_axes=word_axes,
            weights=weights,
            locked_slots=initial_locks,
            contrast_pair_frame=contrast_pair_frame,
            contrast_axis_name=contrast_axis_name,
        )
        ...

solve_shape needs access to the WordStore for word_store.pairs_df. The current signature passes word_df (a pl.DataFrame), not the full store. The simplest fix: have callers pass pairs_df explicitly via a new optional kwarg pairs_df: pl.DataFrame | None = None. Update paradigm_3_csp.solve() to thread it from WordStore:

In paradigm_3_csp.solve():

    # near the existing word_df handling:
    pairs_df = store.pairs_df if hasattr(store, "pairs_df") else None
    # pass to solve_shape:
    solve_shape(..., pairs_df=pairs_df)

In solve_shape's signature:

def solve_shape(
    shape: SkeletonShape,
    *,
    verb: str,
    domain_words: set[str] | frozenset[str],
    sel_df: pl.DataFrame,
    band: str,
    word_axes: dict[str, dict[str, float]],
    cross_axes: dict,
    word_df: pl.DataFrame | None = None,
    pairs_df: pl.DataFrame | None = None,   # NEW (PHON-106)
    weights: dict[str, float] | None = None,
    top_k: int = 8,
    locked_slots: dict[str, str] | None = None,
    constraints: list | None = None,
    ...

(Adapt to whatever solve_shape's actual current signature is — the goal is pairs_df is reachable inside solve_shape.)

[ ] Step 8.4: Add tests for linked-slot enumeration

Append to <spike>/test_contrastive_scorers.py:

def test_minpair_linked_slot_realization(store, sel_df):
    """Every output's (nsubj, dobj) IS a valid (p1, p2) minimal pair."""
    from skeleton_csp import SkeletonShape, parse_arg_structure, solve_shape
    import paradigm_3_csp

    shape = SkeletonShape("nsubj,V,dobj", parse_arg_structure("nsubj,V,dobj"), 0)
    spec_words = paradigm_3_csp.spec_lexicon(store, "spec1")
    constraint = MinpairConstraint(phoneme1="k", phoneme2="b", position="initial")

    top = solve_shape(
        shape, verb="cut", domain_words=spec_words, sel_df=sel_df,
        band="fineweb_adult", word_axes={}, cross_axes={},
        word_df=store.df, pairs_df=store.pairs_df, top_k=8,
        constraints=[constraint],
    )
    assert top, "expected candidates"
    # Verify every (nsubj, dobj) pair in top is in pairs_df for (k, b) initial
    valid_pairs = set()
    for row in store.pairs_df.filter(
        ((pl.col("phoneme1") == "k") & (pl.col("phoneme2") == "b"))
        | ((pl.col("phoneme1") == "b") & (pl.col("phoneme2") == "k"))
    ).filter(pl.col("position_type") == "initial").iter_rows(named=True):
        valid_pairs.add((row["word1"], row["word2"]))
        valid_pairs.add((row["word2"], row["word1"]))

    for c in top:
        nsubj, dobj = c["fillers"]["nsubj"], c["fillers"]["dobj"]
        assert (nsubj, dobj) in valid_pairs, (
            f"{nsubj}, {dobj} not a valid (k, b) initial minimal pair"
        )


def test_minpair_single_content_slot_errors(store, sel_df):
    """Minpair on a shape with < 2 nominal content slots raises ValueError."""
    from skeleton_csp import SkeletonShape, parse_arg_structure, solve_shape
    import paradigm_3_csp

    shape = SkeletonShape("nsubj,V,xcomp", parse_arg_structure("nsubj,V,xcomp"), 0)
    spec_words = paradigm_3_csp.spec_lexicon(store, "spec1")
    with pytest.raises(ValueError, match="content nominal slots"):
        solve_shape(
            shape, verb="want", domain_words=spec_words, sel_df=sel_df,
            band="fineweb_adult", word_axes={}, cross_axes={},
            word_df=store.df, pairs_df=store.pairs_df, top_k=3,
            constraints=[MinpairConstraint(phoneme1="k", phoneme2="b")],
        )


def test_multopp_in_single_sentence_errors(store, sel_df):
    """MultoppConstraint raises ValueError in single-sentence shape."""
    from skeleton_csp import SkeletonShape, parse_arg_structure, solve_shape
    import paradigm_3_csp

    shape = SkeletonShape("nsubj,V,dobj", parse_arg_structure("nsubj,V,dobj"), 0)
    spec_words = paradigm_3_csp.spec_lexicon(store, "spec1")
    with pytest.raises(ValueError, match="paragraph composition"):
        solve_shape(
            shape, verb="cut", domain_words=spec_words, sel_df=sel_df,
            band="fineweb_adult", word_axes={}, cross_axes={},
            word_df=store.df, pairs_df=store.pairs_df, top_k=3,
            constraints=[MultoppConstraint(
                substitute="t", targets=("s", "ʃ", "tʃ"),
            )],
        )


def test_both_minpair_and_maxopp_errors(store, sel_df):
    """Two contrastive constraints in one request raises ValueError."""
    from skeleton_csp import SkeletonShape, parse_arg_structure, solve_shape
    import paradigm_3_csp

    shape = SkeletonShape("nsubj,V,dobj", parse_arg_structure("nsubj,V,dobj"), 0)
    spec_words = paradigm_3_csp.spec_lexicon(store, "spec1")
    with pytest.raises(ValueError, match="at most one contrastive"):
        solve_shape(
            shape, verb="cut", domain_words=spec_words, sel_df=sel_df,
            band="fineweb_adult", word_axes={}, cross_axes={},
            word_df=store.df, pairs_df=store.pairs_df, top_k=3,
            constraints=[
                MinpairConstraint(phoneme1="k", phoneme2="b"),
                MaxoppConstraint(phoneme1="k", phoneme2="m"),
            ],
        )

[ ] Step 8.5: Run tests, verify all pass

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py -v

Expected: 7 passed (3 from Task 7 + 4 new).

[ ] Step 8.6: Smoke-test paradigm_3_csp.solve with a minpair constraint

cd packages/generation/research/2026-05-07-sentence-generation-paradigms && \
  uv run python -c "
import paradigm_3_csp
from constraint_surface import MinpairConstraint
from phonolex_data.runtime.store import WordStore
from pathlib import Path
import polars as pl

repo = Path('../../../..').resolve()
store = WordStore.from_parquet(repo / 'data' / 'runtime' / 'words.parquet')
sel_df = pl.read_parquet(repo / 'data' / 'runtime' / 'selectional.parquet')
spec_words = paradigm_3_csp.spec_lexicon(store, 'spec1')
top, stats = paradigm_3_csp.solve(
    'cut', 'spec1', spec_words, sel_df,
    constraints=[MinpairConstraint(phoneme1='k', phoneme2='b', position='initial')],
    word_df=store.df,
)
print(f'top-1: {top[0][\"sentence\"]!r}')
print(f'fillers: {top[0][\"fillers\"]}')
"

Expected: prints a sentence whose nsubj and dobj fillers form a real (k, b) initial minimal pair (e.g., "the cat cuts the bat").

[ ] Step 8.7: Run all spike tests

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/ -v

Expected: 65 prior + 7 new = 72 passed.

[ ] Step 8.8: Commit

git add packages/generation/research/2026-05-07-sentence-generation-paradigms/skeleton_csp.py \
        packages/generation/research/2026-05-07-sentence-generation-paradigms/paradigm_3_csp.py \
        packages/generation/research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py
git commit -m "$(cat <<'EOF'
PHON-106: linked-slot mode in solve_shape for minpair/maxopp

solve_shape detects MinpairConstraint or MaxoppConstraint, validates
that the shape has both nsubj and dobj content slots, loads matching
pair rows from pairs_df, and passes them into _enumerate_vectorized
as a contrast_pair_frame. _enumerate_vectorized replaces the
independent (nsubj, dobj) slot Cartesian with the pair frame —
guaranteeing every output candidate's (nsubj, dobj) is a valid
minimal pair contrasting the requested phonemes.

MaxoppConstraint additionally pre-filters by sonorant_diff and adds a
contrast_maxopp_<p1>_<p2> = feature_distance score component.

MultoppConstraint and shapes without (nsubj, dobj) raise ValueError.
Two contrastive constraints in one request also raise.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Task 9: Edge case tests + documentation polish¶

Files: - Modify: <spike>/test_contrastive_scorers.py

Catch the remaining edge cases and add a maxopp-specific test for the feature_distance score component.

[ ] Step 9.1: Add tests

Append to <spike>/test_contrastive_scorers.py:

def test_maxopp_feature_distance_in_components(store, sel_df):
    """MaxoppConstraint adds contrast_maxopp_<p1>_<p2> = feature_distance to components."""
    from skeleton_csp import SkeletonShape, parse_arg_structure, solve_shape
    import paradigm_3_csp

    shape = SkeletonShape("nsubj,V,dobj", parse_arg_structure("nsubj,V,dobj"), 0)
    spec_words = paradigm_3_csp.spec_lexicon(store, "spec1")
    constraint = MaxoppConstraint(phoneme1="k", phoneme2="m", position="initial")

    top = solve_shape(
        shape, verb="cut", domain_words=spec_words, sel_df=sel_df,
        band="fineweb_adult", word_axes={}, cross_axes={},
        word_df=store.df, pairs_df=store.pairs_df, top_k=3,
        constraints=[constraint],
    )
    if not top:
        pytest.skip("no (k, m) initial pairs in spec1 — adjust probe verb")

    axis_name = "contrast_maxopp_k_m"
    for c in top:
        assert axis_name in c["score_components"]
        # Feature distance is positive for any actual pair
        assert c["score_components"][axis_name] > 0


def test_pairs_empty_after_filter_returns_no_candidates(store, sel_df):
    """If pair_rows are empty after spec ∩ sonorant filter, solve_shape returns []."""
    from skeleton_csp import SkeletonShape, parse_arg_structure, solve_shape
    import paradigm_3_csp

    shape = SkeletonShape("nsubj,V,dobj", parse_arg_structure("nsubj,V,dobj"), 0)
    # Use an unlikely (p1, p2) pair to force empty
    spec_words = paradigm_3_csp.spec_lexicon(store, "spec1")
    constraint = MinpairConstraint(phoneme1="zh", phoneme2="dh", position="initial")
    top = solve_shape(
        shape, verb="cut", domain_words=spec_words, sel_df=sel_df,
        band="fineweb_adult", word_axes={}, cross_axes={},
        word_df=store.df, pairs_df=store.pairs_df, top_k=8,
        constraints=[constraint],
    )
    assert top == [], f"expected empty top for unlikely pair, got {len(top)} candidates"


def test_minpair_uses_vectorized_path():
    """MinpairConstraint no longer forces python fallback (PHON-104 fix
    routed it to fallback; PHON-106 routes it to vectorized linked-slot)."""
    import skeleton_csp

    # cross_axes is empty for new-style contrastive constraints
    cross = {}
    constraints = [MinpairConstraint(phoneme1="k", phoneme2="b")]
    assert skeleton_csp._should_use_vectorized(
        cross_axes=cross, constraints=constraints
    ) is True


def test_multopp_forces_python_fallback_in_routing():
    """MultoppConstraint routes to python fallback (deferred to PHON-108)."""
    import skeleton_csp

    cross = {}
    constraints = [MultoppConstraint(substitute="t", targets=("s",))]
    assert skeleton_csp._should_use_vectorized(
        cross_axes=cross, constraints=constraints
    ) is False

[ ] Step 9.2: Run tests, verify all pass

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py -v

Expected: 11 passed (7 prior + 4 new).

[ ] Step 9.3: Run all spike tests

cd packages/generation && uv run python -m pytest research/2026-05-07-sentence-generation-paradigms/ -v

Expected: 76 passed total (65 prior + 11 new contrastive).

[ ] Step 9.4: Commit

git add packages/generation/research/2026-05-07-sentence-generation-paradigms/test_contrastive_scorers.py
git commit -m "$(cat <<'EOF'
PHON-106: edge-case tests for contrastive linked-slot mode

Adds: maxopp feature_distance component verification; empty-after-
filter returns []; routing correctness (minpair → vectorized,
multopp → python fallback).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
EOF
)"

Done¶

After Task 9 commits, PHON-106 v1 is complete. PHON-108 (paragraph integration for multopp) is unblocked. PHON-111 (web app continuous-vector parity) is filed and independent.

PHON-106 — CSP Contrastive Scorers Implementation Plan¶

File map¶

Task 1: Add _phoneme_pair_distances helper in derived.py¶

Task 2: Extend _compute_minimal_pairs to emit 8-tuple¶

Task 3: Add pairs_schema() and emit pairs.parquet¶

Task 4: Load pairs_df in WordStore.from_parquet¶

Task 5: export-to-d1.py reads pairs from WordStore¶

Task 6: Replace ContrastiveConstraint with three constraint types¶

Task 7: Add _load_pairs_for_request helper in skeleton_csp.py¶

Task 8: Wire linked-slot mode into _enumerate_vectorized and solve_shape¶