#!/usr/bin/env python3
"""
GPU-Parallel Metaprobe Compression Engine (GPMCE)
==================================================
Wires together the three existing surfaces in a multi-threaded, GPU-aware
feeding mechanism:

  ┌────────────┐     ┌──────────┐     ┌──────────────┐
  │ GPU Surface │────▶│ Metaprobe│────▶│ PIST-GCL     │
  │ (wgpu+CUDA) │     │ Lawcheck │     │ Compression  │
  └────────────┘     └──────────┘     └──────┬───────┘
                                             │
                    ┌────────────────────────┘
                    ▼
              ┌──────────┐
              │ Receipt   │
              │ Generator │
              └──────────┘

Architecture:
  Phase 0 — Source Feeder: multi-threaded ingestion from files, dirs, stdin
  Phase 1 — GPU Compute: wgpu WGSL shader dispatch + CUDA tensor batch ops
  Phase 2 — Metaprobe Gate: lawful resonance/coherence/entropy check
  Phase 3 — PIST-GCL Compress: 4-layer manifold compression
  Phase 4 — Receipt Chain: cryptographic receipt per operation

Thread Model:
  - I/O threads: read sources into a bounded queue
  - GPU feeder threads: push chunks through wgpu compute
  - Metaprobe threads: run lawfulness checks in parallel
  - Compressor threads: PIST-GCL with shared VRAM monitor
  - Receipt writer: single-threaded append-only JSONL output

Usage:
  python gpu_metaprobe_compression_engine.py [--input FILE|DIR] [--output DIR]
  python gpu_metaprobe_compression_engine.py --daemon --watch-dir /data/incoming
"""

import os
import sys
import json
import time
import struct
import hashlib
import signal
import threading
import argparse
from pathlib import Path
from queue import Queue, Empty
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Tuple
from collections import Counter, OrderedDict
from concurrent.futures import ThreadPoolExecutor

BASE = Path("/home/allaun/Documents/Research Stack")
sys.path.insert(0, str(BASE / "5-Applications/scripts"))
sys.path.insert(0, str(BASE / "5-Applications/pist-scripts"))
sys.path.insert(0, str(BASE / "3-Mathematical-Models"))


# ═══════════════════════════════════════════════════════════════════════════
# Hardware Detection
# ═══════════════════════════════════════════════════════════════════════════

HAS_CUDA = False
HAS_WGPU = False
try:
    import torch
    import torch.cuda as cuda_
    HAS_CUDA = cuda_.is_available()
except ImportError:
    pass

try:
    import wgpu
    HAS_WGPU = True
except ImportError:
    pass


# ═══════════════════════════════════════════════════════════════════════════
# Data Structures
# ═══════════════════════════════════════════════════════════════════════════

@dataclass
class Chunk:
    id: str
    source: str
    data: bytes
    index: int
    timestamp: float = 0.0

@dataclass
class GPUResult:
    chunk_id: str
    processed: bytes
    gpu_time_ms: float
    surface: str

@dataclass
class MetaprobeResult:
    chunk_id: str
    resonance: float
    coherence: float
    entropy: float
    lawful: bool

@dataclass
class CompressResult:
    chunk_id: str
    compressed: bytes
    ratio: float
    stats: dict
    layer_costs: dict

@dataclass
class Receipt:
    chunk_id: str
    source: str
    original_size: int
    compressed_size: int
    ratio: float
    gpu_time_ms: float
    resonance: float
    coherence: float
    lawful: bool
    timestamp: str
    hash: str


# ═══════════════════════════════════════════════════════════════════════════
# Phase 0 — Multi-Threaded Source Feeder
# ═══════════════════════════════════════════════════════════════════════════

