zorch.transcript

Fiat-Shamir transcript: the Transcript interface and a real duplex-sponge implementation.

DuplexTranscript is the device-side duplex sponge (fixed-size buffers + position scalars) — a JAX pytree whose state threads functionally under @jit, with no host callback or zkVM FFI.

GrindError

Bases: RuntimeError

Raised when a proof-of-work grind cannot run: the field is too wide for the uint32 search (needs x64).

Source code in zorch/transcript.py

class GrindError(RuntimeError):
    """Raised when a proof-of-work grind cannot run: the field is too wide for
    the uint32 search (needs x64)."""

Transcript

Bases: Protocol

Source code in zorch/transcript.py

class Transcript(Protocol):
    @property
    def field(self) -> Any:
        """The field one `sample` word is drawn from.

        A challenge in another field is packed from consecutive words of this
        one, so how many words a challenge costs is a fact about the pair, not
        about the challenge field alone -- an extension-native sponge already
        yields an extension element per word.
        """

    @property
    def has_dedicated_fusion(self) -> bool: ...
    def observe(self, values: Array) -> Self: ...
    def sample(self, n: int = 1) -> tuple[Self, Array]: ...
    def observe_and_sample(self, values: Array, n: int = 1) -> tuple[Self, Array]: ...
    def grind(self, pow_bits: int) -> tuple[Self, Array]: ...
    def check_witness(self, witness: Array, *, pow_bits: int) -> tuple[Self, Array]: ...

field property

field: Any

The field one sample word is drawn from.

A challenge in another field is packed from consecutive words of this one, so how many words a challenge costs is a fact about the pair, not about the challenge field alone -- an extension-native sponge already yields an extension element per word.

DuplexState dataclass

Duplex-sponge state. Fixed-size buffers + position scalars: the buffers keep observe's absorb a single lax.scan (compile size independent of input length), and the constant shape makes the whole state a valid lax.scan carry.

Source code in zorch/transcript.py

@register_dataclass
@dataclass(frozen=True)
class DuplexState:
    """Duplex-sponge state. Fixed-size buffers + position scalars: the buffers
    keep `observe`'s absorb a single `lax.scan` (compile size independent of input
    length), and the constant shape makes the whole state a valid `lax.scan` carry."""

    input_buffer: Array  # (rate,) — valid prefix is [0:in_pos]
    output_buffer: Array  # (rate,) — valid prefix is [0:out_pos]
    sponge_state: Array  # (width,)
    in_pos: Array  # int32 0-D, 0 <= in_pos < rate
    out_pos: Array  # int32 0-D, 0 <= out_pos <= rate

DuplexTranscript dataclass

Overwrite-mode duplex sponge implementing Transcript. A JAX pytree whose state buffers are the leaves and whose permutation/rate are static, so the whole transcript threads through @jit (and, later, a lax.scan carry). No step crosses a zkVM FFI on any backend.

The fs backend (_DeviceFs / _HostFs) chooses where every absorb / squeeze runs; pick one via new(..., fs_on_host=). Under _DeviceFs (the default) every step is a device op and nothing calls back to the host. _HostFs is an eager host primitive -- see the host-FS backend section below for what it trades and why.

Source code in zorch/transcript.py

