Skip to content

zorch.byte_transcript

Byte-oriented Fiat-Shamir transcript: the ByteTranscript seam and a Merlin-over-hash duplex (ByteHashTranscript) parameterized by a ByteHash.

This is the HOST-side, byte-oriented sibling of the device-resident algebraic DuplexTranscript (transcript.py). The two form a taxonomy (see docs/blocks/transcript.md): an algebraic sponge whose observe/sample are device ops fused into the round body, vs a byte hash whose Fiat-Shamir chain is strictly sequential and runs on the host.

ByteHashTranscript holds a bytes buffer and an injected ByteHash (hash/byte_hash.py); the digest substrate — host HostSha256 or the device Sha256 marker — is a value it carries, not a class it hardcodes. Its has_dedicated_fusion delegates to that hash, exactly as DuplexSponge delegates to its Permutation. So the same construction backs both a host byte challenger (inject HostSha256()) and the device-byte row of the taxonomy (inject Sha256()); the two are byte-identical.

The construction — op-tagged absorb, HASH(buffer || ctr) counter-squeeze (SHA-256 is not an XOF), and re-absorb of the squeezed bytes — is a standard Merlin-style duplex, byte-identical to the canonical SHA-256 Fiat-Shamir transcript used by binary-field provers (e.g. succinctlabs/flock's FsChallenger). zorch owns the wire framing on OPAQUE bytes; a consumer supplies only its field<->bytes serialization (see flock-zorch's challenger.py, an F128 (16-byte lo||hi) surface).

ByteTranscript

Bases: Protocol

Byte-oriented Fiat-Shamir seam. observe_* append tagged, length-prefixed bytes; sample_* squeeze raw bytes (the consumer reinterprets to field elements). Distinct from transcript.Transcript, which is field-element- and device-oriented.

Source code in zorch/byte_transcript.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class ByteTranscript(Protocol):
    """Byte-oriented Fiat-Shamir seam. `observe_*` append tagged, length-prefixed
    bytes; `sample_*` squeeze raw bytes (the consumer reinterprets to field
    elements). Distinct from `transcript.Transcript`, which is field-element- and
    device-oriented."""

    def observe_label(self, label: bytes) -> Self: ...
    def observe_bytes(self, data: bytes) -> Self: ...
    def observe_scalar(self, payload: bytes) -> Self: ...
    def observe_slice(self, payload: bytes, count: int) -> Self: ...
    def sample_scalar(self, nbytes: int) -> tuple[Self, bytes]: ...
    def sample_slice(self, count: int, width: int) -> tuple[Self, bytes]: ...
    def grind_pow(self, bits: int) -> tuple[Self, int]: ...
    def verify_pow(self, nonce: int, *, bits: int) -> tuple[Self, bool]: ...

ByteHashTranscript dataclass

Merlin-style byte duplex over an injected ByteHash. Functional: every op returns a new transcript whose buffer is the running absorbed-byte stream. A host object (a bytes buffer, not a jit-traced pytree); the ByteHash chooses the squeeze substrate — HostSha256 (host hashlib) or Sha256 (the hash_frx.sha256 device marker). Byte-identical whichever is injected.

