diff --git a/src/srdatalog/ir/dialects/relation/sorted_array/__init__.py b/src/srdatalog/ir/dialects/relation/sorted_array/__init__.py index 3378da4..a4928a0 100644 --- a/src/srdatalog/ir/dialects/relation/sorted_array/__init__.py +++ b/src/srdatalog/ir/dialects/relation/sorted_array/__init__.py @@ -123,9 +123,78 @@ 'SaValid', 'SaView', 'register', + 'register_terminal_wrap_op', ] +# --------------------------------------------------------------------------- +# Terminal-wrap-op registry (dispatcher extensibility for external plugins) +# --------------------------------------------------------------------------- +# +# The sorted_array chain dispatcher (`_lower_inner_chain`, +# `_supported_pipeline`, `_trailing_inserts` in `lowerings/__init__.py`) +# historically matched only `mir.InsertInto` as the terminal op in an +# `ExecutePipeline.pipeline`. The built-in C-pragma wrap ops +# (`DedupGate` / `WSScope` / `mir.FanOut`) dodged this by routing +# emission through the dual-write bool path on `ExecutePipeline`: +# `materialize_*` short-circuits on `ep.=True`, leaving the +# legacy bool-driven `_lower_insert_into` branch to emit. External +# plugins (e.g. PR #68's jaccard demo) have no equivalent escape +# hatch — they cannot add a back-compat bool field to +# `mir.ExecutePipeline`, so their terminal wrap ops are rejected by +# `_supported_pipeline`. +# +# The registry below opens an ALTERNATE path: any MIR op type +# registered here is treated as "InsertInto-like terminal wrap op" +# by `_trailing_inserts` / `_supported_pipeline`, so it may appear +# at the tail of a pipeline. The built-in wrap ops are pre-populated +# at module-load time so byte-equivalence holds when their dual-write +# bool flips off (Phase A3); external plugins call +# `register_terminal_wrap_op()` from their +# `register(compiler)` callable. +# +# Approach is "option 1" from PR #68's contract-gap discussion — the +# mechanical fix. A future PR may switch the dispatcher to consult +# the `@lowering` registry directly (option 2) and treat any op with +# a registered `iir.cf` lowering as terminal; that is a bigger +# refactor and out of scope here. + +_TERMINAL_WRAP_OPS: set[type] = set() + + +def register_terminal_wrap_op(op_type: type) -> None: + '''Register a MIR op type as a terminal (InsertInto-like) wrap op + for the sorted_array dispatcher. + + Called by built-in pragma modules at module-load time AND by + external plugins from their `register(compiler)` callable. Once + registered, the op type may appear at the tail of an + `ExecutePipeline.pipeline` and `_supported_pipeline` / + `_trailing_inserts` will accept it in the same slot as + `mir.InsertInto`. + + Idempotent: re-registering the same op type is a no-op (set + semantics). Registration order is irrelevant — the dispatcher + consults the set by `type(op) in _TERMINAL_WRAP_OPS`. + ''' + _TERMINAL_WRAP_OPS.add(op_type) + + +def _register_builtin_terminal_wrap_ops() -> None: + '''Pre-populate `_TERMINAL_WRAP_OPS` with the built-in C-pragma + wrap ops. Deferred import to keep the module import graph linear + (mir.types is the canonical home for these op types). + ''' + import srdatalog.ir.mir.types as mir + + register_terminal_wrap_op(mir.DedupGate) + register_terminal_wrap_op(mir.WSScope) + register_terminal_wrap_op(mir.FanOut) + + +_register_builtin_terminal_wrap_ops() + + # --------------------------------------------------------------------------- # Pass registration (S3A.4) # --------------------------------------------------------------------------- diff --git a/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py b/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py index adaf1c6..c7872c9 100644 --- a/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py +++ b/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py @@ -18,6 +18,7 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass, field import srdatalog.ir.mir.types as mir @@ -176,25 +177,122 @@ def fresh(self, prefix: str) -> str: # ----------------------------------------------------------------------------- -def _trailing_inserts(rest: list[mir.MirNode]) -> list[mir.InsertInto]: - '''Return the trailing run of InsertIntos at the end of `rest`. +def _lookup_terminal_wrap_lowering( + op_type: type, +) -> Callable[[mir.MirNode, LoweringCtx], Op]: + '''Find the dialect's `@lowering` registered for `op_type`. - Multi-head rules emit several InsertIntos in sequence at the end - of the pipeline. The legacy emitter walks them all in order. + Returns the `apply` callable (signature `(op, ctx) -> Op`). Raises + `LookupError` if `op_type` is in `_TERMINAL_WRAP_OPS` but has no + registered lowering — a misconfiguration: registry registration + alone is not enough, callers must also `@lowering(..., op_type, ...)` + on the dialect for the actual emission. ''' - out: list[mir.InsertInto] = [] + from srdatalog.ir.dialects.relation.sorted_array import DIALECT + + for low in DIALECT.lowerings: + if low.matches is op_type: + # `Lowering.apply` is typed `Callable[[Any, Any], Any]` on the + # framework side; narrow back to the dispatcher's contract + # here. The cast is safe because every entry in + # `_TERMINAL_WRAP_OPS` requires a paired `@lowering` that + # returns an `iir.cf.Op` for `(mir.MirNode, LoweringCtx)`. + apply_fn: Callable[[mir.MirNode, LoweringCtx], Op] = low.apply + return apply_fn + raise LookupError( + f'no @lowering registered on {DIALECT.name!r} for ' + f'{op_type.__name__!r}. `register_terminal_wrap_op` was called for ' + f'this op type, but the dispatcher needs an `@lowering(DIALECT, ' + f'{op_type.__name__}, ...)` rule to emit it. Built-in wrap ops ' + f'register both in `sorted_array/__init__.py:_register_passes`; ' + f'external plugins must register both before any pipeline using ' + f'the wrap op reaches lowering.' + ) + + +def _dispatch_terminal_wrap_run(rest: list[mir.MirNode], ctx: LoweringCtx) -> Op: + '''Emit each terminal wrap op in `rest` by dispatching to its + registered `@lowering`. The result is a `Block` concatenating the + per-op emissions, matching the legacy multi-head InsertInto run + shape. + + Mixed runs (some `mir.InsertInto`, some wrap op) are supported — + bare InsertInto goes through `_lower_insert_into` exactly as the + pure-InsertInto path does; wrap ops dispatch through their + registered `@lowering`. The legacy multi-head emit pattern is one + bare InsertInto per output, so wrap-op materialize handlers + preserve that by emitting one wrap-op-per-InsertInto. + ''' + from srdatalog.ir.dialects.relation.sorted_array import _TERMINAL_WRAP_OPS + + stmts: list[Op] = [] for op in rest: if isinstance(op, mir.InsertInto): + stmts.extend(_lower_insert_into(op, ctx)) + elif type(op) in _TERMINAL_WRAP_OPS: + apply_fn = _lookup_terminal_wrap_lowering(type(op)) + result = apply_fn(op, ctx) + # The wrap-op lowerings (lower_dedup_gate / lower_ws_scope / + # lower_fan_out / external plugin lowerings) return a single Op + # (typically a Block wrapping the emitted statements). Append it + # as one statement; the outer Block flattens via normal IIR + # rendering. + stmts.append(result) + else: + raise ValueError( + f'_dispatch_terminal_wrap_run: expected InsertInto or registered ' + f'terminal wrap op, got {type(op).__name__!r}' + ) + return Block(stmts=tuple(stmts)) + + +def _is_terminal_op(op: mir.MirNode) -> bool: + '''True iff `op` is `mir.InsertInto` OR a registered terminal + wrap op (InsertInto-like — DedupGate / WSScope / FanOut today, + plus anything registered by external plugins via + `register_terminal_wrap_op`). + + Deferred import: `_TERMINAL_WRAP_OPS` lives on the dialect + package; importing it at module-load time would create a cycle + (the dialect package's `_register_passes` imports back from + this module). + ''' + if isinstance(op, mir.InsertInto): + return True + from srdatalog.ir.dialects.relation.sorted_array import _TERMINAL_WRAP_OPS + + return type(op) in _TERMINAL_WRAP_OPS + + +def _trailing_inserts(rest: list[mir.MirNode]) -> list[mir.MirNode]: + '''Return the trailing run of terminal ops at the end of `rest`. + + A "terminal op" is `mir.InsertInto` OR any MIR op type registered + via `register_terminal_wrap_op` (the built-in C-pragma wrap ops + DedupGate / WSScope / FanOut, plus anything an external plugin + registers from its `register(compiler)` callable). Multi-head + rules emit several InsertIntos in sequence at the end of the + pipeline; the legacy emitter walks them all in order. + + Returns `list[mir.MirNode]` (not `list[mir.InsertInto]`) because + the registry admits non-InsertInto types — call sites that need + pure-InsertInto narrowing must do their own isinstance filter + (today, only `_lower_inner_chain`'s terminal-run emission cares, + and it consults `_lower_insert_into` which expects InsertInto). + ''' + out: list[mir.MirNode] = [] + for op in rest: + if _is_terminal_op(op): out.append(op) elif out: - # Non-InsertInto after InsertInto: the trailing run is just - # the contiguous tail. Stop here and let the caller decide. + # Non-terminal after terminal: the trailing run is just the + # contiguous tail. Stop here and let the caller decide. break return out def _middle_ops(rest: list[mir.MirNode]) -> list[mir.MirNode]: - '''Return `rest` with the trailing InsertIntos stripped off.''' + '''Return `rest` with the trailing terminal ops stripped off.''' trailing_count = len(_trailing_inserts(rest)) return rest[: len(rest) - trailing_count] if trailing_count else list(rest) @@ -202,23 +300,25 @@ def _middle_ops(rest: list[mir.MirNode]) -> list[mir.MirNode]: def _supported_pipeline(ops: list[mir.MirNode]) -> bool: '''True iff the dialect can lower this pipeline shape today. - The pipeline must end in one or more InsertIntos (multi-head rules - emit several outputs from the same body). The middle ops between - the head op and the first InsertInto are constrained per the - milestone (Scan/CJ/Cart/Filter/ConstantBind/Negation). + The pipeline must end in one or more terminal ops (`mir.InsertInto` + or any registered terminal wrap op — see `_is_terminal_op`). + Multi-head rules emit several outputs from the same body. The + middle ops between the head op and the first terminal op are + constrained per the milestone (Scan/CJ/Cart/Filter/ConstantBind/ + Negation). ''' if len(ops) < 2: return False - # Find the start of the trailing InsertInto sequence. + # Find the start of the trailing terminal sequence. insert_start = None for i, op in enumerate(ops): - if isinstance(op, mir.InsertInto): + if _is_terminal_op(op): insert_start = i break if insert_start is None or insert_start == 0: return False - # All ops from insert_start onward must be InsertInto. - if not all(isinstance(op, mir.InsertInto) for op in ops[insert_start:]): + # All ops from insert_start onward must be terminal. + if not all(_is_terminal_op(op) for op in ops[insert_start:]): return False head = ops[0] middle = ops[1:insert_start] @@ -1249,17 +1349,41 @@ def _lower_inner_chain( f'_lower_inner_chain: expected pure InsertInto tail at this ' f'point, got {[type(o).__name__ for o in rest]}' ) + # `_trailing_inserts` admits registered terminal wrap ops; narrow + # back to InsertInto for the legacy emit path. Mixed-tail (some + # InsertInto, some wrap op) is structurally legal under the + # registry but does not occur today — built-in pragma materialize + # handlers wrap every InsertInto uniformly. If a mixed tail + # appears, the wrap-op branch below handles it. + pure_inserts = [op for op in inserts if isinstance(op, mir.InsertInto)] + if len(pure_inserts) != len(inserts): + return _dispatch_terminal_wrap_run(inserts, ctx) # Tiled-Cartesian ballot variant: render the trailing InsertInto # run as a single TiledBallotBlock that owns the ballot setup + # per-output write at once. Replaces the legacy stateful # `tiled_cartesian_ballot_done` flag. if ctx.tiled_cartesian_valid_var: - return _lower_tiled_ballot_block(inserts, ctx) + return _lower_tiled_ballot_block(pure_inserts, ctx) stmts: list[Op] = [] - for ins in inserts: + for ins in pure_inserts: stmts.extend(_lower_insert_into(ins, ctx)) return Block(stmts=tuple(stmts)) + # Terminal wrap op at head: dispatch through the registered + # `@lowering` on the dialect. This is the extension point for + # built-in C-pragma wrap ops (DedupGate / WSScope / FanOut) AND + # external plugin wrap ops (registered via + # `register_terminal_wrap_op` from a plugin's `register(compiler)` + # callable). Pre-A3 the built-ins reach this branch ONLY in the + # pure-typed-pragma test path (the dual-write bool short-circuit + # in `materialize_*` keeps them out of the production pipeline); + # post-A3 the built-ins lose the dual-write escape hatch and this + # is the SOLE dispatch path for every wrap-op-terminated EP. + from srdatalog.ir.dialects.relation.sorted_array import _TERMINAL_WRAP_OPS + + if type(head) in _TERMINAL_WRAP_OPS: + return _dispatch_terminal_wrap_run(rest, ctx) + if isinstance(head, mir.Filter): cond_expr = _filter_expr(head.code) body_op = _lower_inner_chain(tail, ctx) @@ -2229,12 +2353,15 @@ def _lower_nested_cart( def _cart_var_used( var_name: str, rest: list[mir.MirNode], - inserts: list[mir.InsertInto], + inserts: list[mir.MirNode], ) -> bool: '''Counting-phase optimization gate for Cartesian var-binds. - Multi-head: any of the trailing InsertIntos referencing `var_name` - counts as a usage. + Multi-head: any of the trailing terminal ops referencing `var_name` + counts as a usage. `inserts` is the result of `_trailing_inserts` + which now admits registered terminal wrap ops in addition to bare + `mir.InsertInto`; `_var_used_in_op` walks string fields uniformly, + so the widening is transparent. ''' return any(_var_used_in_op(var_name, op) for op in (*rest, *inserts)) diff --git a/tests/test_sorted_array_terminal_wrap_registry.py b/tests/test_sorted_array_terminal_wrap_registry.py new file mode 100644 index 0000000..e61cd23 --- /dev/null +++ b/tests/test_sorted_array_terminal_wrap_registry.py @@ -0,0 +1,324 @@ +'''Tests for the sorted_array terminal-wrap-op registry. + +Covers the dispatcher gap revealed by PR #68 (jaccard external plugin +demo): the sorted_array chain dispatcher historically matched only +`mir.InsertInto` as the terminal op in an `ExecutePipeline.pipeline`. +External plugins whose wrap ops appear at the tail of the pipeline +were rejected by `_supported_pipeline`. The fix: + + - A module-level `_TERMINAL_WRAP_OPS: set[type]` registry on the + dialect (`srdatalog.ir.dialects.relation.sorted_array`). + - A public `register_terminal_wrap_op(op_type)` helper for external + plugins to call from their `register(compiler)` callable. + - `_trailing_inserts` / `_supported_pipeline` consult the registry, + so any registered wrap op may appear at the tail. + - `_lower_inner_chain` dispatches a registered wrap op at the head + through the dialect's `@lowering` registry for that op type. + +This test exercises: + + 1. Registry shape — pre-populated with built-in wrap ops, idempotent + re-registration, accepts arbitrary types. + 2. Predicate behaviour — `_supported_pipeline` accepts a synthetic + EP whose tail is a registered mock wrap op. + 3. End-to-end dispatch — `compile_pipeline(ep)` lowers a synthetic + EP with a registered mock wrap op at the tail (the integration + gate from the task brief). + +PR #68's jaccard demo is INTENTIONALLY NOT exercised here — that demo +isn't merged into design/redesign-package, so its sources are not +present in the worktree. The synthetic mock wrap op below is enough +to gate the dispatcher contract; the jaccard demo will land once PR +#68 merges and `srdatalog_jaccard.register` adds a +`register_terminal_wrap_op(JaccardIndex)` call. +''' + +from __future__ import annotations + +from dataclasses import dataclass +from typing import final + +import pytest + +import srdatalog.ir.mir.types as m +from srdatalog.ir.codegen.cuda.api import compile_pipeline +from srdatalog.ir.core import Op +from srdatalog.ir.core.passes import lowering +from srdatalog.ir.dialects.relation.sorted_array import ( + _TERMINAL_WRAP_OPS, + register_terminal_wrap_op, +) +from srdatalog.ir.dialects.relation.sorted_array import ( + DIALECT as SA_DIALECT, +) +from srdatalog.ir.dialects.relation.sorted_array.lowerings import ( + _is_terminal_op, + _supported_pipeline, + _trailing_inserts, +) +from srdatalog.ir.hir.types import Version + +# ----------------------------------------------------------------------------- +# Synthetic mock wrap op + lowering +# ----------------------------------------------------------------------------- +# +# A minimal stand-in for an external plugin's terminal wrap op. The +# dialect knows nothing about `MockExtensionWrap` until the test +# registers it; mirrors what an external plugin's `register(compiler)` +# would do. + + +@final +@dataclass(frozen=True, slots=True) +class MockExtensionWrap(Op): + '''Synthetic terminal wrap op for the registry test. + + Wraps an `mir.InsertInto` and is lowered by delegating straight + back to the dialect's `_lower_insert_into` helper, so the resulting + IIR is byte-equivalent to a bare `InsertInto` for the same inner + op. The exact IIR isn't load-bearing here — what matters is that + the dispatcher routes through the registered `@lowering` instead + of raising `unsupported inner op`. + ''' + + inner: m.InsertInto + + +def _lower_mock_extension_wrap(op, ctx): + '''Mock @lowering body: delegate to the dialect's + `_lower_insert_into` so the rendered CUDA matches the no-wrap + emission for the same `InsertInto`. + ''' + from srdatalog.ir.dialects.iir.cf import Block + from srdatalog.ir.dialects.relation.sorted_array.lowerings import ( + _lower_insert_into, + ) + + stmts = _lower_insert_into(op.inner, ctx) + return Block(stmts=tuple(stmts)) + + +# Wire the mock wrap op into the dialect ONCE per process — the +# discipline tests in `tests/test_discipline_*.py` assert exactly one +# `@lowering` per op type, so re-registering would trip +# `AmbiguousLowering`. The module-level guard below short-circuits +# re-imports under `pytest`'s plugin discovery. + +_MOCK_REGISTERED = False + + +def _ensure_mock_registered() -> None: + global _MOCK_REGISTERED + if _MOCK_REGISTERED: + return + # The @lowering decorator appends to DIALECT.lowerings; guard + # against double-registration (e.g. test re-runs in the same + # process) by checking first. + already = any(low.matches is MockExtensionWrap for low in SA_DIALECT.lowerings) + if not already: + lowering( + SA_DIALECT, + MockExtensionWrap, + consumes=('mir',), + produces=('iir.cf',), + )(_lower_mock_extension_wrap) + register_terminal_wrap_op(MockExtensionWrap) + _MOCK_REGISTERED = True + + +# ----------------------------------------------------------------------------- +# 1. Registry shape — built-ins + idempotency + arbitrary types +# ----------------------------------------------------------------------------- + + +def test_builtin_wrap_ops_are_preregistered(): + '''The dialect pre-populates the registry with the three built-in + C-pragma wrap ops at module-load time. This is the fallback that + keeps byte-equivalence intact once Phase A3 drops the dual-write + bool fields. + ''' + assert m.DedupGate in _TERMINAL_WRAP_OPS + assert m.WSScope in _TERMINAL_WRAP_OPS + assert m.FanOut in _TERMINAL_WRAP_OPS + + +def test_register_terminal_wrap_op_adds_op_type(): + '''An external plugin's wrap op type lands in the registry after + `register_terminal_wrap_op(...)` is called. Stand-in for the + jaccard demo's `register(compiler)` callable. + ''' + + @final + @dataclass(frozen=True, slots=True) + class _NeverSeenWrap(Op): + pass + + assert _NeverSeenWrap not in _TERMINAL_WRAP_OPS + register_terminal_wrap_op(_NeverSeenWrap) + assert _NeverSeenWrap in _TERMINAL_WRAP_OPS + + +def test_register_terminal_wrap_op_is_idempotent(): + '''Re-registering the same op type is a no-op (set semantics). + External plugins may re-run `register(compiler)` across multiple + Compilers in the same process; the registry must tolerate that + without growing duplicate entries. + ''' + + @final + @dataclass(frozen=True, slots=True) + class _IdempotentWrap(Op): + pass + + register_terminal_wrap_op(_IdempotentWrap) + size_after_first = len(_TERMINAL_WRAP_OPS) + register_terminal_wrap_op(_IdempotentWrap) + register_terminal_wrap_op(_IdempotentWrap) + assert len(_TERMINAL_WRAP_OPS) == size_after_first + + +# ----------------------------------------------------------------------------- +# 2. Predicate behaviour — _supported_pipeline / _trailing_inserts / +# _is_terminal_op accept the mock wrap op +# ----------------------------------------------------------------------------- + + +def _make_scan() -> m.Scan: + return m.Scan( + vars=['v0', 'v1'], + rel_name='Src', + version=Version.FULL, + index=[0, 1], + handle_start=0, + ) + + +def _make_insert() -> m.InsertInto: + return m.InsertInto( + rel_name='Dst', + version=Version.NEW, + vars=['v0', 'v1'], + index=[0, 1], + ) + + +def test_is_terminal_op_matches_insert_into(): + assert _is_terminal_op(_make_insert()) is True + + +def test_is_terminal_op_matches_registered_wrap_op(): + _ensure_mock_registered() + wrap = MockExtensionWrap(inner=_make_insert()) + assert _is_terminal_op(wrap) is True + + +def test_is_terminal_op_rejects_unregistered_op(): + scan = _make_scan() + assert _is_terminal_op(scan) is False + + +def test_trailing_inserts_collects_mixed_terminal_run(): + '''The trailing-run walker collects bare InsertInto AND registered + wrap ops. Order is preserved. + ''' + _ensure_mock_registered() + insert = _make_insert() + wrap = MockExtensionWrap(inner=_make_insert()) + pipeline = [_make_scan(), insert, wrap] + trailing = _trailing_inserts(pipeline) + assert trailing == [insert, wrap] + + +def test_supported_pipeline_accepts_wrap_op_terminal(): + '''A pipeline shape `[Scan, MockExtensionWrap]` is structurally + legal once the mock wrap op is registered. Pre-fix, this returned + False because the trailing-tail predicate only matched + `mir.InsertInto`. + ''' + _ensure_mock_registered() + pipeline = [_make_scan(), MockExtensionWrap(inner=_make_insert())] + assert _supported_pipeline(pipeline) is True + + +def test_supported_pipeline_rejects_unregistered_wrap_at_tail(): + '''Sanity: an unregistered op type at the tail still fails the + predicate — the registry is the sole extension point. + ''' + + @final + @dataclass(frozen=True, slots=True) + class _UnregisteredTail(Op): + inner: m.InsertInto + + pipeline = [_make_scan(), _UnregisteredTail(inner=_make_insert())] + assert _supported_pipeline(pipeline) is False + + +# ----------------------------------------------------------------------------- +# 3. End-to-end dispatch via compile_pipeline +# ----------------------------------------------------------------------------- + + +def _ep_with_mock_tail() -> m.ExecutePipeline: + '''Build a minimal EP whose pipeline ends in a registered mock + wrap op. Mirrors the shape an external plugin would produce after + its pragma's materialize handler wraps the bare `InsertInto`. + ''' + scan = _make_scan() + insert = _make_insert() + wrap = MockExtensionWrap(inner=insert) + return m.ExecutePipeline( + pipeline=[scan, wrap], + source_specs=[scan], + dest_specs=[insert], + rule_name='MockRule', + ) + + +def test_compile_pipeline_dispatches_registered_wrap_op_at_tail(): + '''Integration: `compile_pipeline(ep)` lowers a pipeline ending in + a registered terminal wrap op end-to-end. This is the load-bearing + assertion from PR #68's contract-gap analysis — pre-fix, this call + raised `ValueError: lower_scan_pipeline: unsupported pipeline shape` + (or `unsupported inner op` from `_lower_inner_chain`). + + The emitted CUDA contains the InsertInto's emission (the mock + lowering delegates to `_lower_insert_into`). Exact byte content + isn't asserted — that belongs to per-pragma byte-equivalence + fixtures. + ''' + _ensure_mock_registered() + ep = _ep_with_mock_tail() + out = compile_pipeline(ep) + # The mock wrap delegates to the bare-InsertInto emission, so the + # rendered CUDA must contain the standard emit_direct call shape + # for the inner `InsertInto`. We don't pin the exact text — just + # that the dispatcher reached the inner emission. + assert 'emit_direct' in out or 'output' in out + # The wrap op type itself never appears in rendered CUDA (it is a + # MIR-layer artifact, lowered before render). Sanity guard. + assert 'MockExtensionWrap' not in out + + +def test_compile_pipeline_unregistered_wrap_still_rejected(): + '''The registry is the sole extension point: an EP whose tail is + an unregistered op type still fails loudly at lowering time, + matching the pre-fix behaviour. Guards against accidental over- + acceptance of arbitrary structural shapes. + ''' + + @final + @dataclass(frozen=True, slots=True) + class _NotRegisteredAtTail(Op): + inner: m.InsertInto + + scan = _make_scan() + insert = _make_insert() + ep = m.ExecutePipeline( + pipeline=[scan, _NotRegisteredAtTail(inner=insert)], + source_specs=[scan], + dest_specs=[insert], + rule_name='UnregRule', + ) + with pytest.raises(ValueError, match=r'unsupported pipeline shape'): + compile_pipeline(ep)