class SourceFeeder:
    """Multi-threaded source ingestion into a bounded queue."""

    def __init__(self, queue: Queue, chunk_size: int = 65536):
        self.queue = queue
        self.chunk_size = chunk_size
        self.counter = 0
        self._lock = threading.Lock()
        self._done = threading.Event()

    def feed_file(self, path: Path):
        source = path.name
        try:
            data = path.read_bytes()
            self._split_and_queue(data, source)
        except Exception as e:
            print(f"  [FEEDER] Error reading {path}: {e}")

    def feed_directory(self, path: Path, pattern: str = "*"):
        files = sorted(path.rglob(pattern))
        print(f"  [FEEDER] Found {len(files)} files in {path}")
        with ThreadPoolExecutor(max_workers=min(4, os.cpu_count() or 2)) as pool:
            pool.map(self.feed_file, files)

    def feed_stdin(self):
        data = sys.stdin.buffer.read()
        self._split_and_queue(data, "stdin")

    def feed_bytes(self, data: bytes, source: str = "memory"):
        self._split_and_queue(data, source)

    def _split_and_queue(self, data: bytes, source: str):
        for i in range(0, len(data), self.chunk_size):
            chunk_data = data[i:i + self.chunk_size]
            with self._lock:
                chunk_id = f"{source}:{self.counter}"
                self.counter += 1
            self.queue.put(Chunk(
                id=chunk_id, source=source, data=chunk_data,
                index=self.counter, timestamp=time.time()
            ))

    def done(self):
        self._done.set()

    @property
    def is_done(self) -> bool:
        return self._done.is_set()

    @property
    def total_chunks(self) -> int:
        return self.counter


# ═══════════════════════════════════════════════════════════════════════════
# Phase 1 — GPU Compute Surface
# ═══════════════════════════════════════════════════════════════════════════

