cuda/src/range_curate_cuda/_runtime/phase32a_gpu.py

"""Experimental Phase32A selective-Unicode CUDA classifier.

Derived from frozen phase29_gpu.py; requires separately generated frozen
Unicode tables and a new cubin exporting range_selective_unicode_classify.
No production/default integration is included. ASCII controls stay frozen.
Normalizes eligible rows completely; every unsupported row is CPU-owned.
The current native CPU engine and all nine compact fact fields are retained.
"""

from __future__ import annotations

from collections import deque
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
import subprocess
import tempfile
import time
from typing import Callable

import numpy as np


_SOURCE_SHA256_AT_IMPORT = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()


MAX_CANDIDATES = 128
MAX_CONJUNCTIONS = 32
MAX_DFA_BYTES = 32 * 1024 * 1024
FACT_NAMES = (
    "row_index", "article_status", "sp500", "nasdaq100", "russell2000",
    "market_wide", "observation_count", "native_candidate", "exact_candidate",
)


CUDA_SOURCE = r'''
#define MAX_CANDIDATES 128
#define MAX_CONJUNCTIONS 32

struct Candidate { int kind, index, start, end; };

__device__ __forceinline__ bool token(unsigned char c) {
    return (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_';
}
__device__ __forceinline__ bool exact_token(unsigned char c) {
    return token(c) || (c >= 'A' && c <= 'Z');
}
__device__ __forceinline__ unsigned char lower(unsigned char c) {
    return c >= 'A' && c <= 'Z' ? c + 32 : c;
}
__device__ __forceinline__ unsigned char normalize(unsigned char c) {
    c = lower(c);
    if (c == '-' || c == '.' || c == ',' || c == ':' || c == ';' ||
        c == '(' || c == ')' || c == '[' || c == ']' || c == '{' ||
        c == '}' || c == '/') return ' ';
    return c == '`' ? '\'' : c;
}
__device__ __forceinline__ bool space(unsigned char c) {
    return c == ' ' || (c >= 9 && c <= 13) || (c >= 0x1c && c <= 0x1f);
}

// All candidate identities are unique before the fixed-size buffer is used.
// An overflow transfers the whole row to CPU; no partial result is retained.
__device__ bool append_hit(
    int key, int start, int end, const int* key_offsets,
    const int* key_consumers, const int* consumers,
    Candidate* candidates, int* count, unsigned char* conjunctions
) {
    for (int i = key_offsets[key]; i < key_offsets[key + 1]; ++i) {
        const int* spec = consumers + key_consumers[i] * 4;
        if (spec[0] == 2) {
            conjunctions[spec[1]] |= (unsigned char)(1u << spec[2]);
            continue;
        }
        bool duplicate = false;
        for (int j = 0; j < *count; ++j) {
            Candidate prior = candidates[j];
            if (prior.kind == spec[0] && prior.index == spec[1] &&
                prior.start == start && prior.end == end) {
                duplicate = true; break;
            }
        }
        if (duplicate) continue;
        if (*count == MAX_CANDIDATES) return false;
        candidates[*count] = Candidate{spec[0], spec[1], start, end};
        ++*count;
    }
    return true;
}

__device__ int lower_bound_offset(const unsigned int* offsets, int n, int value) {
    int left = 0, right = n;
    while (left < right) {
        int middle = left + (right - left) / 2;
        if (offsets[middle] < (unsigned int)value) left = middle + 1;
        else right = middle;
    }
    return left;
}
__device__ bool contains_term(
    const unsigned char* text, int left, int right, int term,
    const unsigned char* term_bytes, const int* term_offsets
) {
    int first = term_offsets[term], length = term_offsets[term + 1] - first;
    if (!length || right - left < length) return false;
    for (int start = left; start + length <= right; ++start) {
        if (start > left && token(text[start - 1])) continue;
        if (start + length < right && token(text[start + length])) continue;
        int i = 0;
        while (i < length && text[start + i] == term_bytes[first + i]) ++i;
        if (i == length) return true;
    }
    return false;
}
__device__ void context_bounds(
    const unsigned char* normalized, const unsigned int* offsets, int norm_length,
    int source_length, int start, int end, int width, int* left, int* right
) {
    int source_left = start > width ? start - width : 0;
    int source_right = width > source_length - end ? source_length : end + width;
    *left = lower_bound_offset(offsets, norm_length, source_left);
    *right = lower_bound_offset(offsets, norm_length, source_right);
    // Match Phase24's borrowed-window treatment of collapsed leading spaces.
    if (*left < *right && normalized[*left] == ' ') ++*left;
}
__device__ bool alias_context_ok(
    int context, const int* contexts, const int* clauses, const int* term_ids,
    const unsigned char* term_bytes, const int* term_offsets,
    const unsigned char* normalized, const unsigned int* offsets, int norm_length,
    int source_length, int start, int end
) {
    if (context < 0) return true;
    const int* spec = contexts + context * 5;
    int left, right;
    context_bounds(normalized, offsets, norm_length, source_length,
                   start, end, spec[0], &left, &right);
    for (int i = 0; i < spec[4]; ++i)
        if (contains_term(normalized, left, right, term_ids[spec[3] + i],
                          term_bytes, term_offsets)) return false;
    for (int i = 0; i < spec[2]; ++i) {
        const int* clause = clauses + 2 * (spec[1] + i);
        bool accepted = true;
        for (int j = 0; j < clause[1]; ++j) {
            if (!contains_term(normalized, left, right, term_ids[clause[0] + j],
                               term_bytes, term_offsets)) {
                accepted = false; break;
            }
        }
        if (accepted) return true;
    }
    return false;
}


// Strict UTF-8 scalar decoder. Zero width rejects the entire row.
__device__ __forceinline__ int phase32a_decode(
    const unsigned char* raw, int length, int position, unsigned int* scalar
) {
    unsigned int first = raw[position];
    if (first < 0x80) { *scalar = first; return 1; }
    int width;
    unsigned int value, minimum;
    if (first >= 0xc2 && first <= 0xdf) { width = 2; value = first & 31; minimum = 0x80; }
    else if (first >= 0xe0 && first <= 0xef) { width = 3; value = first & 15; minimum = 0x800; }
    else if (first >= 0xf0 && first <= 0xf4) { width = 4; value = first & 7; minimum = 0x10000; }
    else return 0;
    if (width > length - position) return 0;
    for (int index = 1; index < width; ++index) {
        unsigned int next = raw[position + index];
        if ((next & 0xc0) != 0x80) return 0;
        value = (value << 6) | (next & 63);
    }
    if (value < minimum || value > 0x10ffff || (value >= 0xd800 && value <= 0xdfff)) return 0;
    *scalar = value;
    return width;
}

extern "C" __global__ void range_selective_unicode_classify(
    const unsigned char* data, const long long* row_offsets,
    const long long* timestamps, int rows, int max_row_bytes,
    const unsigned int* transitions, const unsigned short* classes,
    const int* dfa_meta, const int* output_offsets, const int* output_keys,
    const int* key_lengths, const int* key_offsets, const int* key_consumers,
    const int* consumers, const long long* aliases, const int* lexical,
    const int* contexts, const int* clauses, const int* term_ids,
    const unsigned char* term_bytes, const int* term_offsets,
    const unsigned short* unicode_directory, const unsigned int* unicode_pages, int conjunction_count,
    unsigned char* normalized_data, unsigned int* source_offsets,
    Candidate* all_candidates, unsigned char* codes, unsigned int* facts
) {
    int row = blockIdx.x * blockDim.x + threadIdx.x;
    if (row >= rows) return;
    unsigned int* fact = facts + (long long)row * 8;
    for (int i = 0; i < 8; ++i) fact[i] = 0;
    codes[row] = 0; // 2=subset reject, 3=candidate overflow, 4=oversize, 5=invalid UTF-8, 6=normalization capacity.
    long long base = row_offsets[row], raw_length = row_offsets[row + 1] - base;
    if (raw_length > max_row_bytes) { codes[row] = 4; return; }
    int length = (int)raw_length;
    const unsigned char* raw = data + base;
    unsigned char* normalized = normalized_data + base;
    unsigned int* offsets = source_offsets + base;
    int norm_length = 0, source_length = 0;
    bool pending_space = false;
    for (int i = 0; i < length;) {
        unsigned int scalar;
        int width = phase32a_decode(raw, length, i, &scalar);
        if (!width) { codes[row] = 5; return; }
        unsigned int packed;
        if (scalar < 128) packed = 2u | ((unsigned int)normalize((unsigned char)scalar) << 3);
        else {
            unsigned int page = unicode_directory[scalar >> 8];
            packed = unicode_pages[page * 256 + (scalar & 255)];
            if (!(packed & 7)) { codes[row] = 2; return; }
        }
        int unit_count = (packed & 7) - 1;
        // The table proves unit_count <= source UTF-8 width. Guard corruption
        // or a future predicate change; never write outside this row's arena.
        if (unit_count > width || unit_count > 4) { codes[row] = 6; return; }
        for (int unit = 0; unit < unit_count; ++unit) {
            unsigned char c = (packed >> (3 + 7 * unit)) & 127;
            if (space(c)) { pending_space = norm_length != 0; continue; }
            if (pending_space) {
                if (norm_length >= length) { codes[row] = 6; return; }
                normalized[norm_length] = ' '; offsets[norm_length++] = source_length;
                pending_space = false;
            }
            if (norm_length >= length) { codes[row] = 6; return; }
            normalized[norm_length] = c; offsets[norm_length++] = source_length;
        }
        i += width;
        ++source_length;
    }

    Candidate* candidates = all_candidates + (long long)row * MAX_CANDIDATES;
    int count = 0;
    unsigned char conjunctions[MAX_CONJUNCTIONS] = {0};
    bool native_candidate = false, exact_candidate = false;
    // Stream 0: normalized. Stream 1: raw case-sensitive. Stream 2: raw ASCII fold.
    for (int stream = 0; stream < 3; ++stream) {
        const int* meta = dfa_meta + stream * 4;
        if (meta[3] == 0) continue;
        const unsigned char* text = stream == 0 ? normalized : raw;
        int n = stream == 0 ? norm_length : length;
        unsigned int state = 0;
        int raw_source_position = -1;
        for (int pos = 0; pos < n; ++pos) {
            // Validated UTF-8; every lead/ASCII byte starts one original scalar.
            if (stream != 0 && (text[pos] & 0xc0) != 0x80) ++raw_source_position;
            unsigned char c = stream == 2 ? lower(text[pos]) : text[pos];
            state = transitions[meta[0] + state * meta[1] + classes[stream * 256 + c]];
            int first = output_offsets[meta[2] + state];
            int last = output_offsets[meta[2] + state + 1];
            for (int out = first; out < last; ++out) {
                int key = output_keys[out];
                int start = pos + 1 - key_lengths[key], end = pos + 1;
                bool left_ok = start == 0 || !(stream == 0 ? token(text[start - 1]) : exact_token(text[start - 1]));
                bool right_ok = end == n || !(stream == 0 ? token(text[end]) : exact_token(text[end]));
                if (!left_ok || !right_ok) continue;
                if (stream == 0) native_candidate = true;
                else exact_candidate = true;
                // Raw keys remain ASCII: a hit spans key_lengths[key] scalars.
                // Compatibility mappings never enter either raw matching stream.
                int original_end = stream == 0 ? offsets[end - 1] + 1 : raw_source_position + 1;
                int original_start = stream == 0 ? offsets[start] : original_end - key_lengths[key];
                if (!append_hit(key, original_start, original_end, key_offsets,
                                key_consumers, consumers, candidates, &count, conjunctions)) {
                    codes[row] = 3; return;
                }
            }
        }
    }

    bool accepted_nonsemantic = false, accepted_evidence = false;
    bool unresolved_exposure = false, unresolved_event = false;
    unsigned int supported = 0, observations = 0;
    for (int i = 0; i < count; ++i) {
        Candidate candidate = candidates[i];
        if (candidate.kind == 0) {
            const long long* alias = aliases + candidate.index * 8;
            bool longest = true;
            for (int j = 0; j < count; ++j) {
                Candidate other = candidates[j];
                if (other.kind == 0 && other.start == candidate.start &&
                    aliases[other.index * 8] == alias[0] && other.end > candidate.end) {
                    longest = false; break;
                }
            }
            if (!longest) continue;
            ++observations; // Retained observations count before temporal/context checks.
            if (timestamps[row] < alias[1] || (alias[3] && timestamps[row] >= alias[2])) continue;
            if (!alias_context_ok((int)alias[7], contexts, clauses, term_ids,
                                  term_bytes, term_offsets, normalized, offsets,
                                  norm_length, source_length, candidate.start, candidate.end)) continue;
            accepted_nonsemantic = true;
            if (alias[4] != -1 && (alias[5] == 0 || alias[5] == 1)) {
                accepted_evidence = true; supported |= 1u << alias[4];
            } else if (alias[6]) unresolved_exposure = true;
        } else {
            const int* rule = lexical + candidate.index * 7;
            ++observations;
            bool context_ok = rule[2] == 0;
            if (!context_ok) {
                int left, right;
                context_bounds(normalized, offsets, norm_length, source_length,
                               candidate.start, candidate.end, rule[0], &left, &right);
                for (int j = 0; j < rule[2]; ++j) {
                    if (contains_term(normalized, left, right, term_ids[rule[1] + j],
                                      term_bytes, term_offsets)) { context_ok = true; break; }
                }
            }
            if (rule[3]) {
                if (rule[4]) { accepted_evidence = true; supported |= 8; }
                else { if (rule[5]) unresolved_exposure = true; if (rule[6]) unresolved_event = true; }
            } else {
                if (context_ok) accepted_nonsemantic = true;
                if (rule[4] && context_ok) { accepted_evidence = true; supported |= 8; }
                else if (rule[5]) unresolved_exposure = true;
            }
        }
    }
    for (int i = 0; i < conjunction_count; ++i) {
        if (conjunctions[i] == 15) { ++observations; accepted_evidence = true; supported |= 8; }
    }
    if (!observations) return;
    fact[0] = accepted_evidence || accepted_nonsemantic || unresolved_event ? 0 : 2;
    for (int family = 0; family < 4; ++family)
        fact[family + 1] = supported & (1u << family) ? 0 :
            (family != 3 && !supported && unresolved_exposure ? 1 : 2);
    fact[5] = observations; fact[6] = native_candidate; fact[7] = exact_candidate;
    codes[row] = 1;
}
'''


