Fractal Recursive Codec
Compressed Matryoshka Encoding
(Python)
Recursive self-encoding, parity-guided discovery, and deeper generative reconstruction
Conceptual Frame
A normal archive usually behaves like a sealed box: bytes go in, bytes come out. This codec explores a different metaphor. The encoded form behaves more like a seed, a rule set, and a set of constraints. Decoding does not merely unpack a finished object; it unfolds a layered structure where each reconstructed layer can help reveal the next one.
Purpose
This script demonstrates a recursive approach to encoding and reconstruction. It combines a conventional compressed container with two reconstruction paths: one path reconstructs original text exactly, while the other unfolds deeper generated structures from a compact seed, recursive rule, and parity-like constraints.
Main idea: information can be organized as a layered generative structure. The first decoded layer is not only output; it becomes context for the next decoding step.
Core Thesis
The deeper thesis is that encoding does not have to remain flat. A compact representation can describe a recursive process. If the data has self-similarity, procedural regularity, or discoverable structure, then a small upper layer may generate or constrain much larger lower layers.
In this model, the encoded bytes are not only a compressed copy of an object. They are closer to a recipe: seed, rule, checkpoints, parity constraints, and a decoder that spends compute to unfold the next level.
Exact Reconstruction
The script stores residual information so the original UTF-8 text can be reconstructed exactly. This branch proves deterministic round-trip behavior.
Generative Reconstruction
The script stores a seed and constraints. The decoder discovers deeper blocks that satisfy those constraints, then uses the discovered blocks as parents for further layers.
How the Script Works
- Compressed container: payloads are serialized as deterministic JSON, compressed with zlib, and stored as base64.
- Exact residual mode: text bytes are recursively split into coarse samples and detail residuals. Reversing the layers reconstructs the original text exactly.
- Matryoshka mode: a seed expands into child layers. Each layer doubles the byte count and becomes the parent for the next layer.
- Hidden per-level nonce: the encoder uses a nonce to generate each constrained layer but does not store the nonce directly.
- Discovery search: the decoder tries possible nonces until a generated candidate matches the stored parity/checkpoint signature.
- Autonomous continuation: after stored constraints end, the decoder can keep generating deeper layers from the structure it has already reconstructed.
Parity-Guided Discovery
Parity is used here as more than a repair mechanism. It acts as a mathematical clue. The encoder stores compact signatures for layers and bands of layers. During decoding, those signatures guide a search process: the decoder generates candidates, checks them against the signature, and accepts the candidate that satisfies the constraint.
Simple explanation: imagine each layer as a nested doll. The parity signature is not the doll itself; it is a clue for finding the correct key. Once the decoder finds the key for one doll, the revealed doll becomes the starting point for the next search.
The cross-level parity bands are inspired by RAID-style thinking. Instead of only checking isolated blocks, the script verifies stripes of reconstructed levels. That creates a scaffold where several layers must agree with one another.
Compression Model
The script uses zlib for practical compression of the encoded payload. The more experimental compression idea is deeper: if a recursive rule plus compact constraints can describe a large amount of structured output, the stored representation can become much smaller than the generated structure it unfolds.
Demonstration result: the included demo unfolds a 3-byte seed into 768 generated bytes at depth 8. The output is generated structure constrained by the encoded rule and parity targets, not a flat copy stored byte-for-byte.
Why This Is Inspirational
The script points toward a possible class of future codecs where the stored data contains not only content, but also procedures for discovering content. Such a codec would be most powerful when the source has recursive patterns, self-similarity, symmetry, repeated motifs, or generative structure.
In that future direction, compression would not only ask, "How can we store these bytes with fewer bytes?" It would also ask, "What rule, seed, and constraints can regenerate the meaningful structure behind these bytes?"
Important Boundaries
Exact recovery of arbitrary unknown data still requires that the necessary information is represented somewhere. The exact branch stores residuals for this reason. The generative branch can produce much more output than the seed size, but those deeper bytes are model-consistent generated structure. This boundary matters because it keeps the idea technically honest while preserving the creative direction.
Best interpretation: the script is a working model of recursive generative encoding: seed, rule, parity constraints, search, reconstruction, and deeper continuation.
Usage
Run all tests:
python3 fractal_recursive_codec_v3.py testRun the combined exact and generative demonstration:
python3 fractal_recursive_codec_v3.py demoEncode and decode exact text:
python3 fractal_recursive_codec_v3.py encode-exact --text "Fractal exact reconstruction works." --out exact.json
python3 fractal_recursive_codec_v3.py decode-exact --in exact.jsonEncode a seed with constrained depth, then decode deeper than the stored constraint depth:
python3 fractal_recursive_codec_v3.py encode-generative --text "ZKM" --depth 5 --out generated.json
python3 fractal_recursive_codec_v3.py decode-generative --in generated.json --depth 7Verification
The script includes internal checks for exact text reconstruction, compressed container integrity, parity-guided layer discovery, cross-level band verification, and autonomous continuation beyond the stored constraint depth.
Attribution and Further Refinement
Reference note: if this recursive Matryoshka encoding idea, the parity-guided discovery approach, or a refinement inspired by it is used in research, software, articles, prototypes, or derivative experiments, please reference this page as the conceptual source. A suitable citation is: Zero Kelvin Moralist, "Fractal Recursive Codec: Compressed Matryoshka Encoding."
Full Python Script
#!/usr/bin/env python3
"""
Zero Kelvin Moralist - Fractal Recursive Codec
==============================================
This script explores a recursive approach to encoding: a compact upper layer
can unfold into deeper layers, and the first decoded structures can become the
context needed to discover later structures. Instead of treating bytes only as
a flat stream, the script treats them as a seed for layered reconstruction.
Two reconstruction paths are included:
1. Exact archival reconstruction
The decoder reconstructs the original bytes exactly. For arbitrary input,
the missing information must be represented somewhere: residuals, parity,
a dictionary, a model, or another stored state. This branch is intentionally
conservative and reversible.
2. Generative Matryoshka reconstruction
A compact seed, recursive rule, and parity/checkpoint constraints generate
more output than the seed directly stores. Each decoded layer becomes the
parent context for the next layer. When explicit constraints end, the same
rule can continue autonomously into deeper model-consistent structure.
The core engineering idea is not that arbitrary information appears from
nothing. The idea is that a small representation can describe a recursive
process whose generated structure grows with decode depth and compute effort.
That makes the script a compact laboratory for future fractal/procedural
compression experiments.
"""
from __future__ import annotations
import argparse
import base64
import hashlib
import json
import sys
import zlib
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Iterable, Literal
import numpy as np
CONTAINER_CODEC = "zk-fractal-container"
CODEC_NAME = "zk-fractal-recursive-codec"
CODEC_VERSION = 3
PRIME_MOD = 65521 # largest prime below 2^16; useful for compact parity sums
DEFAULT_RULE = None # filled after GenerativeRule is declared
Mode = Literal["exact_text", "generative_matryoshka"]
# ---------------------------------------------------------------------------
# Container compression helpers
# ---------------------------------------------------------------------------
def canonical_json_bytes(payload: dict[str, Any]) -> bytes:
"""Return deterministic JSON bytes so compression and hashes are stable."""
return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
def b64encode(raw: bytes) -> str:
return base64.b64encode(raw).decode("ascii")
def b64decode(text: str) -> bytes:
return base64.b64decode(text.encode("ascii"))
def pack_payload(payload: dict[str, Any], *, level: int = 9) -> dict[str, Any]:
"""
Compress a codec payload into a portable JSON container.
This is the script's explicit compression layer. It does not make every
payload smaller, but it gives the recursive structures a normal compression
backend and provides honest size statistics.
"""
raw = canonical_json_bytes(payload)
compressed = zlib.compress(raw, level=level)
return {
"container_codec": CONTAINER_CODEC,
"container_version": CODEC_VERSION,
"compression": "zlib+base64",
"payload_sha256": hashlib.sha256(raw).hexdigest(),
"uncompressed_payload_bytes": len(raw),
"compressed_payload_bytes": len(compressed),
"compression_ratio": (len(compressed) / len(raw)) if raw else 1.0,
"payload_b64": b64encode(compressed),
}
def unpack_payload(container_or_payload: dict[str, Any]) -> dict[str, Any]:
"""Accept either a compressed container or a raw codec payload."""
if container_or_payload.get("container_codec") != CONTAINER_CODEC:
return container_or_payload
compressed = b64decode(container_or_payload["payload_b64"])
raw = zlib.decompress(compressed)
expected_hash = container_or_payload.get("payload_sha256")
actual_hash = hashlib.sha256(raw).hexdigest()
if expected_hash and expected_hash != actual_hash:
raise ValueError("Compressed payload hash mismatch.")
return json.loads(raw.decode("utf-8"))
# ---------------------------------------------------------------------------
# v2-compatible exact recursive residual codec
# ---------------------------------------------------------------------------
def predict_odd_from_coarse(coarse: np.ndarray, odd_count: int) -> np.ndarray:
"""
Predict odd-indexed byte values from neighboring even-indexed values.
Integer floor-midpoint prediction keeps the exact residuals integer-valued.
The residuals are the information needed for exact archival reconstruction.
"""
if odd_count == 0:
return np.array([], dtype=np.int64)
predictions: list[int] = []
for index in range(odd_count):
left = int(coarse[index])
right = int(coarse[index + 1]) if index + 1 < len(coarse) else left
predictions.append(left + ((right - left) // 2))
return np.array(predictions, dtype=np.int64)
def encode_exact_bytes(raw: bytes) -> dict[str, Any]:
"""
Losslessly encode bytes as recursive coarse samples plus residual layers.
This is the exact/reversible branch. It is deliberately not mystical:
exact recovery is possible because every omitted detail is represented as a
residual somewhere in the encoded structure.
"""
current = np.array(list(raw), dtype=np.int64)
layers: list[dict[str, Any]] = []
level = 0
while len(current) > 1:
coarse = current[0::2]
odd = current[1::2]
predicted = predict_odd_from_coarse(coarse, len(odd))
residuals = odd - predicted
layers.append(
{
"level": level,
"length": int(len(current)),
"coarse_length": int(len(coarse)),
"odd_count": int(len(odd)),
"predictor": "integer_floor_midpoint_boundary_left",
"residuals": [int(value) for value in residuals],
}
)
current = coarse
level += 1
payload = {
"codec": CODEC_NAME,
"version": CODEC_VERSION,
"mode": "exact_text",
"value_kind": "uint8_via_int_residuals",
"original_length": len(raw),
"base": [int(value) for value in current],
"layers": layers,
"commentary": {
"purpose": "Exact recursive residual reconstruction.",
"compression_note": "The residual structure is packed with zlib by the container. The recursive transform itself is reversible organization, not guaranteed compression.",
},
}
payload["stats"] = exact_stats(payload)
return payload
def decode_exact_bytes(payload: dict[str, Any]) -> bytes:
validate_payload_header(payload, expected_mode="exact_text")
current = np.array(payload.get("base", []), dtype=np.int64)
for layer in reversed(payload["layers"]):
length = int(layer["length"])
odd_count = int(layer["odd_count"])
residuals = np.array(layer["residuals"], dtype=np.int64)
if len(current) != int(layer["coarse_length"]):
raise ValueError(f"Layer {layer.get('level')} coarse length mismatch.")
if len(residuals) != odd_count:
raise ValueError(f"Layer {layer.get('level')} residual length mismatch.")
predicted = predict_odd_from_coarse(current, odd_count)
odd = predicted + residuals
reconstructed = np.empty(length, dtype=np.int64)
reconstructed[0::2] = current
reconstructed[1::2] = odd
current = reconstructed
if len(current) != int(payload["original_length"]):
raise ValueError("Decoded exact byte length mismatch.")
out: list[int] = []
for value in current:
byte = int(value)
if not 0 <= byte <= 255:
raise ValueError(f"Decoded byte out of range: {byte}")
out.append(byte)
return bytes(out)
def encode_exact_text(text: str) -> dict[str, Any]:
payload = encode_exact_bytes(text.encode("utf-8"))
payload["text_encoding"] = "utf-8"
return payload
def decode_exact_text(payload: dict[str, Any]) -> str:
return decode_exact_bytes(payload).decode(payload.get("text_encoding", "utf-8"))
def exact_stats(payload: dict[str, Any]) -> dict[str, Any]:
residuals = [float(value) for layer in payload.get("layers", []) for value in layer.get("residuals", [])]
zeros = sum(1 for value in residuals if value == 0.0)
stored_values = len(payload.get("base", [])) + len(residuals)
original = int(payload.get("original_length", 0))
return {
"layers": len(payload.get("layers", [])),
"original_values": original,
"stored_numeric_values_before_container_overhead": stored_values,
"numeric_ratio_before_container_overhead": (stored_values / original) if original else 1.0,
"zero_residual_ratio": (zeros / len(residuals)) if residuals else 1.0,
"max_abs_residual": max((abs(value) for value in residuals), default=0.0),
"mean_abs_residual": (sum(abs(value) for value in residuals) / len(residuals)) if residuals else 0.0,
}
# ---------------------------------------------------------------------------
# Matryoshka generative codec
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class GenerativeRule:
"""
Parameters for the recursive byte expansion rule.
The rule maps each parent byte to two child bytes using local context,
level number, a salt, and a per-level nonce. The nonce is intentionally
omitted from the encoded payload in constrained mode. The decoder discovers
it by searching for a child layer whose parity signature matches the stored
target. That is the core mechanism for "decoded information helps open the next
doll": the current layer is needed to test and discover the next layer.
"""
a: int = 73
b: int = 41
c: int = 19
d: int = 97
e: int = 53
salt: int = 131
DEFAULT_RULE = GenerativeRule()
def rule_from_dict(data: dict[str, Any] | None) -> GenerativeRule:
if not data:
return DEFAULT_RULE
return GenerativeRule(**{field: int(data.get(field, getattr(DEFAULT_RULE, field))) for field in asdict(DEFAULT_RULE)})
def expand_layer(parent: bytes, *, level: int, nonce: int, rule: GenerativeRule) -> bytes:
"""
Expand one layer into the next deeper layer.
This is the fractal/procedural step: the same local formula is reused at
every level. Output length doubles each level, making depth compute-bound.
"""
if not parent:
parent = b"\x00"
out = bytearray(len(parent) * 2)
n = len(parent)
for index, value in enumerate(parent):
left = parent[index - 1] if n > 1 else value
right = parent[(index + 1) % n] if n > 1 else value
child_a = (
value
+ rule.a * left
+ rule.b * level
+ rule.c * nonce
+ rule.salt
+ index
) & 0xFF
child_b = (
(value ^ right)
+ rule.d * level
+ rule.e * nonce
+ rule.salt
+ (index * 31)
) & 0xFF
out[2 * index] = child_a
out[2 * index + 1] = child_b
return bytes(out)
def parity_signature(data: bytes) -> dict[str, int]:
"""
Compact parity/checkpoint signature for one layer.
xor/sum/weighted sums are parity-like constraints. hash16 is included as a
small engineering guard to reduce ambiguity in this prototype. A future version
could replace this with Reed-Solomon, LDPC, fountain codes, or algebraic
constraints that are more efficient than this demonstrator.
"""
xor_value = 0
sum_mod = 0
weighted_mod = 0
alternating_mod = 0
for index, byte in enumerate(data):
xor_value ^= byte
sum_mod = (sum_mod + byte) % PRIME_MOD
weighted_mod = (weighted_mod + ((index + 1) * byte)) % PRIME_MOD
alternating_mod = (alternating_mod + (byte if index % 2 == 0 else -byte)) % PRIME_MOD
digest = hashlib.sha256(data).digest()
return {
"length": len(data),
"xor8": xor_value,
"sum_mod": sum_mod,
"weighted_mod": weighted_mod,
"alternating_mod": alternating_mod,
"hash16": int.from_bytes(digest[:2], "big"),
}
def signatures_match(candidate: dict[str, int], target: dict[str, int]) -> bool:
return all(int(candidate.get(key, -1)) == int(value) for key, value in target.items())
def authoring_nonce(parent: bytes, *, level: int, rule: GenerativeRule) -> int:
"""
Deterministically choose a hidden authoring nonce for the constrained path.
This nonce is not stored in the payload. During decoding it must be found
by search against the parity signature. The deterministic authoring choice
simply makes examples and tests reproducible.
"""
material = b"authoring-nonce-v3|" + bytes([level & 0xFF]) + canonical_json_bytes(asdict(rule)) + parent
return hashlib.sha256(material).digest()[0]
def continuation_nonce(parent: bytes, *, level: int, rule: GenerativeRule) -> int:
"""
Nonce for levels beyond stored parity constraints.
This is where the Matryoshka idea becomes autonomous: once explicit targets
are exhausted, the already reconstructed layer becomes the source for the
next nonce. The deeper output is model-consistent generated information,
not exact recovery of an arbitrary hidden original.
"""
material = b"autonomous-continuation-v3|" + bytes([level & 0xFF]) + canonical_json_bytes(asdict(rule)) + parent
return hashlib.sha256(material).digest()[0]
def discover_next_layer(parent: bytes, *, level: int, rule: GenerativeRule, target: dict[str, int]) -> tuple[int, bytes, dict[str, int], int]:
"""
Discover the per-level nonce by brute force and return the matching block.
Search space is deliberately tiny (0..255) for a transparent prototype. Future
engineering could turn this into a larger search, a solver, GPU work, or an
algebraic parity decoder.
"""
matches: list[tuple[int, bytes, dict[str, int]]] = []
for nonce in range(256):
candidate = expand_layer(parent, level=level, nonce=nonce, rule=rule)
signature = parity_signature(candidate)
if signatures_match(signature, target):
matches.append((nonce, candidate, signature))
if not matches:
raise ValueError(f"No nonce found for constrained level {level}.")
nonce, layer, signature = matches[0]
return nonce, layer, signature, len(matches)
def band_signature(levels: list[bytes]) -> dict[str, int]:
"""
RAID-inspired cross-level parity band.
The band signature does not store the generated levels. It stores compact
verification constraints over a stripe of levels. The decoder checks these
after reconstructing the stripe.
"""
joined = b"".join(len(level).to_bytes(8, "big") + level for level in levels)
sig = parity_signature(joined)
sig["covered_levels"] = len(levels)
return sig
def encode_generative_matryoshka(
seed: bytes,
*,
depth: int,
rule: GenerativeRule = DEFAULT_RULE,
band_size: int = 3,
max_generated_bytes: int = 2_000_000,
) -> dict[str, Any]:
"""
Encode a compact generative seed plus parity constraints up to `depth`.
The payload does not store generated layer bytes and does not store the
per-level nonces. It stores parity/checkpoint targets that allow the
decoder to discover each next layer from the current one.
"""
if depth < 0:
raise ValueError("depth must be non-negative")
if band_size <= 0:
raise ValueError("band_size must be positive")
if len(seed) * (2 ** depth) > max_generated_bytes:
raise ValueError("Requested depth exceeds max_generated_bytes guard.")
current = bytes(seed) or b"\x00"
level_constraints: list[dict[str, Any]] = []
generated_for_band: list[bytes] = []
bands: list[dict[str, Any]] = []
hidden_nonce_trace: list[int] = [] # omitted from payload; used only for self-checks before return
for level in range(1, depth + 1):
nonce = authoring_nonce(current, level=level, rule=rule)
next_layer = expand_layer(current, level=level, nonce=nonce, rule=rule)
target = parity_signature(next_layer)
level_constraints.append(
{
"level": level,
"target": target,
"role": "parity_target_for_nonce_discovery",
"commentary": "The nonce is omitted. Decoding searches nonce 0..255 until this target matches, then uses the discovered layer as context for the next level.",
}
)
hidden_nonce_trace.append(nonce)
generated_for_band.append(next_layer)
if len(generated_for_band) == band_size or level == depth:
start_level = level - len(generated_for_band) + 1
bands.append(
{
"start_level": start_level,
"end_level": level,
"band_size": len(generated_for_band),
"target": band_signature(generated_for_band),
"role": "raid_style_cross_level_parity_checkpoint",
}
)
generated_for_band = []
current = next_layer
payload = {
"codec": CODEC_NAME,
"version": CODEC_VERSION,
"mode": "generative_matryoshka",
"seed_b64": b64encode(seed),
"seed_length": len(seed),
"rule": asdict(rule),
"stored_constraint_depth": depth,
"band_size": band_size,
"level_constraints": level_constraints,
"band_constraints": bands,
"concept": {
"thesis": "A small seed plus recursive rules and parity constraints can unfold into deeper model-consistent layers.",
"decode_process": "Each constrained layer is discovered from the previous layer by searching for the nonce that satisfies the stored parity target.",
"beyond_stored_depth": "After stored constraints run out, deeper layers can continue autonomously from the reconstructed structure, but they are generated continuations rather than exact hidden source recovery.",
"engineering_warning": "This prototype demonstrates structure and computable expansion. It does not prove arbitrary lossless compression beyond information-theoretic limits.",
},
}
# Verify the payload can be decoded without hidden_nonce_trace before returning.
decoded = decode_generative_matryoshka(payload, depth=depth, max_generated_bytes=max_generated_bytes)
discovered = [item["nonce"] for item in decoded["trace"] if item["phase"] == "constrained"]
if discovered != hidden_nonce_trace:
raise AssertionError("Internal generative encode/decode trace mismatch.")
payload["stats"] = {
"seed_bytes": len(seed),
"stored_constraint_depth": depth,
"constrained_output_bytes_at_depth": len(decoded["final_bytes"]),
"expansion_ratio_vs_seed": (len(decoded["final_bytes"]) / len(seed)) if seed else len(decoded["final_bytes"]),
"level_constraints": len(level_constraints),
"band_constraints": len(bands),
}
return payload
def decode_generative_matryoshka(
payload: dict[str, Any],
*,
depth: int | None = None,
max_generated_bytes: int = 2_000_000,
) -> dict[str, Any]:
"""
Decode constrained levels, then optionally continue deeper autonomously.
Return includes the final bytes and a trace explaining which levels were
parity-discovered and which levels were autonomous continuation.
"""
validate_payload_header(payload, expected_mode="generative_matryoshka")
rule = rule_from_dict(payload.get("rule"))
current = b64decode(payload["seed_b64"]) or b"\x00"
stored_depth = int(payload.get("stored_constraint_depth", len(payload.get("level_constraints", []))))
requested_depth = stored_depth if depth is None else int(depth)
if requested_depth < 0:
raise ValueError("decode depth must be non-negative")
if len(current) * (2 ** requested_depth) > max_generated_bytes:
raise ValueError("Requested decode depth exceeds max_generated_bytes guard.")
constraints_by_level = {int(item["level"]): item["target"] for item in payload.get("level_constraints", [])}
band_targets = payload.get("band_constraints", [])
trace: list[dict[str, Any]] = []
generated_levels: dict[int, bytes] = {}
for level in range(1, requested_depth + 1):
if level in constraints_by_level:
nonce, next_layer, signature, ambiguity = discover_next_layer(
current, level=level, rule=rule, target=constraints_by_level[level]
)
phase = "constrained"
else:
nonce = continuation_nonce(current, level=level, rule=rule)
next_layer = expand_layer(current, level=level, nonce=nonce, rule=rule)
signature = parity_signature(next_layer)
ambiguity = 1
phase = "autonomous_continuation"
generated_levels[level] = next_layer
trace.append(
{
"level": level,
"phase": phase,
"nonce": nonce,
"length": len(next_layer),
"signature": signature,
"candidate_matches": ambiguity,
}
)
current = next_layer
verified_bands: list[dict[str, Any]] = []
for band in band_targets:
start = int(band["start_level"])
end = int(band["end_level"])
if end > requested_depth:
continue
levels = [generated_levels[level] for level in range(start, end + 1)]
actual = band_signature(levels)
target = band["target"]
ok = signatures_match(actual, target)
verified_bands.append({"start_level": start, "end_level": end, "ok": ok})
if not ok:
raise ValueError(f"Band parity verification failed for levels {start}-{end}.")
return {
"final_bytes": current,
"trace": trace,
"verified_bands": verified_bands,
"stored_depth": stored_depth,
"requested_depth": requested_depth,
"autonomous_levels": max(0, requested_depth - stored_depth),
}
# ---------------------------------------------------------------------------
# Validation, I/O, CLI, tests
# ---------------------------------------------------------------------------
def validate_payload_header(payload: dict[str, Any], *, expected_mode: Mode | None = None) -> None:
if payload.get("codec") != CODEC_NAME:
raise ValueError(f"Unsupported codec marker: {payload.get('codec')!r}")
if payload.get("version") != CODEC_VERSION:
raise ValueError(f"Unsupported codec version: {payload.get('version')!r}")
if expected_mode and payload.get("mode") != expected_mode:
raise ValueError(f"Expected mode {expected_mode!r}, got {payload.get('mode')!r}")
def read_json(path: str) -> dict[str, Any]:
if path == "-":
return json.load(sys.stdin)
with Path(path).open("r", encoding="utf-8") as handle:
return json.load(handle)
def write_json(payload: dict[str, Any], path: str, *, pretty: bool = True) -> None:
kwargs: dict[str, Any] = {"ensure_ascii": False}
if pretty:
kwargs["indent"] = 2
text = json.dumps(payload, **kwargs)
if path == "-":
print(text)
return
with Path(path).open("w", encoding="utf-8") as handle:
handle.write(text)
handle.write("\n")
def printable_bytes_preview(data: bytes, limit: int = 80) -> str:
preview = data[:limit]
ascii_preview = "".join(chr(byte) if 32 <= byte <= 126 else "." for byte in preview)
suffix = "..." if len(data) > limit else ""
return f"{ascii_preview}{suffix}"
def summarize_container(container_or_payload: dict[str, Any]) -> str:
if container_or_payload.get("container_codec") == CONTAINER_CODEC:
return (
f"container={CONTAINER_CODEC}, compression={container_or_payload['compression']}, "
f"raw={container_or_payload['uncompressed_payload_bytes']} bytes, "
f"compressed={container_or_payload['compressed_payload_bytes']} bytes, "
f"ratio={container_or_payload['compression_ratio']:.3f}"
)
payload = container_or_payload
return f"raw_payload mode={payload.get('mode')} codec={payload.get('codec')} v{payload.get('version')}"
def run_demo() -> None:
print("Fractal Recursive Codec demo")
print("================================")
print()
exact_text = "Fractal recursion works as an exact residual codec."
exact_payload = encode_exact_text(exact_text)
exact_container = pack_payload(exact_payload)
exact_roundtrip = decode_exact_text(unpack_payload(exact_container))
print("1) Exact recursive residual mode")
print("--------------------------------")
print(f"Original: {exact_text!r}")
print(f"Container: {summarize_container(exact_container)}")
print(f"Stats: {exact_payload['stats']}")
print(f"Decoded equals original: {exact_roundtrip == exact_text}")
print()
seed_text = "ZKM"
depth = 6
requested_depth = 8
gen_payload = encode_generative_matryoshka(seed_text.encode("utf-8"), depth=depth, band_size=3)
gen_container = pack_payload(gen_payload)
decoded = decode_generative_matryoshka(unpack_payload(gen_container), depth=requested_depth)
print("2) Generative Matryoshka mode")
print("-----------------------------")
print(f"Seed: {seed_text!r} ({len(seed_text.encode('utf-8'))} bytes)")
print(f"Stored constrained depth: {depth}")
print(f"Requested decode depth: {requested_depth}")
print(f"Final generated bytes: {len(decoded['final_bytes'])}")
print(f"Expansion vs seed: {len(decoded['final_bytes']) / len(seed_text.encode('utf-8')):.1f}x")
print(f"Container: {summarize_container(gen_container)}")
print(f"Verified parity bands: {decoded['verified_bands']}")
print("Trace:")
for item in decoded["trace"]:
print(
f" level={item['level']:02d} phase={item['phase']:<24} "
f"nonce={item['nonce']:03d} length={item['length']} matches={item['candidate_matches']}"
)
print(f"Generated preview: {printable_bytes_preview(decoded['final_bytes'])}")
print()
print("Interpretation:")
print(" - Levels 1-6 were discovered using stored parity/checkpoint targets.")
print(" - Levels 7-8 continue autonomously from the reconstructed structure.")
print(" - The generated bytes are model-consistent deeper structure, not a claim")
print(" that arbitrary hidden source bytes were recovered from nothing.")
def run_self_tests() -> None:
# Exact mode tests.
for text in ["", "A", "Hello", "Fractal recursive exact text round trip.", "UTF-8: recursion and structure"]:
payload = encode_exact_text(text)
container = pack_payload(payload)
assert decode_exact_text(unpack_payload(container)) == text
# Generative constrained decode and deterministic trace tests.
payload = encode_generative_matryoshka(b"ZKM", depth=5, band_size=2)
container = pack_payload(payload)
unpacked = unpack_payload(container)
decoded_5 = decode_generative_matryoshka(unpacked, depth=5)
decoded_7 = decode_generative_matryoshka(unpacked, depth=7)
assert len(decoded_5["final_bytes"]) == 3 * (2 ** 5)
assert len(decoded_7["final_bytes"]) == 3 * (2 ** 7)
assert decoded_5["autonomous_levels"] == 0
assert decoded_7["autonomous_levels"] == 2
assert all(item["candidate_matches"] >= 1 for item in decoded_5["trace"])
assert all(item["ok"] for item in decoded_5["verified_bands"])
# Corruption/constraint test: changing a parity target should fail.
corrupted = json.loads(json.dumps(unpacked))
corrupted["level_constraints"][0]["target"]["xor8"] ^= 1
try:
decode_generative_matryoshka(corrupted, depth=1)
except ValueError:
pass
else:
raise AssertionError("Corrupted constraint did not fail.")
# Container hash test.
bad_container = json.loads(json.dumps(container))
bad_container["payload_sha256"] = "0" * 64
try:
unpack_payload(bad_container)
except ValueError:
pass
else:
raise AssertionError("Container hash mismatch did not fail.")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fractal recursive codec with exact and Matryoshka modes.")
sub = parser.add_subparsers(dest="command", required=True)
sub.add_parser("demo", help="Run exact + generative demonstration.")
sub.add_parser("test", help="Run built-in self-tests.")
enc_exact = sub.add_parser("encode-exact", help="Encode text exactly as recursive residuals.")
enc_exact.add_argument("--text", required=True)
enc_exact.add_argument("--out", required=True)
enc_exact.add_argument("--raw-payload", action="store_true", help="Do not wrap in compressed container.")
enc_exact.add_argument("--compact", action="store_true")
dec_exact = sub.add_parser("decode-exact", help="Decode exact text payload/container.")
dec_exact.add_argument("--in", dest="input_path", required=True)
enc_gen = sub.add_parser("encode-generative", help="Encode seed plus parity constraints for Matryoshka generation.")
enc_gen.add_argument("--text", required=True, help="Seed text.")
enc_gen.add_argument("--depth", type=int, default=6, help="Stored constrained depth.")
enc_gen.add_argument("--band-size", type=int, default=3)
enc_gen.add_argument("--out", required=True)
enc_gen.add_argument("--raw-payload", action="store_true", help="Do not wrap in compressed container.")
enc_gen.add_argument("--compact", action="store_true")
enc_gen.add_argument("--max-generated-bytes", type=int, default=2_000_000)
dec_gen = sub.add_parser("decode-generative", help="Decode/generate Matryoshka layers.")
dec_gen.add_argument("--in", dest="input_path", required=True)
dec_gen.add_argument("--depth", type=int, default=None, help="Requested depth; may exceed stored constrained depth.")
dec_gen.add_argument("--out-bytes", default=None, help="Optional binary output path for final generated bytes.")
dec_gen.add_argument("--max-generated-bytes", type=int, default=2_000_000)
inspect = sub.add_parser("inspect", help="Print a summary of a payload/container.")
inspect.add_argument("--in", dest="input_path", required=True)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.command == "demo":
run_demo()
return 0
if args.command == "test":
run_self_tests()
print("All self-tests passed.")
return 0
if args.command == "encode-exact":
payload = encode_exact_text(args.text)
out = payload if args.raw_payload else pack_payload(payload)
write_json(out, args.out, pretty=not args.compact)
return 0
if args.command == "decode-exact":
payload = unpack_payload(read_json(args.input_path))
print(decode_exact_text(payload))
return 0
if args.command == "encode-generative":
payload = encode_generative_matryoshka(
args.text.encode("utf-8"),
depth=args.depth,
band_size=args.band_size,
max_generated_bytes=args.max_generated_bytes,
)
out = payload if args.raw_payload else pack_payload(payload)
write_json(out, args.out, pretty=not args.compact)
return 0
if args.command == "decode-generative":
payload = unpack_payload(read_json(args.input_path))
decoded = decode_generative_matryoshka(payload, depth=args.depth, max_generated_bytes=args.max_generated_bytes)
if args.out_bytes:
Path(args.out_bytes).write_bytes(decoded["final_bytes"])
print(
json.dumps(
{
"stored_depth": decoded["stored_depth"],
"requested_depth": decoded["requested_depth"],
"final_bytes": len(decoded["final_bytes"]),
"autonomous_levels": decoded["autonomous_levels"],
"verified_bands": decoded["verified_bands"],
"trace": decoded["trace"],
"preview": printable_bytes_preview(decoded["final_bytes"]),
},
ensure_ascii=False,
indent=2,
)
)
return 0
if args.command == "inspect":
container_or_payload = read_json(args.input_path)
print(summarize_container(container_or_payload))
payload = unpack_payload(container_or_payload)
print(f"payload mode={payload.get('mode')} codec={payload.get('codec')} v{payload.get('version')}")
if payload.get("mode") == "generative_matryoshka":
print(f"seed_length={payload.get('seed_length')} stored_depth={payload.get('stored_constraint_depth')} bands={len(payload.get('band_constraints', []))}")
print(f"concept={payload.get('concept', {}).get('thesis')}")
elif payload.get("mode") == "exact_text":
print(f"original_length={payload.get('original_length')} layers={len(payload.get('layers', []))} stats={payload.get('stats')}")
return 0
raise AssertionError(f"Unhandled command {args.command!r}")
if __name__ == "__main__":
raise SystemExit(main())