Source code in zorch/byte_transcript.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
@dataclass(frozen=True)
class ByteHashTranscript:
    """Merlin-style byte duplex over an injected `ByteHash`. Functional: every op
    returns a new transcript whose `buffer` is the running absorbed-byte stream. A
    host object (a `bytes` buffer, not a jit-traced pytree); the `ByteHash` chooses
    the squeeze substrate — `HostSha256` (host `hashlib`) or `Sha256` (the
    `hash_frx.sha256` device marker). Byte-identical whichever is injected."""

    buffer: bytes
    byte_hash: ByteHash

    @property
    def has_dedicated_fusion(self) -> bool:
        # One-kernel-ness is a fact about the injected hash's backend routing,
        # which the transcript cannot know — the same deferral hash-frx's
        # `DuplexSponge` makes to its `Permutation`. Names no concrete hash.
        return self.byte_hash.fusion_path.is_one_kernel

    @classmethod
    def new(cls, domain: bytes, byte_hash: ByteHash) -> ByteHashTranscript:
        """Seed with a length-prefixed domain so prefix domains can't collide:
        `[OP_DOMAIN] || len8(domain) || domain`."""
        return cls(bytes([OP_DOMAIN]) + _len8(len(domain)) + bytes(domain), byte_hash)

    # ---- internal absorb / squeeze (over byte_hash.digest) ----
    def _absorb(self, payload: bytes) -> ByteHashTranscript:
        return ByteHashTranscript(self.buffer + payload, self.byte_hash)

    def _digest(self) -> bytes:
        """`HASH(buffer)` — the proof-of-work state digest (no counter, no tag)."""
        row = np.frombuffer(self.buffer, dtype=np.uint8)[None, :]  # [1, len(buffer)]
        return bytes(np.asarray(self.byte_hash.digest(row))[0])

    def _squeeze(self, n: int) -> bytes:
        """`n` bytes as `HASH(buffer || ctr_le8)` for ctr=0,1,… (digest_size B per
        block, a hash is not an XOF). The counter blocks share a length, so the
        whole squeeze is ONE batched `byte_hash.digest` call."""
        if n <= 0:
            return b""
        d = self.byte_hash.digest_size
        nblocks = (n + d - 1) // d
        batch = np.stack(
            [
                np.frombuffer(self.buffer + _len8(ctr), dtype=np.uint8)
                for ctr in range(nblocks)
            ]
        )  # [nblocks, len(buffer)+8]
        digs = np.asarray(self.byte_hash.digest(batch))  # [nblocks, digest_size]
        return digs.reshape(-1).tobytes()[:n]

    # ---- observe ----
    def observe_label(self, label: bytes) -> ByteHashTranscript:
        return self._absorb(bytes([OP_LABEL]) + _len8(len(label)) + bytes(label))

    def observe_bytes(self, data: bytes) -> ByteHashTranscript:
        return self._absorb(bytes([OP_BYTES]) + _len8(len(data)) + bytes(data))

    def observe_scalar(self, payload: bytes) -> ByteHashTranscript:
        # No length prefix — a scalar's width is implicit in the consumer.
        return self._absorb(bytes([OP_OBSERVE, KIND_SCALAR]) + bytes(payload))

    def observe_slice(self, payload: bytes, count: int) -> ByteHashTranscript:
        return self._absorb(
            bytes([OP_OBSERVE, KIND_SLICE]) + _len8(count) + bytes(payload)
        )

    # ---- sample (absorb tag, squeeze without mutating, re-absorb the squeeze) ----
    def sample_scalar(self, nbytes: int) -> tuple[ByteHashTranscript, bytes]:
        if nbytes < 0:
            raise ValueError(f"nbytes must be non-negative, got {nbytes}")
        t = self._absorb(bytes([OP_SQUEEZE, KIND_SCALAR]))
        buf = t._squeeze(nbytes)
        return t._absorb(buf), buf

    def sample_slice(self, count: int, width: int) -> tuple[ByteHashTranscript, bytes]:
        if count < 0 or width < 0:
            raise ValueError(f"count/width must be non-negative, got {count}/{width}")
        t = self._absorb(bytes([OP_SQUEEZE, KIND_SLICE]) + _len8(count))
        buf = t._squeeze(count * width)
        return t._absorb(buf), buf

    # ---- proof-of-work ----
    def _grind(self, state_digest: bytes, bits: int) -> int:
        """Lowest u64 nonce with `HASH(state_digest || nonce_le8)` having `bits`
        leading zero bits. Tests a window of nonces per `digest` call (window 1 for
        a host hash = sequential early-exit); tiles windows until a hit — unbounded,
        never returns an unchecked nonce."""
        window = _GRIND_WINDOW if self.byte_hash.fusion_path.is_one_kernel else 1
        base = 0
        while True:
            batch = np.stack(
                [
                    np.frombuffer(state_digest + _len8(base + i), dtype=np.uint8)
                    for i in range(window)
                ]
            )  # [window, len(state_digest)+8]
            hits = np.flatnonzero(
                _leading_zero_bits_ok(np.asarray(self.byte_hash.digest(batch)), bits)
            )
            if hits.size:
                return base + int(hits[0])
            base += window

    def grind_pow(self, bits: int) -> tuple[ByteHashTranscript, int]:
        """Lowest u64 nonce whose PoW passes (0 if bits==0), then absorb it via
        `observe_bytes` so subsequent challenges bind to it."""
        _validate_pow_bits(bits, self.byte_hash.digest_size)
        nonce = 0 if bits == 0 else self._grind(self._digest(), bits)
        return self.observe_bytes(_len8(nonce)), nonce

    def verify_pow(self, nonce: int, *, bits: int) -> tuple[ByteHashTranscript, bool]:
        """Verifier mirror: check the PoW (bits==0 requires the canonical nonce 0),
        then absorb the nonce REGARDLESS so the transcript stays in lockstep."""
        _validate_pow_bits(bits, self.byte_hash.digest_size)
        if bits == 0:
            ok = nonce == 0
        else:
            row = np.frombuffer(self._digest() + _len8(nonce), dtype=np.uint8)[None, :]
            ok = bool(
                _leading_zero_bits_ok(np.asarray(self.byte_hash.digest(row)), bits)[0]
            )
        return self.observe_bytes(_len8(nonce)), ok