class GPUComputeSurface:
    """Unified GPU compute — CUDA (PyTorch) primary, wgpu (WebGPU/Vulkan) fallback."""

    def __init__(self):
        self.wgpu_device = None
        self.cuda_available = HAS_CUDA
        if HAS_WGPU:
            try:
                self.wgpu_device = wgpu.get_default_device()
                print(f"  [GPU] wgpu device ready")
            except Exception as e:
                print(f"  [GPU] wgpu init: {e} — using CUDA")

    def process(self, chunk: Chunk) -> GPUResult:
        t0 = time.perf_counter()
        if self.cuda_available:
            result = self._cuda_process(chunk.data)
            surface = "cuda"
        elif self.wgpu_device:
            result = self._wgpu_process(chunk.data)
            surface = "wgpu"
        else:
            result = self._cpu_process(chunk.data)
            surface = "cpu"
        elapsed_ms = (time.perf_counter() - t0) * 1000
        return GPUResult(chunk_id=chunk.id, processed=result,
                         gpu_time_ms=elapsed_ms, surface=surface)

    def _wgpu_process(self, data: bytes) -> bytes:
        key = 0x1F1F1F1F
        padding = (4 - len(data) % 4) % 4
        padded = data + b'\x00' * padding
        arr = memoryview(padded).cast('I')

        shader_code = """
        @group(0) @binding(0) var<storage, read> input: array<u32>;
        @group(0) @binding(1) var<storage, read_write> output: array<u32>;
        @compute @workgroup_size(256)
        fn main(@builtin(global_invocation_id) global_id: vec3<u32>) {
            let idx = global_id.x;
            if (idx >= arrayLength(&input)) { return; }
            let key = 0x1F1F1F1Fu;
            output[idx] = input[idx] ^ key;
        }
        """

        shader = self.wgpu_device.create_shader_module(code=shader_code)
        pipeline = self.wgpu_device.create_compute_pipeline(
            layout="auto",
            compute={"module": shader, "entry_point": "main"}
        )

        nbytes = len(arr) * 4
        in_buf = self.wgpu_device.create_buffer(
            size=nbytes, usage=wgpu.BufferUsage.STORAGE | wgpu.BufferUsage.COPY_DST)
        out_buf = self.wgpu_device.create_buffer(
            size=nbytes, usage=wgpu.BufferUsage.STORAGE | wgpu.BufferUsage.COPY_SRC)

        self.wgpu_device.queue.write_buffer(in_buf, 0, arr.tobytes())

        bind_group = self.wgpu_device.create_bind_group(
            layout=pipeline.get_bind_group_layout(0),
            entries=[
                {"binding": 0, "resource": {"buffer": in_buf, "offset": 0, "size": nbytes}},
                {"binding": 1, "resource": {"buffer": out_buf, "offset": 0, "size": nbytes}}
            ]
        )

        encoder = self.wgpu_device.create_command_encoder()
        cpass = encoder.begin_compute_pass()
        cpass.set_pipeline(pipeline)
        cpass.set_bind_group(0, bind_group)
        cpass.dispatch_workgroups((len(arr) + 255) // 256)
        cpass.end()

        readback = self.wgpu_device.create_buffer(
            size=nbytes, usage=wgpu.BufferUsage.MAP_READ | wgpu.BufferUsage.COPY_DST)
        encoder.copy_buffer_to_buffer(out_buf, 0, readback, 0, nbytes)
        self.wgpu_device.queue.submit([encoder.finish()])

        result = self.wgpu_device.queue.read_buffer(readback)
        return bytes(result)[:len(data)]

    def _cuda_process(self, data: bytes) -> bytes:
        import numpy as np
        arr = np.frombuffer(data, dtype=np.uint8)
        tensor = torch.from_numpy(arr.copy()).to(torch.device("cuda:0"))
        key_tensor = torch.full_like(tensor, 0x1F, dtype=torch.uint8)
        result = torch.bitwise_xor(tensor, key_tensor).cpu().numpy().tobytes()
        return result

    def _cpu_process(self, data: bytes) -> bytes:
        return bytes(b ^ 0x1F for b in data)


# ═══════════════════════════════════════════════════════════════════════════
# Phase 2 — Metaprobe Lawfulness Gate
# ═══════════════════════════════════════════════════════════════════════════

class MetaprobeGate:
    """Lawfulness checking using metaprobe resonance + coherence."""

    def __init__(self, threshold: float = 0.6):
        self.threshold = threshold

    def check(self, chunk: Chunk, gpu_result: GPUResult) -> MetaprobeResult:
        data = gpu_result.processed
        if not data:
            return MetaprobeResult(chunk.id, 0.0, 0.0, 0.0, False)

        entropy = self._shannon(data)
        coherence = self._structural_coherence(data)
        resonance = self._combined_resonance(data, entropy)

        lawful = (resonance >= self.threshold and
                  coherence >= self.threshold and
                  0.1 < entropy < 0.9)

        return MetaprobeResult(
            chunk_id=chunk.id,
            resonance=round(resonance, 4),
            coherence=round(coherence, 4),
            entropy=round(entropy, 4),
            lawful=lawful
        )

    def _shannon(self, data: bytes) -> float:
        if len(data) < 2:
            return 0.0
        c = Counter(data)
        n = len(data)
        h = -sum((cnt / n) * __import__('math').log2(cnt / n) for cnt in c.values())
        return h / 8.0

    def _structural_coherence(self, data: bytes) -> float:
        if len(data) < 2:
            return 0.0
        smooth = sum(1 for i in range(len(data) - 1)
                     if abs(data[i] - data[i + 1]) < 32)
        return smooth / (len(data) - 1)

    def _combined_resonance(self, data: bytes, entropy: float) -> float:
        byte_range = max(data) - min(data) if data else 1
        spread_score = min(1.0, byte_range / 128.0)
        return (spread_score + (1.0 - abs(entropy - 0.5) * 2)) / 2.0


# ═══════════════════════════════════════════════════════════════════════════
# Phase 3 — PIST-GCL Compression
# ═══════════════════════════════════════════════════════════════════════════

class CompressionEngine:
    """Multi-strategy compression: PIST-GCL, brotli, zlib — with GPU-aware throttling."""

    def __init__(self):
        self._lock = threading.Lock()
        import zlib
        self._zlib = zlib
        try:
            import brotli
            self._brotli = brotli
        except ImportError:
            self._brotli = None
        try:
            from pist_gcl_compression import PISTGCLCompressor
            self._pist = PISTGCLCompressor()
        except Exception:
            self._pist = None
        if self._pist is None and self._brotli is None:
            print(f"  [COMPRESS] Using zlib-6 — brotli not installed")

    def compress(self, chunk: Chunk, metaprobe: MetaprobeResult) -> CompressResult:
        data = chunk.data
        if len(data) < 256:
            return CompressResult(chunk.id, data, 1.0,
                                  {"strategy": "passthrough-small"},
                                  {})

        ratio = 1.0
        compressed = data
        strategy = "passthrough"
        layer_costs = {}

        with self._lock:
            if self._pist:
                try:
                    result = self._pist.compress(data)
                    comp = result.get("compressed", data)
                    r = result.get("stats", {}).get("ratio", 1.0)
                    if r > 1.1 and len(comp) < len(data):
                        ratio = r
                        compressed = comp
                        strategy = "pist-gcl"
                        layer_costs = result.get("layer_costs", {})
                except Exception:
                    pass

            if strategy == "passthrough" and self._brotli:
                try:
                    comp = self._brotli.compress(data, quality=6)
                    r = len(data) / max(len(comp), 1)
                    if r > ratio and len(comp) < len(compressed):
                        ratio = r
                        compressed = comp
                        strategy = "brotli-6"
                except Exception:
                    pass

            if strategy == "passthrough":
                try:
                    comp = self._zlib.compress(data, 6)
                    r = len(data) / max(len(comp), 1)
                    ratio = r
                    compressed = comp
                    strategy = "zlib-6"
                except Exception:
                    pass

        return CompressResult(chunk.id, compressed, round(ratio, 3),
                              {"strategy": strategy}, layer_costs)


# ═══════════════════════════════════════════════════════════════════════════
# Phase 4 — Receipt Chain
# ═══════════════════════════════════════════════════════════════════════════

class ReceiptGenerator:
    """Cryptographic receipt chain — append-only JSONL output."""

    def __init__(self, output_path: Path):
        self.output_path = output_path
        self._lock = threading.Lock()
        self.count = 0
        self.total_original = 0
        self.total_compressed = 0
        self.total_lawful = 0
        output_path.parent.mkdir(parents=True, exist_ok=True)
        self._fh = open(str(output_path), "a")

    def emit(self, chunk: Chunk, gpu: GPUResult, metaprobe: MetaprobeResult,
             comp: CompressResult) -> Receipt:
        receipt = Receipt(
            chunk_id=chunk.id,
            source=chunk.source,
            original_size=len(chunk.data),
            compressed_size=len(comp.compressed),
            ratio=comp.ratio,
            gpu_time_ms=round(gpu.gpu_time_ms, 3),
            resonance=metaprobe.resonance,
            coherence=metaprobe.coherence,
            lawful=metaprobe.lawful,
            timestamp=datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f"),
            hash=hashlib.sha256(comp.compressed).hexdigest()[:32]
        )

        with self._lock:
            self._fh.write(json.dumps({
                "chunk_id": receipt.chunk_id,
                "source": receipt.source,
                "original_size": receipt.original_size,
                "compressed_size": receipt.compressed_size,
                "ratio": receipt.ratio,
                "gpu_time_ms": receipt.gpu_time_ms,
                "resonance": receipt.resonance,
                "coherence": receipt.coherence,
                "entropy": metaprobe.entropy,
                "lawful": receipt.lawful,
                "timestamp": receipt.timestamp,
                "hash": receipt.hash
            }) + "\n")
            self._fh.flush()

            self.count += 1
            self.total_original += receipt.original_size
            self.total_compressed += receipt.compressed_size
            if receipt.lawful:
                self.total_lawful += 1

        return receipt

    def stats(self) -> dict:
        return {
            "total_chunks": self.count,
            "total_original_bytes": self.total_original,
            "total_compressed_bytes": self.total_compressed,
            "overall_ratio": self.total_original / max(self.total_compressed, 1),
            "lawful_fraction": self.total_lawful / max(self.count, 1)
        }


# ═══════════════════════════════════════════════════════════════════════════
# The Engine — Orchestrator
# ═══════════════════════════════════════════════════════════════════════════

class GPUMetaprobeCompressionEngine:
    """
    Master orchestrator. Phase pipeline with thread-pool parallelism:

    Each chunk flows through: GPU → Metaprobe → Compress → Receipt
    Thread pools for GPU, metaprobe, and compression run concurrently.
    """

    def __init__(self, output_dir: Path, workers: Optional[int] = None,
                 metaprobe_threshold: float = 0.6):
        self.output_dir = output_dir
        output_dir.mkdir(parents=True, exist_ok=True)
        self.chunk_size = 65536
        self.threshold = metaprobe_threshold

        cpu = os.cpu_count() or 12
        self.n_gpu_workers = min(2, workers or cpu)
        self.n_meta_workers = min(4, workers or cpu)
        self.n_comp_workers = min(2, workers or cpu)

        self.gpu_queue = Queue(maxsize=256)
        self.meta_queue = Queue(maxsize=256)
        self.comp_queue = Queue(maxsize=256)

        self.gpu_surface = GPUComputeSurface()
        self.metaprobe_gate = MetaprobeGate(threshold=self.threshold)
        self.compression = CompressionEngine()
        self.receipts = ReceiptGenerator(output_dir / "receipts.jsonl")

        self.running = False
        self._start_time = 0.0

    def start(self):
        self.running = True
        self._start_time = time.time()
        pools = []

        pools.append(ThreadPoolExecutor(self.n_gpu_workers, thread_name_prefix="gpu"))
        pools.append(ThreadPoolExecutor(self.n_meta_workers, thread_name_prefix="meta"))
        pools.append(ThreadPoolExecutor(self.n_comp_workers, thread_name_prefix="comp"))

        pools[0].submit(self._gpu_worker, pools)
        pools[1].submit(self._metaprobe_worker, pools)
        pools[2].submit(self._compression_worker)

        return pools

    def stop(self, pools: list):
        self.running = False
        for pool in pools:
            pool.shutdown(wait=True)

    def feed(self, feeder: SourceFeeder):
        feeder.done()
        return feeder.total_chunks

    def _gpu_worker(self, pools: list):
        while self.running:
            try:
                chunk = self.gpu_queue.get(timeout=0.5)
            except Empty:
                continue
            try:
                result = self.gpu_surface.process(chunk)
                self.meta_queue.put((chunk, result))
            except Exception as e:
                print(f"  [GPU] Error on {chunk.id}: {e}")
            finally:
                self.gpu_queue.task_done()

    def _metaprobe_worker(self, pools: list):
        while self.running:
            try:
                chunk, gpu_result = self.meta_queue.get(timeout=0.5)
            except Empty:
                continue
            try:
                meta_result = self.metaprobe_gate.check(chunk, gpu_result)
                self.comp_queue.put((chunk, gpu_result, meta_result))
            except Exception as e:
                print(f"  [META] Error on {chunk.id}: {e}")
            finally:
                self.meta_queue.task_done()

    def _compression_worker(self):
        while self.running:
            try:
                chunk, gpu_result, meta_result = self.comp_queue.get(timeout=0.5)
            except Empty:
                continue
            try:
                comp_result = self.compression.compress(chunk, meta_result)
                receipt = self.receipts.emit(chunk, gpu_result, meta_result, comp_result)
                if self.receipts.count % 100 == 0:
                    self._print_progress(receipt)
            except Exception as e:
                print(f"  [COMP] Error on {chunk.id}: {e}")
            finally:
                self.comp_queue.task_done()

    def _print_progress(self, receipt: Receipt):
        elapsed = time.time() - self._start_time
        stats = self.receipts.stats()
        throughput = stats["total_original_bytes"] / max(elapsed, 0.001) / (1024**2)
        print(f"\n  [ENGINE] {stats['total_chunks']} chunks | "
              f"{stats['total_original_bytes']/1024:.1f}KB → "
              f"{stats['total_compressed_bytes']/1024:.1f}KB | "
              f"ratio={stats['overall_ratio']:.2f}x | "
              f"lawful={stats['lawful_fraction']:.1%} | "
              f"{throughput:.1f} MB/s")

    def finalize(self, pools: list) -> dict:
        self.stop(pools)
        stats = self.receipts.stats()
        elapsed = time.time() - self._start_time
        stats["elapsed_seconds"] = round(elapsed, 1)
        stats["throughput_mbps"] = round(
            stats["total_original_bytes"] / max(elapsed, 0.001) / (1024**2), 1
        )
        stats["timestamp"] = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
        stats["gpu_surface"] = "wgpu" if HAS_WGPU else ("cuda" if HAS_CUDA else "cpu")

        report_path = self.output_dir / "compression_report.json"
        with open(str(report_path), "w") as f:
            json.dump(stats, f, indent=2)

        print(f"\n{'='*60}")
        print(f"  ENGINE COMPLETE")
        print(f"{'='*60}")
        for k, v in stats.items():
            if isinstance(v, float):
                print(f"  {k}: {v:.3f}")
            else:
                print(f"  {k}: {v}")
        print(f"  receipts: {self.output_dir}/receipts.jsonl")
        print(f"  report:   {report_path}")
        print(f"{'='*60}")
        return stats


# ═══════════════════════════════════════════════════════════════════════════
# Daemon Mode
# ═══════════════════════════════════════════════════════════════════════════

class Daemon:
    """File-watching daemon that feeds the engine continuously."""

    def __init__(self, engine: GPUMetaprobeCompressionEngine, watch_dir: Path,
                 interval: float = 30.0):
        self.engine = engine
        self.watch_dir = Path(watch_dir)
        self.interval = interval
        self.seen_hashes = set()

    def run(self):
        print(f"  [DAEMON] Watching {self.watch_dir} every {self.interval}s")
        feeder = SourceFeeder(self.engine.gpu_queue, self.engine.chunk_size)
        self.engine.gpu_queue = feeder.queue

        while True:
            files = sorted(self.watch_dir.rglob("*"))
            new_files = []
            for f in files:
                if not f.is_file() or f.suffix in (".jsonl", ".json", ".tmp"):
                    continue
                try:
                    h = hashlib.sha256(str(f).encode()).hexdigest()
                    if h not in self.seen_hashes:
                        self.seen_hashes.add(h)
                        new_files.append(f)
                except Exception:
                    pass

            if new_files:
                print(f"  [DAEMON] {len(new_files)} new files")
                feeder = SourceFeeder(self.engine.gpu_queue, self.engine.chunk_size)
                for f in new_files:
                    feeder.feed_file(f)
                feeder.done()

            time.sleep(self.interval)


# ═══════════════════════════════════════════════════════════════════════════
# CLI Entry Point
# ═══════════════════════════════════════════════════════════════════════════

def main():
    parser = argparse.ArgumentParser(
        description="GPU-Parallel Metaprobe Compression Engine")
    parser.add_argument("--input", "-i", help="Input file or directory")
    parser.add_argument("--output", "-o", default="5-Applications/out/gpmce",
                        help="Output directory")
    parser.add_argument("--daemon", action="store_true", help="Run as file-watching daemon")
    parser.add_argument("--watch-dir", default="~/Downloads",
                        help="Directory to watch in daemon mode")
    parser.add_argument("--threshold", type=float, default=0.6,
                        help="Metaprobe lawfulness threshold (0-1)")
    parser.add_argument("--benchmark", action="store_true",
                        help="Run self-benchmark with synthetic data")
    parser.add_argument("--workers", type=int, default=None,
                        help="Worker thread count")
    args = parser.parse_args()

    output_dir = (BASE / args.output).resolve()
    print(f"\n{'='*60}")
    print(f"  GPU-Parallel Metaprobe Compression Engine")
    print(f"  GPU: wgpu={'✓' if HAS_WGPU else '✗'} CUDA={'✓' if HAS_CUDA else '✗'}")
    print(f"  Output: {output_dir}")
    print(f"{'='*60}")

    engine = GPUMetaprobeCompressionEngine(
        output_dir, workers=args.workers,
        metaprobe_threshold=args.threshold
    )

    if args.benchmark:
        _run_benchmark(engine)
        return

    pools = engine.start()

    if args.daemon:
        watch_dir = Path(args.watch_dir).expanduser()
        daemon = Daemon(engine, watch_dir)
        daemon.run()
        return

    feeder = SourceFeeder(engine.gpu_queue, engine.chunk_size)
    engine.gpu_queue = feeder.queue

    if args.input:
        path = Path(args.input).expanduser()
        if path.is_dir():
            feeder.feed_directory(path)
        elif path.is_file():
            feeder.feed_file(path)
        else:
            print(f"  [ERROR] Input not found: {path}")
            sys.exit(1)
    else:
        feeder.feed_stdin()

    feeder.done()
    engine.finalize(pools)


def _run_benchmark(engine: GPUMetaprobeCompressionEngine):
    """Self-benchmark with structured compressible data across sizes."""
    print(f"\n  [BENCH] Running self-benchmark...")

    import zlib as bench_zlib
    patterns = [
        b"The quick brown fox jumps over the lazy dog. " * 5000,
        b"Lorem ipsum dolor sit amet consectetur adipiscing elit. " * 5000,
        b"ABCDEFGHIJKLMNOPQRSTUVWXYZ" * 1000,
        b"package main; import fmt; func main() { fmt.Println('hello'); } " * 2000,
    ]
    json_lines = []
    for i in range(2000):
        json_lines.append(
            b'{"id":' + str(i).encode() +
            b',"name":"item-' + str(i).encode() +
            b'","value":' + str(i * 1.5).encode() +
            b',"tags":["alpha","beta"]}'
        )
    patterns.append(b"\n".join(json_lines))
    text = b"".join(patterns)
    ref_ratio = len(text) / max(len(bench_zlib.compress(text, 9)), 1)
    print(f"  [BENCH] Reference: zlib-9 ratio = {ref_ratio:.2f}x on {len(text)} bytes")

    feeder = SourceFeeder(engine.gpu_queue, engine.chunk_size)
    engine.gpu_queue = feeder.queue
    pools = engine.start()
    feeder.feed_bytes(text, "bench-structured")
    feeder.done()
    stats = engine.finalize(pools)

    bench_path = engine.output_dir / "benchmark_results.json"
    with open(str(bench_path), "w") as f:
        json.dump({"benchmark": stats, "reference_zlib9_ratio": round(ref_ratio, 3),
                   "hardware": {"wgpu": HAS_WGPU, "cuda": HAS_CUDA,
                                "cpu_count": os.cpu_count(),
                                "surface": stats["gpu_surface"]}}, f, indent=2)
    print(f"\n  [BENCH] PIST-GCL ratio={stats['overall_ratio']:.2f}x vs zlib-9={ref_ratio:.2f}x")
    print(f"  [BENCH] Results: {bench_path}")


if __name__ == "__main__":
    main()