@partial(
    register_dataclass,
    data_fields=["state"],
    meta_fields=["permutation", "rate", "fs"],
)
@dataclass(frozen=True)
class DuplexTranscript:
    """Overwrite-mode duplex sponge implementing `Transcript`. A JAX pytree whose
    `state` buffers are the leaves and whose `permutation`/`rate` are static, so
    the whole transcript threads through `@jit` (and, later, a `lax.scan` carry).
    No step crosses a zkVM FFI on any backend.

    The `fs` backend (`_DeviceFs` / `_HostFs`) chooses where every absorb / squeeze
    runs; pick one via `new(..., fs_on_host=)`. Under `_DeviceFs` (the default)
    every step is a device op and nothing calls back to the host. `_HostFs` is an
    eager host primitive -- see the host-FS backend section below for what it
    trades and why."""

    permutation: Permutation
    rate: int
    state: DuplexState
    fs: _FsBackend = _DEVICE_FS

    @property
    def fs_on_host(self) -> bool:
        """Whether Fiat-Shamir runs on the host CPU — read off the `fs` backend.
        Kept as a bool for callers (the `sumcheck.prover` gate, the consumer's
        `.new(fs_on_host=...)`)."""
        return self.fs.on_host

    @property
    def has_dedicated_fusion(self) -> bool:
        """Whether the Fiat-Shamir permutation lowers to a dedicated fusion marker
        a vendor can expand — the LogUp-GKR jagged prover's gate
        (`zorch.logup_gkr.jagged_prover`) reads it to mark its sumcheck scan as one
        register-resident kernel (mirrors `Sponge`/`Compression`). False for a test
        `CheapPermutation`, so unit tests keep the plain scan."""
        return self.permutation.fusion_path.is_one_kernel

    @classmethod
    def new(
        cls, permutation: Permutation, rate: int, fs_on_host: bool = False
    ) -> DuplexTranscript:
        if not 1 <= rate < permutation.width:
            raise ValueError(
                f"rate ({rate}) must satisfy 1 <= rate < width ({permutation.width})"
            )
        dtype: Any = permutation.dtype
        state = DuplexState(
            input_buffer=fnp.zeros(rate, dtype=dtype),
            output_buffer=fnp.zeros(rate, dtype=dtype),
            sponge_state=fnp.zeros(permutation.width, dtype=dtype),
            in_pos=fnp.int32(0),
            out_pos=fnp.int32(0),
        )
        return cls(permutation, rate, state, _HOST_FS if fs_on_host else _DEVICE_FS)

    def _with_state(self, state: DuplexState) -> DuplexTranscript:
        return DuplexTranscript(self.permutation, self.rate, state, self.fs)

    def _duplexing(self) -> DuplexTranscript:
        """Flush the pending input prefix and refill the output buffer."""
        st = self.state
        sponge = _absorb_permute(
            self.permutation, st.sponge_state, st.input_buffer, st.in_pos, self.rate
        )
        return self._with_state(
            DuplexState(
                input_buffer=fnp.zeros(self.rate, dtype=sponge.dtype),
                output_buffer=sponge[: self.rate],
                sponge_state=sponge,
                in_pos=fnp.int32(0),
                out_pos=fnp.int32(self.rate),
            )
        )

    def observe(self, values: Array) -> DuplexTranscript:
        """Absorb `values` (any field, flattened to the base field) into the
        transcript. The absorb is one `lax.scan` over the flat input, so the
        compiled graph size is independent of `len(values)`."""
        return self.fs.observe(self, values)

    def absorb_on_host(self, *messages: Array) -> DuplexTranscript:
        """Absorb `messages` in order through the CPU sponge, leaving the state
        on the device it came from. Byte-identical to the same sequence of
        `observe` calls -- the same `_observe_body`, only relocated.

        The win is placement, not algorithm. A sponge on an accelerator runs one
        warp-cooperative permute per rate-block, where ~21 rounds of full-warp
        shuffle latency dominate a few hundred field ops; a CPU pays none of that.
        Measured on koalabear16 at rate 8: 5.4 us per permutation on an RTX 5090
        against 1.0 us on the host. Two host crossings are not worth a short
        message, so the caller applies its own length threshold -- the call site
        is where the message length that justifies relocating is known.

        Variadic because the crossing, not the absorb, is the repeated cost: a
        Fiat-Shamir step is usually several messages in a fixed order (a length
        prefix, then its payload), and absorbing them one call at a time drags the
        state back to the device between each. The whole sequence costs one round
        trip.

        Only THIS absorb relocates; `fs` still decides where the stream runs.

        Eager only -- it moves buffers across the host boundary, so it cannot
        run inside a traced region; `observe` is the in-graph form.
        """
        if not messages:
            return self
        if self.fs.on_host:
            # The stream already runs there; relocating is the backend's job.
            t = self
            for m in messages:
                t = t.observe(m)
            return t
        if isinstance(self.state.sponge_state, frx.core.Tracer) or any(
            isinstance(m, frx.core.Tracer) for m in messages
        ):
            raise ValueError(
                "absorb_on_host is eager (it moves the sponge state across the "
                "host boundary); call observe() inside a traced region"
            )
        device = next(iter(self.state.sponge_state.devices()))
        t = self
        for m in messages:
            # `_observe_host` commits the state to the CPU on the first hop and
            # returns host leaves, so the rest of the sequence finds it resident.
            t = _observe_host(t, m)
        return t._with_state(_state_on_device(t.state, device))

    def _sample_one(self) -> tuple[DuplexTranscript, Array]:
        # Permute when input is pending or the output buffer is drained.
        need_perm = (self.state.in_pos > 0) | (self.state.out_pos == 0)
        # `select`, not `lax.cond`: a traced-predicate `cond` reads `need_perm`
        # back to the host to choose a branch -- one device->host sync per
        # sample. The two branches are shape-equal, so selecting the
        # unconditionally-computed `_duplexing()` is byte-identical to the cond;
        # the only cost is running the permute on the no-perm path too, a net win
        # because it removes the host round-trip.
        permuted = self._duplexing()
        t = self._with_state(
            tree_map(
                lambda p, c: fnp.where(need_perm, p, c), permuted.state, self.state
            )
        )
        out_pos = t.state.out_pos - 1
        item = t.state.output_buffer[out_pos]
        return t._with_state(replace(t.state, out_pos=out_pos)), item

    def sample(self, n: int = 1) -> tuple[DuplexTranscript, Array]:
        return self.fs.sample(self, n)

    @property
    def field(self) -> Any:
        return self.state.sponge_state.dtype

    def observe_and_sample(
        self, values: Array, n: int = 1
    ) -> tuple[DuplexTranscript, Array]:
        """Absorb `values`, then squeeze `n` challenges — the per-round
        Fiat-Shamir primitive (commit -> challenge). One method so the absorb and
        squeeze fuse into a single kernel under `@jit` by construction, never by a
        per-primitive pattern-match (the repo's fusion contract)."""
        return self.fs.observe_and_sample(self, values, n)

    def check_witness(
        self, witness: Array, *, pow_bits: int
    ) -> tuple[DuplexTranscript, Array]:
        """Observe `witness`, squeeze one challenge, and report whether its low
        `pow_bits` canonical bits are zero -- the verifier-side proof-of-work
        check, and the predicate `grind` searches against. `witness` must be a
        scalar element of the transcript's field -- the domain `grind`
        enumerates -- so the verifier accepts exactly the witness space the
        prover searched (`observe` itself would bitcast-flatten any array).
        Fully jit-traceable, so a verifier runs it inside its own `@jit` zone.
        Returns the advanced transcript (observe + one sample applied), so prover
        and verifier reach the same state from the same witness."""
        _validate_pow_bits(pow_bits)
        field_dtype = self.state.sponge_state.dtype
        _require_uint32_field(field_dtype)
        if witness.shape != () or witness.dtype != field_dtype:
            raise ValueError(
                f"witness must be a scalar {field_dtype} field element (the grind "
                f"search's domain), got shape {witness.shape} dtype {witness.dtype}"
            )
        return self.fs.check_witness(self, witness, pow_bits=pow_bits)

    @partial(jit, static_argnames=("pow_bits", "chunk"))
    def _grind_search(self, pow_bits: int, chunk: int) -> Array:
        """Search canonical witnesses `0, 1, 2, ...` for the lowest one whose
        challenge has `pow_bits` zero low bits — `grind.grind_search` over the
        challenge predicate (`vmap` over the window, not a sequential
        `lax.map`). Returns the winning witness (or the trailing fallback on
        exhaustion -- `grind` re-checks it before returning). Fields wider than
        32 bits raise (the uint32 counter/bit-check would need x64);
        koalabear-class fields are searched in full (`bound` = the field
        order)."""
        field_dtype = self.state.sponge_state.dtype
        modulus = _require_uint32_field(field_dtype)

        def satisfies(witness: Array) -> Array:
            _, sample = self.observe(witness).sample(1)
            return _pow_satisfied(sample[0], pow_bits)

        def check_batch(counters: Array) -> Array:
            return vmap(satisfies)(counters.astype(field_dtype))

        return grind_search(check_batch, modulus, chunk).astype(field_dtype)

    def grind(
        self, pow_bits: int, *, chunk: int = _GRIND_CHUNK
    ) -> tuple[DuplexTranscript, Array]:
        """Find a proof-of-work witness and return the transcript advanced past
        it via `check_witness`, so a verifier replaying it reaches the same state.
        Searches canonical witnesses for the lowest whose squeezed challenge has
        `pow_bits` zero low bits. Jit-traceable and does not raise on an exhausted
        search: `check_witness` is the soundness gate, so which witness the search
        returns is soundness-neutral."""
        _validate_pow_bits(pow_bits)
        if chunk < 1:
            raise ValueError(f"chunk must be >= 1, got {chunk}")
        field_dtype = self.state.sponge_state.dtype
        if pow_bits == 0:
            # No work required: the canonical zero witness always passes.
            witness = fnp.zeros((), field_dtype)
            return self.check_witness(witness, pow_bits=pow_bits)[0], witness
        witness = self._grind_search(pow_bits, chunk)
        advanced, _ = self.check_witness(witness, pow_bits=pow_bits)
        return advanced, witness