def build_cubin(output_path, *, nvcc="nvcc", architecture="sm_75", timeout=1200):
    """Explicit maintainer build; never called by classifier construction/execution."""
    if not architecture.startswith("sm_") or not architecture[3:].isdigit():
        raise ValueError("architecture must name an explicit SM, for example sm_75")
    target = Path(output_path).resolve()
    if target.exists():
        raise FileExistsError(f"choose a fresh output path: {target}")
    target.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="range-phase32a-") as temporary:
        source = Path(temporary) / "range_phase32a.cu"
        source.write_text(CUDA_SOURCE, encoding="utf-8")
        command = [str(nvcc), "--cubin", "-O3", "--std=c++14",
                   f"-arch={architecture}", str(source), "-o", str(target)]
        completed = subprocess.run(command, check=True, capture_output=True,
                                   text=True, timeout=timeout)
    return {"cubin": str(target), "architecture": architecture,
            "cubin_sha256": hashlib.sha256(target.read_bytes()).hexdigest(),
            "cuda_source_sha256": hashlib.sha256(CUDA_SOURCE.encode()).hexdigest(),
            "compiler_stdout": completed.stdout, "compiler_stderr": completed.stderr}


class UnsupportedGpuTaxonomy(ValueError):
    """The experimental kernel cannot represent this valid CPU configuration."""


def _ascii(value):
    if not isinstance(value, str) or not value.isascii():
        raise UnsupportedGpuTaxonomy("GPU experiment requires ASCII pattern/context literals")
    if not value:
        raise UnsupportedGpuTaxonomy("GPU experiment does not support empty literals")
    return value.encode("ascii")