new classmethod 

new(
    domain: bytes, byte_hash: ByteHash
) -> ByteHashTranscript

Seed with a length-prefixed domain so prefix domains can't collide: [OP_DOMAIN] || len8(domain) || domain.

Source code in zorch/byte_transcript.py 
110
111
112
113
114
@classmethod
def new(cls, domain: bytes, byte_hash: ByteHash) -> ByteHashTranscript:
    """Seed with a length-prefixed domain so prefix domains can't collide:
    `[OP_DOMAIN] || len8(domain) || domain`."""
    return cls(bytes([OP_DOMAIN]) + _len8(len(domain)) + bytes(domain), byte_hash)

grind_pow 

grind_pow(bits: int) -> tuple[ByteHashTranscript, int]

Lowest u64 nonce whose PoW passes (0 if bits==0), then absorb it via observe_bytes so subsequent challenges bind to it.

Source code in zorch/byte_transcript.py 
195
196
197
198
199
200
def grind_pow(self, bits: int) -> tuple[ByteHashTranscript, int]:
    """Lowest u64 nonce whose PoW passes (0 if bits==0), then absorb it via
    `observe_bytes` so subsequent challenges bind to it."""
    _validate_pow_bits(bits, self.byte_hash.digest_size)
    nonce = 0 if bits == 0 else self._grind(self._digest(), bits)
    return self.observe_bytes(_len8(nonce)), nonce

verify_pow 

verify_pow(
    nonce: int, *, bits: int
) -> tuple[ByteHashTranscript, bool]

Verifier mirror: check the PoW (bits==0 requires the canonical nonce 0), then absorb the nonce REGARDLESS so the transcript stays in lockstep.

Source code in zorch/byte_transcript.py 
202
203
204
205
206
207
208
209
210
211
212
213
def verify_pow(self, nonce: int, *, bits: int) -> tuple[ByteHashTranscript, bool]:
    """Verifier mirror: check the PoW (bits==0 requires the canonical nonce 0),
    then absorb the nonce REGARDLESS so the transcript stays in lockstep."""
    _validate_pow_bits(bits, self.byte_hash.digest_size)
    if bits == 0:
        ok = nonce == 0
    else:
        row = np.frombuffer(self._digest() + _len8(nonce), dtype=np.uint8)[None, :]
        ok = bool(
            _leading_zero_bits_ok(np.asarray(self.byte_hash.digest(row)), bits)[0]
        )
    return self.observe_bytes(_len8(nonce)), ok