fs_on_host property

fs_on_host: bool

Whether Fiat-Shamir runs on the host CPU — read off the fs backend. Kept as a bool for callers (the sumcheck.prover gate, the consumer's .new(fs_on_host=...)).

has_dedicated_fusion property

has_dedicated_fusion: bool

Whether the Fiat-Shamir permutation lowers to a dedicated fusion marker a vendor can expand — the LogUp-GKR jagged prover's gate (zorch.logup_gkr.jagged_prover) reads it to mark its sumcheck scan as one register-resident kernel (mirrors Sponge/Compression). False for a test CheapPermutation, so unit tests keep the plain scan.

observe

observe(values: Array) -> DuplexTranscript

Absorb values (any field, flattened to the base field) into the transcript. The absorb is one lax.scan over the flat input, so the compiled graph size is independent of len(values).

Source code in zorch/transcript.py

def observe(self, values: Array) -> DuplexTranscript:
    """Absorb `values` (any field, flattened to the base field) into the
    transcript. The absorb is one `lax.scan` over the flat input, so the
    compiled graph size is independent of `len(values)`."""
    return self.fs.observe(self, values)

absorb_on_host

absorb_on_host(*messages: Array) -> DuplexTranscript

Absorb messages in order through the CPU sponge, leaving the state on the device it came from. Byte-identical to the same sequence of observe calls -- the same _observe_body, only relocated.

The win is placement, not algorithm. A sponge on an accelerator runs one warp-cooperative permute per rate-block, where ~21 rounds of full-warp shuffle latency dominate a few hundred field ops; a CPU pays none of that. Measured on koalabear16 at rate 8: 5.4 us per permutation on an RTX 5090 against 1.0 us on the host. Two host crossings are not worth a short message, so the caller applies its own length threshold -- the call site is where the message length that justifies relocating is known.

Variadic because the crossing, not the absorb, is the repeated cost: a Fiat-Shamir step is usually several messages in a fixed order (a length prefix, then its payload), and absorbing them one call at a time drags the state back to the device between each. The whole sequence costs one round trip.

Only THIS absorb relocates; fs still decides where the stream runs.

Eager only -- it moves buffers across the host boundary, so it cannot run inside a traced region; observe is the in-graph form.

Source code in zorch/transcript.py

def absorb_on_host(self, *messages: Array) -> DuplexTranscript:
    """Absorb `messages` in order through the CPU sponge, leaving the state
    on the device it came from. Byte-identical to the same sequence of
    `observe` calls -- the same `_observe_body`, only relocated.

    The win is placement, not algorithm. A sponge on an accelerator runs one
    warp-cooperative permute per rate-block, where ~21 rounds of full-warp
    shuffle latency dominate a few hundred field ops; a CPU pays none of that.
    Measured on koalabear16 at rate 8: 5.4 us per permutation on an RTX 5090
    against 1.0 us on the host. Two host crossings are not worth a short
    message, so the caller applies its own length threshold -- the call site
    is where the message length that justifies relocating is known.

    Variadic because the crossing, not the absorb, is the repeated cost: a
    Fiat-Shamir step is usually several messages in a fixed order (a length
    prefix, then its payload), and absorbing them one call at a time drags the
    state back to the device between each. The whole sequence costs one round
    trip.

    Only THIS absorb relocates; `fs` still decides where the stream runs.

    Eager only -- it moves buffers across the host boundary, so it cannot
    run inside a traced region; `observe` is the in-graph form.
    """
    if not messages:
        return self
    if self.fs.on_host:
        # The stream already runs there; relocating is the backend's job.
        t = self
        for m in messages:
            t = t.observe(m)
        return t
    if isinstance(self.state.sponge_state, frx.core.Tracer) or any(
        isinstance(m, frx.core.Tracer) for m in messages
    ):
        raise ValueError(
            "absorb_on_host is eager (it moves the sponge state across the "
            "host boundary); call observe() inside a traced region"
        )
    device = next(iter(self.state.sponge_state.devices()))
    t = self
    for m in messages:
        # `_observe_host` commits the state to the CPU on the first hop and
        # returns host leaves, so the rest of the sequence finds it resident.
        t = _observe_host(t, m)
    return t._with_state(_state_on_device(t.state, device))

observe_and_sample

observe_and_sample(
    values: Array, n: int = 1
) -> tuple[DuplexTranscript, Array]

Absorb values, then squeeze n challenges — the per-round Fiat-Shamir primitive (commit -> challenge). One method so the absorb and squeeze fuse into a single kernel under @jit by construction, never by a per-primitive pattern-match (the repo's fusion contract).

Source code in zorch/transcript.py

def observe_and_sample(
    self, values: Array, n: int = 1
) -> tuple[DuplexTranscript, Array]:
    """Absorb `values`, then squeeze `n` challenges — the per-round
    Fiat-Shamir primitive (commit -> challenge). One method so the absorb and
    squeeze fuse into a single kernel under `@jit` by construction, never by a
    per-primitive pattern-match (the repo's fusion contract)."""
    return self.fs.observe_and_sample(self, values, n)

check_witness

check_witness(
    witness: Array, *, pow_bits: int
) -> tuple[DuplexTranscript, Array]

Observe witness, squeeze one challenge, and report whether its low pow_bits canonical bits are zero -- the verifier-side proof-of-work check, and the predicate grind searches against. witness must be a scalar element of the transcript's field -- the domain grind enumerates -- so the verifier accepts exactly the witness space the prover searched (observe itself would bitcast-flatten any array). Fully jit-traceable, so a verifier runs it inside its own @jit zone. Returns the advanced transcript (observe + one sample applied), so prover and verifier reach the same state from the same witness.

Source code in zorch/transcript.py

def check_witness(
    self, witness: Array, *, pow_bits: int
) -> tuple[DuplexTranscript, Array]:
    """Observe `witness`, squeeze one challenge, and report whether its low
    `pow_bits` canonical bits are zero -- the verifier-side proof-of-work
    check, and the predicate `grind` searches against. `witness` must be a
    scalar element of the transcript's field -- the domain `grind`
    enumerates -- so the verifier accepts exactly the witness space the
    prover searched (`observe` itself would bitcast-flatten any array).
    Fully jit-traceable, so a verifier runs it inside its own `@jit` zone.
    Returns the advanced transcript (observe + one sample applied), so prover
    and verifier reach the same state from the same witness."""
    _validate_pow_bits(pow_bits)
    field_dtype = self.state.sponge_state.dtype
    _require_uint32_field(field_dtype)
    if witness.shape != () or witness.dtype != field_dtype:
        raise ValueError(
            f"witness must be a scalar {field_dtype} field element (the grind "
            f"search's domain), got shape {witness.shape} dtype {witness.dtype}"
        )
    return self.fs.check_witness(self, witness, pow_bits=pow_bits)

grind

grind(
    pow_bits: int, *, chunk: int = _GRIND_CHUNK
) -> tuple[DuplexTranscript, Array]

Find a proof-of-work witness and return the transcript advanced past it via check_witness, so a verifier replaying it reaches the same state. Searches canonical witnesses for the lowest whose squeezed challenge has pow_bits zero low bits. Jit-traceable and does not raise on an exhausted search: check_witness is the soundness gate, so which witness the search returns is soundness-neutral.

Source code in zorch/transcript.py

def grind(
    self, pow_bits: int, *, chunk: int = _GRIND_CHUNK
) -> tuple[DuplexTranscript, Array]:
    """Find a proof-of-work witness and return the transcript advanced past
    it via `check_witness`, so a verifier replaying it reaches the same state.
    Searches canonical witnesses for the lowest whose squeezed challenge has
    `pow_bits` zero low bits. Jit-traceable and does not raise on an exhausted
    search: `check_witness` is the soundness gate, so which witness the search
    returns is soundness-neutral."""
    _validate_pow_bits(pow_bits)
    if chunk < 1:
        raise ValueError(f"chunk must be >= 1, got {chunk}")
    field_dtype = self.state.sponge_state.dtype
    if pow_bits == 0:
        # No work required: the canonical zero witness always passes.
        witness = fnp.zeros((), field_dtype)
        return self.check_witness(witness, pow_bits=pow_bits)[0], witness
    witness = self._grind_search(pow_bits, chunk)
    advanced, _ = self.check_witness(witness, pow_bits=pow_bits)
    return advanced, witness

reinterpret_challenge

reinterpret_challenge(raw: Array, dtype: Any) -> Array

Reinterpret consecutive transcript squeezes raw as one dtype challenge: the identity when dtype is the transcript's own field, else the extension element whose coefficients are the squeezes. The single definition of the limbs/dtype packing -- shared by sample_challenge and the sumcheck scan driver so a prover and its verifier dual cannot drift.

Fails loud on a packing mismatch: the squeezes are already consumed, so silently truncating to the first element would leave the stream advanced past a challenge nobody received. The check runs at trace time (shapes are static), so jitting does not cost the loud failure.

Jitted because these three ops are called eagerly once per FS squeeze, and unjitted each one launches its own kernel -- loose single-op dispatch that dominates a warm prove.

Source code in zorch/transcript.py

@partial(jit, static_argnums=(1,))
def reinterpret_challenge(raw: Array, dtype: Any) -> Array:
    """Reinterpret consecutive transcript squeezes `raw` as one `dtype` challenge:
    the identity when `dtype` is the transcript's own field, else the extension
    element whose coefficients are the squeezes. The single definition of the
    limbs/dtype packing -- shared by `sample_challenge` and the sumcheck scan
    driver so a prover and its verifier dual cannot drift.

    Fails loud on a packing mismatch: the squeezes are already consumed, so
    silently truncating to the first element would leave the stream advanced past
    a challenge nobody received. The check runs at trace time (shapes are static),
    so jitting does not cost the loud failure.

    Jitted because these three ops are called eagerly once per FS squeeze, and
    unjitted each one launches its own kernel -- loose single-op dispatch that
    dominates a warm prove."""
    viewed = raw.view(dtype)
    if viewed.shape != (1,):
        raise ValueError(
            f"{raw.shape[0]} squeezes reinterpret to {viewed.shape} elements of "
            f"{dtype}; a challenge needs exactly one"
        )
    return viewed[0]

sample_challenge

sample_challenge(
    transcript: TranscriptT, dtype: Any, limbs: int = 1
) -> tuple[TranscriptT, Array]

Squeeze one challenge of dtype as limbs transcript samples.

A transcript squeezes elements of its own field; a challenge field that extends it takes limbs consecutive squeezes reinterpreted as the extension element's coefficients (limbs == 1 with the transcript's own field is the identity reinterpret, via reinterpret_challenge). Module-level so a prover, its verifier dual, and any binding glue derive challenges from one definition -- a drift would desynchronize their Fiat-Shamir streams.

Source code in zorch/transcript.py

def sample_challenge(
    transcript: TranscriptT, dtype: Any, limbs: int = 1
) -> tuple[TranscriptT, Array]:
    """Squeeze one challenge of `dtype` as `limbs` transcript samples.

    A transcript squeezes elements of its own field; a challenge field that
    extends it takes `limbs` consecutive squeezes reinterpreted as the
    extension element's coefficients (`limbs == 1` with the transcript's own
    field is the identity reinterpret, via `reinterpret_challenge`). Module-level
    so a prover, its verifier dual, and any binding glue derive challenges from
    one definition -- a drift would desynchronize their Fiat-Shamir streams.
    """
    if limbs < 1:
        raise ValueError(f"limbs must be >= 1, got {limbs}")
    transcript, raw = transcript.sample(limbs)
    return transcript, reinterpret_challenge(raw, dtype)

observe_and_sample_marked

observe_and_sample_marked(
    t: DuplexTranscript, values: Array, n: int
) -> tuple[DuplexTranscript, Array]

observe_and_sample under a zorch.duplex_fs fusion marker so a vendor fuses the ~9-kernel hop (two permutes + duplex glue) into one register-resident kernel. Only a dedicated-fusion permutation is marked, so a test CheapPermutation keeps the plain path.

Source code in zorch/transcript.py

def observe_and_sample_marked(
    t: DuplexTranscript, values: Array, n: int
) -> tuple[DuplexTranscript, Array]:
    """`observe_and_sample` under a `zorch.duplex_fs` fusion marker so a vendor
    fuses the ~9-kernel hop (two permutes + duplex glue) into one register-resident
    kernel. Only a dedicated-fusion permutation is marked, so a test
    `CheapPermutation` keeps the plain path."""
    if not t.has_dedicated_fusion:
        return _observe_and_sample_body(t, values, n)
    return _duplex_fs_zone(t, values, n)