def _dfa(patterns):
    """Failure-completed byte-class DFA; terminal outputs preserve pattern IDs."""
    children, outputs = [{}], [set()]
    alphabet = sorted({byte for pattern, _key in patterns for byte in pattern})
    representatives = list(alphabet)
    absent = next((byte for byte in range(256) if byte not in alphabet), None)
    if absent is not None:
        representatives.append(absent)
    if not representatives:
        representatives = [0]
    class_map = np.full(256, len(representatives) - 1, dtype=np.uint16)
    for index, byte in enumerate(alphabet):
        class_map[byte] = index
    for pattern, key in patterns:
        state = 0
        for byte in pattern:
            if byte not in children[state]:
                children[state][byte] = len(children)
                children.append({})
                outputs.append(set())
            state = children[state][byte]
        outputs[state].add(key)
    if len(children) * len(representatives) * 4 > MAX_DFA_BYTES:
        raise UnsupportedGpuTaxonomy("GPU DFA exceeds the 32 MiB experiment limit")
    failure = [0] * len(children)
    queue = deque(children[0].values())
    order = []
    while queue:
        state = queue.popleft()
        order.append(state)
        for byte, child in children[state].items():
            queue.append(child)
            fallback = failure[state]
            while fallback and byte not in children[fallback]:
                fallback = failure[fallback]
            failure[child] = children[fallback].get(byte, 0)
            outputs[child].update(outputs[failure[child]])
    dense = np.zeros((len(children), len(representatives)), dtype=np.uint32)
    for index, byte in enumerate(representatives):
        dense[0, index] = children[0].get(byte, 0)
    for state in order:
        for index, byte in enumerate(representatives):
            dense[state, index] = children[state].get(byte, dense[failure[state], index])
    output_ids, offsets = [], [0]
    for values in outputs:
        output_ids.extend(sorted(values))
        offsets.append(len(output_ids))
    return dense, class_map, offsets, output_ids


def _pack_specs(specs):
    patterns, pattern_consumers, consumers, exact, aliases, lexical, contexts, conjunctions = specs
    if len(conjunctions) > MAX_CONJUNCTIONS:
        raise UnsupportedGpuTaxonomy(f"GPU supports at most {MAX_CONJUNCTIONS} conjunctions")
    if len(patterns) > 512 or sum(len(value) for value in patterns) > 16384:
        raise UnsupportedGpuTaxonomy("GPU normalized-pattern budget exceeded")
    consumers = [tuple(map(int, value)) for value in consumers]
    key_consumers = [list(map(int, value)) for value in pattern_consumers]
    keys = [_ascii(value) for value in patterns]
    streams = [[(value, index) for index, value in enumerate(keys)], [], []]
    raw_keys = {}
    for value, mode, kind, index in exact:
        encoded = _ascii(value)
        if mode == "case_sensitive":
            stream = 1
        elif mode == "python_ignorecase_ascii_literal":
            stream, encoded = 2, encoded.lower()
        else:
            raise UnsupportedGpuTaxonomy(f"unsupported exact mode: {mode}")
        identity = (stream, encoded)
        if identity not in raw_keys:
            key = len(keys)
            raw_keys[identity] = key
            keys.append(encoded)
            key_consumers.append([])
            streams[stream].append((encoded, key))
        consumer = len(consumers)
        consumers.append((int(kind), int(index), 0, 0))
        key_consumers[raw_keys[identity]].append(consumer)
    # Validate every device index/bit operation before copying the config.
    for kind, index, field, phrase in consumers:
        if kind == 0 and 0 <= index < len(aliases):
            continue
        if kind == 1 and 0 <= index < len(lexical):
            continue
        if (kind == 2 and 0 <= index < len(conjunctions) and 0 <= field < 4
                and 0 <= phrase < conjunctions[index][field]):
            continue
        raise UnsupportedGpuTaxonomy("unsupported consumer index/empty conjunction field")
    flat_consumers, consumer_offsets = [], [0]
    for values in key_consumers:
        flat_consumers.extend(values)
        consumer_offsets.append(len(flat_consumers))
    transitions, classes, dfa_meta, output_offsets, output_keys = [], [], [], [], []
    for stream in streams:
        dense, class_map, offsets, ids = _dfa(stream)
        dfa_meta.extend((len(transitions), dense.shape[1], len(output_offsets), len(stream)))
        transitions.extend(dense.reshape(-1).tolist())
        classes.extend(class_map.tolist())
        output_offsets.extend(len(output_keys) + value for value in offsets)
        output_keys.extend(ids)
    if len(transitions) * 4 > MAX_DFA_BYTES:
        raise UnsupportedGpuTaxonomy("combined GPU DFAs exceed 32 MiB")

    term_bytes, term_offsets, term_ids, clause_specs = bytearray(), [0], [], []
    term_map = {}

    def term(value):
        encoded = _ascii(value)
        if encoded not in term_map:
            term_map[encoded] = len(term_offsets) - 1
            term_bytes.extend(encoded)
            term_offsets.append(len(term_bytes))
        return term_map[encoded]

    def term_list(values):
        start = len(term_ids)
        term_ids.extend(term(value) for value in values)
        return start, len(values)

    context_specs = []
    for window, accepts, rejects in contexts:
        first_clause = len(clause_specs) // 2
        for clause in accepts:
            clause_specs.extend(term_list(clause))
        reject_start, reject_count = term_list(rejects)
        context_specs.extend((window, first_clause, len(accepts), reject_start, reject_count))
    lexical_specs = []
    for window, terms, event, market_wide, unresolved, family in lexical:
        first, count = term_list(terms)
        lexical_specs.extend((window, first, count, event, market_wide, unresolved, family))
    alias_specs = []
    for group, start, end, family, kind, unresolved, context in aliases:
        if family not in (-1, 0, 1, 2, 3) or (context is not None and not 0 <= context < len(contexts)):
            raise UnsupportedGpuTaxonomy("unsupported alias family/context")
        alias_specs.extend((group, start, end or 0, end is not None, family,
                            kind, unresolved, -1 if context is None else context))
    if any(window < 0 or window > 2**31 - 1 for window, *_ in (*contexts, *lexical)):
        raise UnsupportedGpuTaxonomy("GPU context window must fit a nonnegative int32")
    arrays = {
        "transitions": np.asarray(transitions, dtype=np.uint32),
        "classes": np.asarray(classes, dtype=np.uint16),
        "dfa_meta": np.asarray(dfa_meta, dtype=np.int32),
        "output_offsets": np.asarray(output_offsets, dtype=np.int32),
        "output_keys": np.asarray(output_keys, dtype=np.int32),
        "key_lengths": np.asarray([len(value) for value in keys], dtype=np.int32),
        "key_offsets": np.asarray(consumer_offsets, dtype=np.int32),
        "key_consumers": np.asarray(flat_consumers, dtype=np.int32),
        "consumers": np.asarray(consumers, dtype=np.int32).reshape(-1),
        "aliases": np.asarray(alias_specs, dtype=np.int64),
        "lexical": np.asarray(lexical_specs, dtype=np.int32),
        "contexts": np.asarray(context_specs, dtype=np.int32),
        "clauses": np.asarray(clause_specs, dtype=np.int32),
        "term_ids": np.asarray(term_ids, dtype=np.int32),
        "term_bytes": np.frombuffer(bytes(term_bytes), dtype=np.uint8),
        "term_offsets": np.asarray(term_offsets, dtype=np.int32),
    }
    return arrays, len(conjunctions)


def _utf8_buffers(array):
    """Same Arrow buffer convention as Phase18, without importing its engine."""
    import pyarrow as pa
    if not (pa.types.is_string(array.type) or pa.types.is_large_string(array.type)):
        raise TypeError("text must be Arrow string or large_string")
    if array.null_count:
        raise ValueError("text must not contain nulls")
    width = 8 if pa.types.is_large_string(array.type) else 4
    offsets = np.frombuffer(array.buffers()[1], dtype=np.int64 if width == 8 else np.int32,
                            count=len(array) + 1, offset=array.offset * width).astype(np.int64)
    start, end = int(offsets[0]), int(offsets[-1])
    buffer = array.buffers()[2]
    data = (np.frombuffer(buffer, dtype=np.uint8, count=end - start, offset=start)
            if buffer is not None else np.empty(0, dtype=np.uint8))
    offsets -= start
    return data, offsets


def _timestamps(array):
    import pyarrow as pa
    if not pa.types.is_timestamp(array.type) or array.type.tz is None:
        raise ValueError("timestamp must be a timezone-aware Arrow timestamp")
    if array.null_count:
        raise ValueError("timestamp must not contain nulls")
    values = array.cast(pa.int64()).to_numpy(zero_copy_only=False)
    unit = array.type.unit
    if unit == "us":
        return np.ascontiguousarray(values)
    if unit == "ns":
        # Rust signed integer division truncates toward zero, not minus infinity.
        return values // 1000 + ((values < 0) & (values % 1000 != 0)).astype(np.int64)
    multiplier = 1000 if unit == "ms" else 1_000_000
    limit = np.iinfo(np.int64)
    if len(values) and (int(values.min()) < -((-int(limit.min)) // multiplier)
                        or int(values.max()) > int(limit.max) // multiplier):
        raise ValueError("timestamp value overflows microseconds")
    return values * multiplier


def _fact_schema():
    import pyarrow as pa
    types = (pa.uint64(), *(pa.uint8() for _ in range(5)), pa.uint32(), pa.bool_(), pa.bool_())
    return pa.schema([pa.field(name, dtype, nullable=False) for name, dtype in zip(FACT_NAMES, types)])


def _materialize(value):
    import pyarrow as pa
    if hasattr(value, "to_pyarrow"):
        value = value.to_pyarrow()
    if not isinstance(value, (pa.Table, pa.RecordBatch)):
        value = pa.record_batch(value)
    return value


def _record_interval(target, stage, lane, started, ended):
    """Append primitive host-clock timestamps; diagnostic callers only."""
    origin = target["_origin"]
    target["intervals"].append({"stage": stage, "lane": lane,
                                "start_seconds": started - origin,
                                "end_seconds": ended - origin})


@dataclass
class GpuScreenResult:
    row_count: int
    facts: object
    profile: dict

    def to_pyarrow(self):
        return self.facts



PHASE32A_PREDICATE_ID = "range-u13-ascii-output-no-byte-growth-exclude-raw4-v1"
PHASE32A_SOURCE_SHA256 = {
    "unicode_index.bin": "fe152ac184aa77c40d0f040ba725140b17d845bae9fae255b2b3d3c43bb79c8f",
    "unicode_pool.bin": "109a8da1563408ad1da82919a5dfa1ab9fa18870f0f4d22e2818857b5a4fc30f",
    "python_whitespace.bin": "6c314375b6c075589b515139e50068d72c0cbd7daf4246e555a9dcfc9748ecc9",
}


def _load_phase32a_tables(directory):
    """Validate an explicitly built table artifact before any GPU operation."""
    directory = Path(directory).resolve()
    manifest_path = directory / "manifest.json"
    if manifest_path.stat().st_size > 65536:
        raise ValueError("Phase32A manifest exceeds the bounded metadata size")
    manifest_bytes = manifest_path.read_bytes()
    manifest = json.loads(manifest_bytes)
    if (manifest.get("schema") != 1 or manifest.get("predicate_id") != PHASE32A_PREDICATE_ID
            or manifest.get("unicode") != "13.0.0"
            or manifest.get("source_sha256") != PHASE32A_SOURCE_SHA256
            or manifest.get("raw_excluded_scalars") != [0x0130, 0x0131, 0x017F, 0x212A]
            or manifest.get("page_shift") != 8 or manifest.get("max_output_units") != 4):
        raise ValueError("wrong Phase32A Unicode table identity")
    names = ("unicode_ascii_directory.bin", "unicode_ascii_pages.bin")
    if set(manifest.get("payloads", {})) != set(names):
        raise ValueError("unexpected Phase32A payload manifest entries")
    pages = manifest["page_count"]
    if not isinstance(pages, int) or isinstance(pages, bool) or not 0 < pages <= 4353:
        raise ValueError("wrong Phase32A page count")
    expected_sizes = {names[0]: 4352 * 2, names[1]: pages * 256 * 4}
    payloads = {}
    for name in names:
        if (directory / name).stat().st_size != expected_sizes[name]:
            raise ValueError("wrong bounded Phase32A payload size: " + name)
        data = (directory / name).read_bytes()
        identity = manifest["payloads"][name]
        if len(data) != identity["bytes"] or hashlib.sha256(data).hexdigest() != identity["sha256"]:
            raise ValueError("Phase32A table payload mismatch: " + name)
        payloads[name] = data
    if (not isinstance(pages, int) or isinstance(pages, bool) or not 0 < pages <= 4353
            or len(payloads[names[0]]) != 4352 * 2 or len(payloads[names[1]]) != pages * 256 * 4):
        raise ValueError("wrong Phase32A page dimensions")
    directory_array = np.frombuffer(payloads[names[0]], dtype="<u2")
    pages_array = np.frombuffer(payloads[names[1]], dtype="<u4")
    if int(directory_array.max()) >= pages or np.any(pages_array[:256]):
        raise ValueError("invalid Phase32A page directory / unsupported page")
    # Kernel code depends on this exclusion for original-text exact matching.
    for scalar in (0x0130, 0x0131, 0x017F, 0x212A):
        if pages_array[int(directory_array[scalar >> 8]) * 256 + (scalar & 255)] != 0:
            raise ValueError("raw-ignorecase exception unexpectedly eligible")
    return ({"unicode_directory": directory_array, "unicode_pages": pages_array},
            {"predicate_id": PHASE32A_PREDICATE_ID,
             "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
             "payloads": manifest["payloads"], "source_sha256": PHASE32A_SOURCE_SHA256})


class ExactSelectiveUnicodeCudaClassifier:
    """Experimental screen_arrow-compatible adapter; no global backend mutation."""

    def __init__(self, cpu_engine, cubin_path, *, masked_cpu: Callable, unicode_tables,
                 batch_bytes=64 * 1024 * 1024, batch_rows=65536, max_row_bytes=65536,
                 overlap_cpu=False, diagnostics=False):
        from range_curate._core.accelerated.compact import _build_native_specs
        for name, value in (("batch_bytes", batch_bytes), ("batch_rows", batch_rows),
                            ("max_row_bytes", max_row_bytes)):
            if isinstance(value, bool) or not isinstance(value, int) or not 0 < value < 2**31:
                raise ValueError(f"{name} must be a positive int32")
        if max_row_bytes > batch_bytes:
            raise ValueError("max_row_bytes must not exceed batch_bytes")
        if not callable(masked_cpu):
            raise TypeError("masked_cpu must be the current native masked scanner callable")
        if not isinstance(overlap_cpu, bool):
            raise TypeError("overlap_cpu must be a boolean")
        if not isinstance(diagnostics, bool):
            raise TypeError("diagnostics must be a boolean")
        self.cpu_engine, self.masked_cpu = cpu_engine, masked_cpu
        self.cubin_path = Path(cubin_path).resolve()
        self.batch_bytes, self.batch_rows, self.max_row_bytes = batch_bytes, batch_rows, max_row_bytes
        self.overlap_cpu = overlap_cpu
        self.diagnostics = diagnostics
        self._unicode_arrays, self._unicode_identity = _load_phase32a_tables(unicode_tables)
        self.unsupported_reason = None
        self._arrays, self._conjunction_count = {}, 0
        try:
            self._arrays, self._conjunction_count = _pack_specs(_build_native_specs(cpu_engine._screening_indicator))
        except UnsupportedGpuTaxonomy as exc:
            self.unsupported_reason = str(exc)
        if not self.unsupported_reason:
            self._arrays.update(self._unicode_arrays)
        self.cp = self._kernel = None
        self._device_config, self._workspace = {}, {}
        self._capacity_bytes = self._capacity_rows = 0
        self._cubin_sha256 = None
        self._cuda_source_sha256 = hashlib.sha256(CUDA_SOURCE.encode()).hexdigest()
        self._kernel_attributes = None
        self.last_profile = {}

    @property
    def backend_report(self):
        return {"backend": "cpu" if self.unsupported_reason else "experimental_selective_unicode_cuda",
                "unsupported_taxonomy_reason": self.unsupported_reason,
                "unicode_table_identity": self._unicode_identity,
                "unicode_fallback_definition": "outside verified subset; original raw semantics retained",
                "cubin": str(self.cubin_path), "batch_bytes": self.batch_bytes,
                "batch_rows": self.batch_rows, "max_row_bytes": self.max_row_bytes,
                "candidate_limit": MAX_CANDIDATES, "overlap_cpu": self.overlap_cpu,
                "execution": "cpu_gpu_chunk_pipeline" if self.overlap_cpu else "synchronous",
                "component_times_may_overlap": self.overlap_cpu,
                "cubin_sha256": self._cubin_sha256,
                "cuda_source_sha256": self._cuda_source_sha256,
                "kernel_attributes": self._kernel_attributes}

    def _initialize(self):
        if self.cp is not None:
            return
        if not self.cubin_path.is_file():
            raise FileNotFoundError("Build the experimental cubin explicitly before running CUDA: " + str(self.cubin_path))
        import cupy as cp
        module = cp.RawModule(path=str(self.cubin_path))
        kernel = module.get_function("range_selective_unicode_classify")
        # Keep load-time metadata: later reports must not describe a replacement
        # file at the same path or rehash source text inside screening timings.
        loaded_cubin_sha256 = hashlib.sha256(self.cubin_path.read_bytes()).hexdigest()
        attributes = kernel.attributes
        kernel_attributes = {
            name: int(attributes[name]) if attributes.get(name) is not None else None
            for name in ("num_regs", "local_size_bytes", "shared_size_bytes")
        }
        config = {name: cp.asarray(value if value.size else np.zeros(1, dtype=value.dtype))
                  for name, value in self._arrays.items()}
        cp.cuda.Stream.null.synchronize()
        self.cp, self._module, self._kernel, self._device_config = cp, module, kernel, config
        self._cubin_sha256 = loaded_cubin_sha256
        self._kernel_attributes = kernel_attributes

    def diagnostic_unicode_shadow(self, data, *, enabled=False, text_column="text"):
        """Derived host replay, explicitly requested AFTER diagnostic timing.

        This method is never called by screen_arrow or its CUDA stages. It does
        not measure device instructions, memory transactions or service time.
        Input must satisfy the normal valid-UTF-8 Arrow contract. It retains at
        most one bounded row's decoded text and returns aggregate numbers only.
        """
        if enabled is not True:
            raise ValueError("Unicode host shadow requires explicit enabled=True outside timing")
        import pyarrow as pa
        if not isinstance(data, (pa.Table, pa.RecordBatch)):
            raise TypeError("Unicode host shadow accepts materialized Arrow input")
        if self.unsupported_reason:
            return {"scope": "derived host model; no device counters", "applicable": False,
                    "reason": "unsupported taxonomy routes the complete call to CPU"}
        started = time.perf_counter_ns()
        counts = dict.fromkeys(("input_rows", "input_bytes", "modeled_decoded_bytes",
            "modeled_decoded_scalars", "modeled_supported_scalars", "modeled_prefix_output_units",
            "modeled_completed_rows", "modeled_completed_input_bytes", "modeled_completed_source_scalars",
            "modeled_completed_output_units", "modeled_unsupported_rows", "modeled_unsupported_row_bytes",
            "modeled_oversized_rows", "modeled_oversized_row_bytes"), 0)
        mapping_units = {str(value): 0 for value in range(5)}
        expansion_rows = {name: 0 for name in ("zero_output", "up_to_1", "up_to_2", "up_to_3", "up_to_4")}
        directory, pages = self._unicode_arrays["unicode_directory"], self._unicode_arrays["unicode_pages"]
        for batch, _, _ in self._chunks(data, text_column):
            raw, offsets = _utf8_buffers(batch.column(batch.schema.get_field_index(text_column)))
            for left, right in zip(offsets[:-1], offsets[1:]):
                left, right = int(left), int(right)
                byte_count = right - left
                counts["input_rows"] += 1
                counts["input_bytes"] += byte_count
                if byte_count > self.max_row_bytes:
                    counts["modeled_oversized_rows"] += 1
                    counts["modeled_oversized_row_bytes"] += byte_count
                    continue
                text = raw[left:right].tobytes().decode("utf-8", errors="strict")
                output = source_scalars = 0
                pending = False
                complete = True
                for character in text:
                    scalar = ord(character)
                    width = 1 if scalar < 128 else 2 if scalar < 2048 else 3 if scalar < 65536 else 4
                    counts["modeled_decoded_bytes"] += width
                    counts["modeled_decoded_scalars"] += 1
                    if scalar < 128:
                        unit = scalar + 32 if 65 <= scalar <= 90 else scalar
                        if unit in b"-.,:;()[]{}/":
                            unit = 32
                        elif unit == 96:
                            unit = 39
                        packed = 2 | (unit << 3)
                    else:
                        packed = int(pages[int(directory[scalar >> 8]) * 256 + (scalar & 255)])
                        if not (packed & 7):
                            counts["modeled_unsupported_rows"] += 1
                            counts["modeled_unsupported_row_bytes"] += byte_count
                            complete = False
                            break
                    units = (packed & 7) - 1
                    if not 0 <= units <= min(width, 4):
                        raise ValueError("Unicode table violates the shadow model's capacity predicate")
                    mapping_units[str(units)] += 1
                    counts["modeled_supported_scalars"] += 1
                    for index in range(units):
                        unit = (packed >> (3 + 7 * index)) & 127
                        if unit == 32 or 9 <= unit <= 13 or 28 <= unit <= 31:
                            pending = output != 0
                            continue
                        if pending:
                            output += 1
                            pending = False
                        output += 1
                    source_scalars += 1
                if output > byte_count:
                    raise ValueError("Unicode shadow output exceeds the declared per-row arena")
                counts["modeled_prefix_output_units"] += output
                if complete:
                    counts["modeled_completed_rows"] += 1
                    counts["modeled_completed_input_bytes"] += byte_count
                    counts["modeled_completed_source_scalars"] += source_scalars
                    counts["modeled_completed_output_units"] += output
                    bucket = ("zero_output" if not output else "up_to_1" if output <= source_scalars else
                              "up_to_2" if output <= 2 * source_scalars else
                              "up_to_3" if output <= 3 * source_scalars else "up_to_4")
                    expansion_rows[bucket] += 1
        output = counts["modeled_completed_output_units"]
        return {"scope": "derived host normalization replay after screen_arrow timer/profiler range; not device-measured",
                "applicable": True, "performance_evidence": False, "counts": counts,
                "modeled_normalizer_unique_byte_positions_touched": counts["modeled_decoded_bytes"],
                "touch_definition": "valid UTF-8 positions through and including the first unsupported scalar; oversized rows touch zero; excludes raw matching/context/repeated loads/cache transactions",
                "actual_device_decoded_bytes": None, "actual_device_touched_bytes": None,
                "device_internal_service_seconds": None,
                "mapping_output_units_per_supported_scalar_histogram": mapping_units,
                "completed_row_output_units_per_source_scalar_histogram": expansion_rows,
                "modeled_completed_output_units_per_source_scalar": (
                    output / counts["modeled_completed_source_scalars"] if counts["modeled_completed_source_scalars"] else None),
                "modeled_completed_output_bytes_per_input_byte": (
                    output / counts["modeled_completed_input_bytes"] if counts["modeled_completed_input_bytes"] else None),
                "normalization_completion_note": "includes rows later assigned CPU fallback for candidate overflow; it is not final eligible-row coverage",
                "host_shadow_elapsed_ns": time.perf_counter_ns() - started,
                "unicode_table_identity": self._unicode_identity}

    def _reserve(self, byte_count, row_count):
        cp = self.cp
        if byte_count > self._capacity_bytes:
            self._capacity_bytes = max(byte_count, self._capacity_bytes)
            self._workspace.update(data=cp.empty(self._capacity_bytes, dtype=cp.uint8),
                                   normalized=cp.empty(self._capacity_bytes, dtype=cp.uint8),
                                   source_offsets=cp.empty(self._capacity_bytes, dtype=cp.uint32))
        if row_count > self._capacity_rows:
            self._capacity_rows = max(row_count, self._capacity_rows)
            self._workspace.update(offsets=cp.empty(self._capacity_rows + 1, dtype=cp.int64),
                                   timestamps=cp.empty(self._capacity_rows, dtype=cp.int64),
                                   candidates=cp.empty(self._capacity_rows * MAX_CANDIDATES * 4, dtype=cp.int32),
                                   codes=cp.empty(self._capacity_rows, dtype=cp.uint8),
                                   facts=cp.empty((self._capacity_rows, 8), dtype=cp.uint32))

    def _gpu_stage(self, batch, *, text_column, timestamp_column, diagnostic=None):
        """Complete all CUDA work and return independent, read-only host arrays."""
        cp = self.cp
        total_started = time.perf_counter()
        data, offsets = _utf8_buffers(batch.column(batch.schema.get_field_index(text_column)))
        if diagnostic is not None:
            boundary = time.perf_counter()
            _record_interval(diagnostic, "utf8_buffer_prepare", "main", total_started, boundary)
            timestamp_started = boundary
        epochs = _timestamps(batch.column(batch.schema.get_field_index(timestamp_column)))
        if diagnostic is not None:
            boundary = time.perf_counter()
            _record_interval(diagnostic, "timestamp_prepare", "main", timestamp_started, boundary)
            reserve_started = boundary
        rows, byte_count = batch.num_rows, data.nbytes
        self._reserve(max(1, byte_count), rows)
        buffers, config = self._workspace, self._device_config
        preparation = time.perf_counter() - total_started
        if diagnostic is not None:
            _record_interval(diagnostic, "workspace_reserve", "main", reserve_started,
                             total_started + preparation)
        started = time.perf_counter()
        if byte_count:
            buffers["data"][:byte_count].set(data)
        buffers["offsets"][:rows + 1].set(offsets)
        buffers["timestamps"][:rows].set(epochs)
        cp.cuda.Stream.null.synchronize()
        h2d = time.perf_counter() - started
        if diagnostic is not None:
            _record_interval(diagnostic, "h2d", "main", started, started + h2d)
        started = time.perf_counter()
        arguments = [buffers["data"], buffers["offsets"], buffers["timestamps"],
                     np.int32(rows), np.int32(self.max_row_bytes)]
        arguments.extend(config[name] for name in self._arrays)
        arguments.extend((np.int32(self._conjunction_count), buffers["normalized"],
                          buffers["source_offsets"], buffers["candidates"], buffers["codes"], buffers["facts"]))
        self._kernel(((rows + 127) // 128,), (128,), tuple(arguments))
        cp.cuda.Stream.null.synchronize()
        kernel = time.perf_counter() - started
        if diagnostic is not None:
            _record_interval(diagnostic, "kernel_host_interval", "main", started, started + kernel)
        started = time.perf_counter()
        codes = cp.asnumpy(buffers["codes"][:rows])
        # One fixed 32-byte result slot per row, not a row-by-pattern matrix.
        # Host compaction below retains only actual facts in the returned Arrow data.
        dense = cp.asnumpy(buffers["facts"][:rows])
        d2h = time.perf_counter() - started
        if diagnostic is not None:
            _record_interval(diagnostic, "d2h", "main", started, started + d2h)
            finalize_started = started + d2h
        # Neither an outstanding host task nor its Arrow input retains a device
        # workspace view. Reusing the single CUDA workspace for the next chunk
        # cannot overwrite this task's codes, facts or source-byte offsets.
        for values in (codes, dense, offsets):
            values.setflags(write=False)
        profile = {"rows": rows, "bytes": byte_count, "host_prepare_seconds": preparation,
                   "h2d_seconds": h2d, "kernel_seconds": kernel, "d2h_seconds": d2h}
        if diagnostic is not None:
            diagnostic["transfer_payload_bytes"] = {
                "h2d": {"text": int(byte_count), "row_offsets": int(offsets.nbytes),
                        "timestamps": int(epochs.nbytes)},
                "d2h": {"route_codes": int(codes.nbytes), "dense_facts": int(dense.nbytes)},
            }
            _record_interval(diagnostic, "gpu_host_finalize", "main", finalize_started,
                             time.perf_counter())
        return codes, dense, offsets, profile

    def _complete_gpu_chunk(self, batch, stage, *, workers, text_column, timestamp_column,
                            diagnostic=None):
        """CPU-only ownership and Arrow assembly; safe in the one host worker."""
        import pyarrow as pa
        codes, dense, offsets, gpu_profile = stage
        started = time.perf_counter()
        fallback_mask = codes >= 2
        fallback = None
        if np.any(fallback_mask):
            fallback = _materialize(self.masked_cpu(
                batch, fallback_mask.astype(np.uint8).tobytes(), workers=workers,
                text_column=text_column, timestamp_column=timestamp_column))
        fallback_seconds = time.perf_counter() - started
        if diagnostic is not None:
            _record_interval(diagnostic, "cpu_fallback", diagnostic["completion_lane"],
                             started, started + fallback_seconds)
        started = time.perf_counter()
        selected = np.flatnonzero(codes == 1)
        values = dense[selected]
        schema = _fact_schema()
        columns = [pa.array(selected, type=pa.uint64())]
        columns.extend(pa.array(values[:, index].astype(np.bool_) if index >= 6 else values[:, index],
                                type=schema.field(index + 1).type)
                       for index in range(8))
        gpu_facts = pa.RecordBatch.from_arrays(columns, schema=schema)
        tables = [pa.Table.from_batches([gpu_facts])]
        if fallback is not None:
            tables.append(pa.Table.from_batches([fallback]) if isinstance(fallback, pa.RecordBatch) else fallback)
        result = pa.concat_tables(tables).sort_by([("row_index", "ascending")])
        assembly = time.perf_counter() - started
        if diagnostic is not None:
            _record_interval(diagnostic, "chunk_assembly", diagnostic["completion_lane"],
                             started, started + assembly)
            accounting_started = started + assembly
        byte_lengths = np.diff(offsets)
        profile = {**gpu_profile, "gpu_completed_rows": int(np.count_nonzero(codes < 2)),
                   "unicode_rows": int(np.count_nonzero(codes == 2)),
                   "invalid_utf8_rows": int(np.count_nonzero(codes == 5)),
                   "normalization_capacity_rows": int(np.count_nonzero(codes == 6)),
                   "overflow_rows": int(np.count_nonzero(codes == 3)),
                   "oversized_rows": int(np.count_nonzero(codes == 4)),
                   "fallback_rows": int(np.count_nonzero(fallback_mask)),
                   "fallback_bytes": int(byte_lengths[fallback_mask].sum()),
                   "cpu_fallback_seconds": fallback_seconds, "assembly_seconds": assembly}
        if diagnostic is not None:
            diagnostic["fallback_bytes_by_reason"] = {
                "unicode": int(byte_lengths[codes == 2].sum()),
                "invalid_utf8": int(byte_lengths[codes == 5].sum()),
                "normalization_capacity": int(byte_lengths[codes == 6].sum()),
                "candidate_overflow": int(byte_lengths[codes == 3].sum()),
                "oversized": int(byte_lengths[codes == 4].sum()),
            }
            diagnostic["gpu_completed_bytes"] = int(byte_lengths[codes < 2].sum())
            diagnostic["measurements"] = profile.copy()
            _record_interval(diagnostic, "completion_accounting", diagnostic["completion_lane"],
                             accounting_started, time.perf_counter())
        return result, profile

    def _gpu_chunk(self, batch, *, workers, text_column, timestamp_column):
        """The original synchronous scheduling, retained as the default control."""
        stage = self._gpu_stage(batch, text_column=text_column, timestamp_column=timestamp_column)
        return self._complete_gpu_chunk(batch, stage, workers=workers, text_column=text_column,
                                        timestamp_column=timestamp_column)

    def _oversized_chunk(self, piece, byte_count, *, workers, text_column, timestamp_column,
                         diagnostic=None):
        import pyarrow as pa
        started = time.perf_counter()
        facts = _materialize(self.masked_cpu(piece, bytes([1]), workers=workers,
                                            text_column=text_column, timestamp_column=timestamp_column))
        facts = pa.Table.from_batches([facts]) if isinstance(facts, pa.RecordBatch) else facts
        profile = {"rows": 1, "bytes": byte_count, "oversized_rows": 1,
                   "fallback_rows": 1, "fallback_bytes": byte_count,
                   "cpu_fallback_seconds": time.perf_counter() - started}
        if diagnostic is not None:
            _record_interval(diagnostic, "cpu_oversized_fallback", diagnostic["completion_lane"],
                             started, started + profile["cpu_fallback_seconds"])
            diagnostic["fallback_bytes_by_reason"] = {
                "unicode": 0, "candidate_overflow": 0, "oversized": int(byte_count),
            }
            diagnostic["gpu_completed_bytes"] = 0
            diagnostic["transfer_payload_bytes"] = {
                "h2d": {"text": 0, "row_offsets": 0, "timestamps": 0},
                "d2h": {"route_codes": 0, "dense_facts": 0},
            }
            diagnostic["measurements"] = profile.copy()
            _record_interval(diagnostic, "completion_accounting", diagnostic["completion_lane"],
                             started + profile["cpu_fallback_seconds"], time.perf_counter())
        return facts, profile

    def _chunks(self, data, text_column):
        import pyarrow as pa
        batches = data.to_batches(max_chunksize=self.batch_rows) if isinstance(data, pa.Table) else [data]
        row_base = 0
        for batch in batches:
            _, offsets = _utf8_buffers(batch.column(batch.schema.get_field_index(text_column)))
            start = 0
            while start < batch.num_rows:
                end = min(batch.num_rows, start + self.batch_rows,
                          int(np.searchsorted(offsets, offsets[start] + self.batch_bytes, side="right")) - 1)
                end = max(start + 1, end)
                yield batch.slice(start, end - start), row_base + start, int(offsets[end] - offsets[start])
                start = end
            row_base += batch.num_rows

    def screen_arrow(self, data, *, workers=10, row_count=None, text_column="text", timestamp_column="timestamp"):
        import pyarrow as pa
        if isinstance(workers, bool) or not isinstance(workers, int) or workers < 1:
            raise ValueError("workers must be a positive integer")
        started = time.perf_counter()
        if not isinstance(data, (pa.Table, pa.RecordBatch)):
            raise TypeError("this experiment accepts a materialized Arrow Table or RecordBatch")
        if row_count is not None and (isinstance(row_count, bool) or not isinstance(row_count, int) or row_count < 0):
            raise ValueError("row_count must be a nonnegative integer or None")
        if row_count is not None and row_count != data.num_rows:
            raise ValueError("row_count differs from Arrow input")
        if self.diagnostics:
            return self._screen_arrow_diagnostics(
                data, workers=workers, text_column=text_column,
                timestamp_column=timestamp_column, started=started)
        if self.unsupported_reason:
            result = self.cpu_engine.screen_arrow(data, workers=workers, row_count=data.num_rows,
                                                   text_column=text_column, timestamp_column=timestamp_column)
            profile = {**self.backend_report, "rows": data.num_rows,
                       "total_seconds": time.perf_counter() - started}
            self.last_profile = profile
            return GpuScreenResult(data.num_rows, _materialize(result), profile)
        for name in (text_column, timestamp_column):
            if data.schema.get_field_index(name) < 0:
                raise ValueError(f"missing or ambiguous Arrow column: {name}")
        text_type = data.schema.field(text_column).type
        timestamp_type = data.schema.field(timestamp_column).type
        if not (pa.types.is_string(text_type) or pa.types.is_large_string(text_type)):
            raise TypeError("text must be Arrow string or large_string")
        if not pa.types.is_timestamp(timestamp_type) or timestamp_type.tz is None:
            raise ValueError("timestamp must be a timezone-aware Arrow timestamp")
        if data.column(text_column).null_count or data.column(timestamp_column).null_count:
            raise ValueError("text and timestamp must not contain nulls")
        self._initialize()
        output, aggregate = [], {}
        keywords = dict(workers=workers, text_column=text_column, timestamp_column=timestamp_column)

        def collect(completed, global_offset):
            facts, profile = completed
            if facts.num_rows:
                indices = facts.column("row_index").to_numpy().astype(np.uint64) + np.uint64(global_offset)
                output.append(facts.set_column(0, _fact_schema().field(0), pa.array(indices, type=pa.uint64())))
            for key, value in profile.items():
                aggregate[key] = aggregate.get(key, 0) + value

        if not self.overlap_cpu:
            for piece, global_offset, byte_count in self._chunks(data, text_column):
                if byte_count > self.batch_bytes:
                    completed = self._oversized_chunk(piece, byte_count, **keywords)
                else:
                    completed = self._gpu_chunk(piece, **keywords)
                collect(completed, global_offset)
        else:
            # At most one host completion is outstanding. The next GPU stage
            # uses the same device workspace while that task holds independent
            # host copies. No CUDA calls occur in the host worker.
            pending, pending_offset = None, 0
            with ThreadPoolExecutor(max_workers=1, thread_name_prefix="range-cuda-fallback") as executor:
                try:
                    for piece, global_offset, byte_count in self._chunks(data, text_column):
                        stage = None
                        if byte_count <= self.batch_bytes:
                            stage = self._gpu_stage(piece, text_column=text_column,
                                                    timestamp_column=timestamp_column)
                        if pending is not None:
                            collect(pending.result(), pending_offset)
                            pending = None
                        if stage is None:
                            pending = executor.submit(self._oversized_chunk, piece, byte_count, **keywords)
                        else:
                            pending = executor.submit(self._complete_gpu_chunk, piece, stage, **keywords)
                        pending_offset = global_offset
                    if pending is not None:
                        collect(pending.result(), pending_offset)
                        pending = None
                except BaseException as foreground_error:
                    # Drain the bounded host job before leaving this call. If
                    # GPU/input work and the CPU job both failed, retain both
                    # errors rather than abandoning the background exception.
                    if pending is not None:
                        try:
                            pending.result()
                        except BaseException as background_error:
                            if background_error is not foreground_error:
                                raise foreground_error from background_error
                    raise
        facts = pa.concat_tables(output).combine_chunks() if output else pa.Table.from_batches([], schema=_fact_schema())
        batch = (facts.to_batches(max_chunksize=max(1, facts.num_rows))[0] if facts.num_rows
                 else pa.RecordBatch.from_arrays([pa.array([], type=field.type) for field in facts.schema], schema=facts.schema))
        profile = {**self.backend_report, **aggregate, "rows": data.num_rows,
                   "fact_rows": batch.num_rows, "total_seconds": time.perf_counter() - started,
                   "device_workspace_bytes": sum(value.nbytes for value in self._workspace.values()),
                   "device_config_bytes": sum(value.nbytes for value in self._device_config.values())}
        self.last_profile = profile
        return GpuScreenResult(data.num_rows, batch, profile)

    def _screen_arrow_diagnostics(self, data, *, workers, text_column, timestamp_column, started):
        """Opt-in host timeline, kept out of the legacy scheduling/timing path.

        The kernel and synchronization are unchanged. Intervals describe host
        service or coordination, not CUDA-event durations. Each completion job
        owns its chunk record; the main thread reads it only after completion.
        No article text, Arrow buffers, futures or device views enter the report.
        """
        import pyarrow as pa
        call_trace = {"_origin": started, "intervals": []}
        diagnostic = {
            "schema_version": 1,
            "clock": "perf_counter_seconds_since_screen_arrow_start",
            "device_event_timings": False,
            "timing_scope": "extra_instrumented_call_not_headline_timing",
            "call_intervals": call_trace["intervals"],
            "chunks": [],
        }
        keywords = dict(workers=workers, text_column=text_column, timestamp_column=timestamp_column)
        if self.unsupported_reason:
            boundary = time.perf_counter()
            _record_interval(call_trace, "input_validation", "main", started, boundary)
            result = self.cpu_engine.screen_arrow(data, row_count=data.num_rows, **keywords)
            completed = time.perf_counter()
            _record_interval(call_trace, "unsupported_taxonomy_cpu", "main", boundary, completed)
            facts = _materialize(result)
            finalized = time.perf_counter()
            _record_interval(call_trace, "final_output_materialization", "main", completed, finalized)
            diagnostic["initialization_h2d_config_bytes"] = 0
            diagnostic["transfer_payload_bytes"] = {"h2d": 0, "d2h": 0, "initialization_h2d": 0}
            diagnostic["call_total_seconds"] = finalized - started
            diagnostic["unsupported_taxonomy_reason"] = self.unsupported_reason
            profile = {**self.backend_report, "rows": data.num_rows,
                       "total_seconds": finalized - started, "diagnostics": diagnostic}
            self.last_profile = profile
            return GpuScreenResult(data.num_rows, facts, profile)

        # Mirror the legacy validation before initializing any CUDA resource.
        for name in (text_column, timestamp_column):
            if data.schema.get_field_index(name) < 0:
                raise ValueError(f"missing or ambiguous Arrow column: {name}")
        text_type = data.schema.field(text_column).type
        timestamp_type = data.schema.field(timestamp_column).type
        if not (pa.types.is_string(text_type) or pa.types.is_large_string(text_type)):
            raise TypeError("text must be Arrow string or large_string")
        if not pa.types.is_timestamp(timestamp_type) or timestamp_type.tz is None:
            raise ValueError("timestamp must be a timezone-aware Arrow timestamp")
        if data.column(text_column).null_count or data.column(timestamp_column).null_count:
            raise ValueError("text and timestamp must not contain nulls")
        boundary = time.perf_counter()
        _record_interval(call_trace, "input_validation", "main", started, boundary)
        was_initialized = self.cp is not None
        self._initialize()
        initialized = time.perf_counter()
        _record_interval(call_trace, "initialization", "main", boundary, initialized)
        diagnostic["initialization_h2d_config_bytes"] = (
            0 if was_initialized else sum(int(value.nbytes) for value in self._device_config.values()))
        output, aggregate = [], {}

        def chunks():
            iterator = iter(self._chunks(data, text_column))
            while True:
                fetch_started = time.perf_counter()
                try:
                    piece, global_offset, byte_count = next(iterator)
                except StopIteration:
                    _record_interval(call_trace, "chunk_iterator_finish", "main", fetch_started,
                                     time.perf_counter())
                    return
                fetched = time.perf_counter()
                item = {"_origin": started, "intervals": [],
                        "chunk_index": len(diagnostic["chunks"]),
                        "row_index_start": int(global_offset), "rows": int(piece.num_rows),
                        "input_bytes": int(byte_count),
                        "route": "cpu_oversized" if byte_count > self.batch_bytes else "gpu_then_cpu",
                        "completion_lane": "cpu_completion" if self.overlap_cpu else "main"}
                _record_interval(item, "chunk_fetch", "main", fetch_started, fetched)
                diagnostic["chunks"].append(item)
                yield piece, global_offset, byte_count, item

        def collect(completed, global_offset, item):
            collect_started = time.perf_counter()
            facts, chunk_profile = completed
            if facts.num_rows:
                indices = facts.column("row_index").to_numpy().astype(np.uint64) + np.uint64(global_offset)
                output.append(facts.set_column(0, _fact_schema().field(0), pa.array(indices, type=pa.uint64())))
            for key, value in chunk_profile.items():
                aggregate[key] = aggregate.get(key, 0) + value
            item["fact_rows"] = int(facts.num_rows)
            _record_interval(item, "collect", "main", collect_started, time.perf_counter())

        def wait_and_collect(pending, global_offset, item):
            wait_started = time.perf_counter()
            completed = pending.result()
            _record_interval(item, "wait_for_completion", "main", wait_started, time.perf_counter())
            collect(completed, global_offset, item)

        if not self.overlap_cpu:
            for piece, global_offset, byte_count, item in chunks():
                if byte_count > self.batch_bytes:
                    completed = self._oversized_chunk(piece, byte_count, diagnostic=item, **keywords)
                else:
                    stage = self._gpu_stage(piece, text_column=text_column,
                                            timestamp_column=timestamp_column, diagnostic=item)
                    completed = self._complete_gpu_chunk(piece, stage, diagnostic=item, **keywords)
                collect(completed, global_offset, item)
        else:
            pending, pending_offset, pending_item = None, 0, None
            executor_started = time.perf_counter()
            with ThreadPoolExecutor(max_workers=1, thread_name_prefix="range-cuda-fallback") as executor:
                _record_interval(call_trace, "executor_setup", "main", executor_started, time.perf_counter())
                try:
                    for piece, global_offset, byte_count, item in chunks():
                        stage = None
                        if byte_count <= self.batch_bytes:
                            stage = self._gpu_stage(piece, text_column=text_column,
                                                    timestamp_column=timestamp_column, diagnostic=item)
                        if pending is not None:
                            wait_and_collect(pending, pending_offset, pending_item)
                            pending = None
                        submit_started = time.perf_counter()
                        item["submitted_seconds"] = submit_started - started
                        if stage is None:
                            pending = executor.submit(self._oversized_chunk, piece, byte_count,
                                                      diagnostic=item, **keywords)
                        else:
                            pending = executor.submit(self._complete_gpu_chunk, piece, stage,
                                                      diagnostic=item, **keywords)
                        # This main-lane interval is separate from the worker-owned record.
                        _record_interval(call_trace, "completion_submit", "main", submit_started,
                                         time.perf_counter())
                        pending_offset, pending_item = global_offset, item
                    if pending is not None:
                        wait_and_collect(pending, pending_offset, pending_item)
                        pending = None
                except BaseException as foreground_error:
                    if pending is not None:
                        try:
                            pending.result()
                        except BaseException as background_error:
                            if background_error is not foreground_error:
                                raise foreground_error from background_error
                    raise
                shutdown_started = time.perf_counter()
            _record_interval(call_trace, "executor_shutdown", "main", shutdown_started, time.perf_counter())

        finalize_started = time.perf_counter()
        facts = pa.concat_tables(output).combine_chunks() if output else pa.Table.from_batches([], schema=_fact_schema())
        batch = (facts.to_batches(max_chunksize=max(1, facts.num_rows))[0] if facts.num_rows
                 else pa.RecordBatch.from_arrays([pa.array([], type=field.type) for field in facts.schema], schema=facts.schema))
        finalized = time.perf_counter()
        _record_interval(call_trace, "final_output_assembly", "main", finalize_started, finalized)
        for item in diagnostic["chunks"]:
            del item["_origin"]
        diagnostic["fallback_bytes_by_reason"] = {
            reason: sum(item["fallback_bytes_by_reason"].get(reason, 0) for item in diagnostic["chunks"])
            for reason in ("unicode", "candidate_overflow", "oversized", "invalid_utf8", "normalization_capacity")}
        diagnostic["transfer_payload_bytes"] = {
            direction: sum(sum(item["transfer_payload_bytes"][direction].values())
                           for item in diagnostic["chunks"])
            for direction in ("h2d", "d2h")}
        diagnostic["transfer_payload_bytes"]["initialization_h2d"] = diagnostic["initialization_h2d_config_bytes"]
        completed = time.perf_counter()
        _record_interval(call_trace, "diagnostic_aggregation", "main", finalized, completed)
        diagnostic["call_total_seconds"] = completed - started
        profile = {**self.backend_report, **aggregate, "rows": data.num_rows,
                   "fact_rows": batch.num_rows, "total_seconds": completed - started,
                   "device_workspace_bytes": sum(value.nbytes for value in self._workspace.values()),
                   "device_config_bytes": sum(value.nbytes for value in self._device_config.values()),
                   "diagnostics": diagnostic}
        self.last_profile = profile
        return GpuScreenResult(data.num_rows, batch, profile)


def adversarial_inputs():
    """Small all-column parity inputs; no execution or GPU initialization on import."""
    texts = [
        "", "nothing matches here", "SPY espionage spy agency", "SPY ETF",
        "SPY shares S&P 500", "SPDR S&P 500 ETF Trust", "Invesco QQQ ETF",
        "Nasdaq-100", "Nasdaq 100", "rate-cut", "interest-rate cut",
        "PowerShares QQQ Trust, Series 1", "CPI consumer price index; GDP growth",
        "Fed", "Fed inflation", "SEC proposes rule for public companies",
        "SEC proposes rule", "U.S. Securities and Exchange Commission adopts regulation for all exchanges",
        "rate-cut; interest-rate cut; Federal Reserve", "federal\x1c\x1dreserve",
        "federal\x00reserve", "CPIx xCPI _CPI CPI_", "İ ſ K ı CPI inflation",
        "S&P 500 — Federal Reserve", "SPY ETF " + "é", "Federal Reserve " * 140,
        "x" * 65537, "ETF " + "x" * 126 + " SPY", "SPY " + "x" * 126 + " ETF",
    ]
    return [(text, timestamp) for timestamp in (
        "1990-01-01T00:00:00Z", "2018-06-04T03:59:59.999999Z", "2018-06-04T04:00:00Z",
    ) for text in texts]


def validate_adversarial_parity(backend, *, workers=10):
    """Explicit small GPU check, including both flags omitted by older digests.

    This helper is never called automatically. Run only after compiling/loading
    the candidate, before accepting any throughput measurement.
    """
    from datetime import datetime
    import pyarrow as pa
    cases = adversarial_inputs()
    summaries = []
    for text_type in (pa.string(), pa.large_string()):
        # Nonzero Arrow array offsets exercise sliced UTF-8 and timestamp buffers.
        texts = ["padding"] + [text for text, _ in cases] + ["padding"]
        timestamps = [datetime.fromisoformat("2020-01-01T00:00:00+00:00")]
        timestamps.extend(datetime.fromisoformat(value.replace("Z", "+00:00")) for _, value in cases)
        timestamps.append(timestamps[0])
        batch = pa.record_batch(
            [pa.array(texts, type=text_type), pa.array(timestamps, type=pa.timestamp("us", tz="UTC"))],
            names=["text", "timestamp"],
        ).slice(1, len(cases))
        expected = _materialize(backend.cpu_engine.screen_arrow(batch, workers=workers, row_count=batch.num_rows))
        observed = backend.screen_arrow(batch, workers=workers, row_count=batch.num_rows)
        expected = pa.Table.from_batches([expected]) if isinstance(expected, pa.RecordBatch) else expected
        actual = pa.Table.from_batches([observed.to_pyarrow()])
        for name in FACT_NAMES:
            if not expected.column(name).equals(actual.column(name)):
                raise AssertionError(f"Phase29 adversarial {text_type} parity failed: {name}")
        summaries.append({"text_type": str(text_type), "rows": batch.num_rows,
                          "fact_rows": actual.num_rows, "all_nine_columns_equal": True,
                          "profile": observed.profile})
    return summaries