#!/usr/bin/env python3 """ pipeline_core.py — Module-swappable six-stage search engine. Each stage is a Filter with a standard interface: input: List[Config] → output: List[Config] Stages can be swapped without rewriting the pipeline. All arithmetic is integer-based (Q16_16 raw where ratios needed). No floats. No native_decide. Pure Python stdlib. Usage: from pipeline_core import Pipeline, BraidStorm, TreeBraid, AngrySphinx, MultisurfacePacker, COUCHFilter, SidonFilter pipe = Pipeline([BraidStorm(k=8), TreeBraid(), AngrySphinx(budget=128), MultisurfacePacker(max_surfaces=64), COUCHFilter(threshold=49152), SidonFilter()]) result = pipe.run(labels=[1,2,4,8,16,32,64,128], S=128, moduli=[7,3,5,11,13,17,19,23,29]) To add a custom filter: class MyFilter(Filter): def apply(self, configs, ctx): # filter logic here return [c for c in configs if ...] @property def name(self): return "MyFilter" """ import sys, math, json, time, hashlib, random from abc import ABC, abstractmethod from dataclasses import dataclass, field from pathlib import Path from itertools import product from collections import Counter from typing import Any REPO_ROOT = Path(__file__).resolve().parent.parent ARTIFACTS_DIR = REPO_ROOT / ".openresearch" / "artifacts" ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True) # Q16_16 constants (no floats) Q16_ONE = 65536 Q16_THRESHOLD_COUCH = 49152 # 0.75 × 65536 Q16_SUBLEQ_SELFLOOP = 53908 # 0.823 × 65536 Q16_AVX_SELFLOOP = 57942 # 0.885 × 65536 Q16_RING_SELFLOOP = 0 # 0.0 # ── Config: the unit that flows through the pipeline ────────────────── @dataclass class Config: """A single configuration flowing through the pipeline.""" chiral: tuple # binary tuple (0=over, 1=under) per crossing labels: tuple # Sidon label set (integers) S: int # reflection point moduli: tuple # CRT moduli (L0, L1, ..., Lk) cost: int = 0 # compute cost (AngrySphinx) self_loop: int = 0 # contention proxy (COUCH, Q16_16 raw) sidon_score: int = 0 # Sidon score (Q16_16 raw: 65536 = perfect) collisions: int = 0 # collision count metadata: dict = field(default_factory=dict) # stage-specific data # ── Pipeline Context: shared state ───────────────────────────────────── @dataclass class PipelineContext: """Shared context across all stages.""" crossing_pairs: tuple = () # which strands cross: [(i,j), ...] groups: tuple = () # TreeBraid factorization seed: int = 0 stage_timings: dict = field(default_factory=dict) # ── Filter: the standard interface ──────────────────────────────────── class Filter(ABC): """Abstract base: every pipeline stage implements this.""" @abstractmethod def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: """Filter input configs → output configs.""" ... @property @abstractmethod def name(self) -> str: """Stage name for reporting.""" ... def run_stage(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: """Apply with timing.""" t0 = time.time() result = self.apply(configs, ctx) elapsed = time.time() - t0 ctx.stage_timings[self.name] = { "input": len(configs), "output": len(result), "time_s": round(elapsed, 4), } return result # ── Stage 1: BraidStorm — Generate ──────────────────────────────────── class BraidStorm(Filter): """Generates all 2^k chiral configurations.""" def __init__(self, k: int = 8): self.k = k @property def name(self) -> str: return f"BraidStorm(k={self.k})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: if configs: # Use first config as template template = configs[0] else: return [] all_chiral = list(product([0, 1], repeat=self.k)) return [ Config( chiral=c, labels=template.labels, S=template.S, moduli=template.moduli, ) for c in all_chiral ] # ── Stage 2: TreeBraid — Factorize ──────────────────────────────────── class TreeBraid(Filter): """Factorizes crossing space via braid relations. σ_i σ_j = σ_j σ_i when |i-j| >= 2 (independent). Marks configs with their factorization group. Does NOT filter — just annotates. Actual reduction happens in subsequent stages that can use the group structure. """ @property def name(self) -> str: return "TreeBraid" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: if not configs: return [] k = len(configs[0].chiral) pairs = ctx.crossing_pairs if not pairs: pairs = tuple((i, i+1) for i in range(k)) ctx.crossing_pairs = pairs groups = self._factorize(k, pairs) ctx.groups = groups # Annotate each config with its group signature for c in configs: # Group signature: which groups have at least one under-crossing sig = tuple( any(c.chiral[idx] for idx in group) for group in groups ) c.metadata["group_sig"] = sig c.metadata["groups"] = groups return configs # no filtering, just annotation def _factorize(self, k: int, pairs: tuple) -> tuple: groups = [] remaining = list(range(k)) while remaining: group = [remaining[0]] for idx in remaining[1:]: si, sj = pairs[idx] independent = True for gidx in group: gi, gj = pairs[gidx] if (abs(si - gi) < 2 or abs(si - gj) < 2 or abs(sj - gi) < 2 or abs(sj - gj) < 2): independent = False break if independent: group.append(idx) for g in group: remaining.remove(g) groups.append(tuple(group)) return tuple(groups) # ── Stage 3: AngrySphinx — Resource Budget ──────────────────────────── class AngrySphinx(Filter): """Filters by compute budget. Cost = 2^(under-crossings).""" def __init__(self, budget: int = 128): self.budget = budget @property def name(self) -> str: return f"AngrySphinx(budget={self.budget})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: result = [] for c in configs: under_count = sum(c.chiral) cost = 1 << under_count # 2^under_count — integer, no floats c.cost = cost if cost <= self.budget: result.append(c) return result # ── Stage 4: MultisurfacePacker — Spatial Fit ───────────────────────── class MultisurfacePacker(Filter): """Packs configs into available surfaces. Greedy by cost.""" def __init__(self, max_surfaces: int = 64): self.max_surfaces = max_surfaces @property def name(self) -> str: return f"MultisurfacePacker(max={self.max_surfaces})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: if len(configs) <= self.max_surfaces: return configs # Sort by cost (cheapest first = most efficient packing) sorted_configs = sorted(configs, key=lambda c: c.cost) return sorted_configs[:self.max_surfaces] # ── Stage 5: COUCH — Geometric Stability ────────────────────────────── class COUCHFilter(Filter): """COUCH gate: contention below threshold. Self-loop proxy: under-crossing count → contention level. 0 under = ring dispatch (self_loop=0, always passes) k/2 under = SUBLEQ (self_loop=53908) all under = AVX-512 (self_loop=57942) """ def __init__(self, threshold: int = Q16_THRESHOLD_COUCH): self.threshold = threshold @property def name(self) -> str: return f"COUCH(threshold={self.threshold})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: result = [] for c in configs: under_count = sum(c.chiral) k = len(c.chiral) # Self-loop: linear interpolation between ring (0) and AVX (57942) # All integer arithmetic: self_loop = 57942 * under_count // k c.self_loop = (Q16_AVX_SELFLOOP * under_count) // max(k, 1) if c.self_loop < self.threshold: result.append(c) return result # ── Stage 6: Sidon Filter — Algebraic Uniqueness ────────────────────── class SidonFilter(Filter): """Checks Sidon property via CRT reconstruction. POSITIONAL chirality: the chiral config permutes which label goes to which strand position. Each position has its own modulus. A permutation is NOT a ring automorphism — different label-to-modulus mappings CAN produce different Sidon results. The chiral tuple (ε₁, ..., εₖ) is interpreted as: εⱼ = 0: strand j stays in position j (no swap) εⱼ = 1: strand j swaps with strand j+1 (positional swap) Multiple swaps compose into a full permutation of labels across positions. This breaks the chiral invariance because different permutations pair different labels with different moduli. """ @property def name(self) -> str: return "SidonFilter" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: result = [] for c in configs: embedded = self._embed_chiral_positional(c) collisions = self._sidon_check(embedded, c.moduli) c.collisions = collisions total_pairs = len(c.labels) * (len(c.labels) + 1) // 2 # Sidon score: Q16_16 raw (65536 = perfect, 0 = all collide) c.sidon_score = Q16_ONE - (Q16_ONE * collisions) // max(total_pairs, 1) if collisions == 0: result.append(c) return result def _permute_labels(self, labels: tuple, chiral: tuple) -> list: """Apply positional chirality: chiral[j]=1 swaps positions j and j+1. This composes into a full permutation. Multiple swaps can interact (e.g., swap(0,1) then swap(1,2) moves label 0→2). """ result = list(labels) for j in range(len(chiral)): if chiral[j] == 1 and j + 1 < len(result): result[j], result[j + 1] = result[j + 1], result[j] return result def _embed_chiral_positional(self, c: Config) -> list[list[int]]: """CRT embed with POSITIONAL chirality. Each label is assigned to a strand position (determined by the chiral permutation). Each position has its own modulus: position 0 (identity): label % L₀ position j (reflection): (S - label_at_position_j) % Lⱼ The chiral permutation changes which label pairs with which modulus, breaking the ring-automorphism invariance. """ permuted = self._permute_labels(c.labels, c.chiral) embedded = [] for pos, a in enumerate(permuted): row = [a % c.moduli[0]] # identity axis (shared) for j in range(1, len(c.moduli)): row.append((c.S - a) % c.moduli[j]) embedded.append(row) return embedded def _sidon_check(self, embedded: list[list[int]], moduli: tuple) -> int: M = 1 for m in moduli: M *= m n = len(embedded) vals = [self._crt_reconstruct(row, moduli) for row in embedded] sums = [] for i in range(n): for j in range(i, n): sums.append((vals[i] + vals[j]) % M) counts = Counter(sums) return sum(c_count - 1 for c_count in counts.values()) def _crt_reconstruct(self, residues: list[int], moduli: tuple) -> int: M = 1 for m in moduli: M *= m x = 0 for r, m in zip(residues, moduli): Mi = M // m inv = self._modinv(Mi % m, m) if inv is None: return 0 x = (x + r * Mi * inv) % M return x def _egcd(self, a: int, b: int) -> tuple: if b == 0: return a, 1, 0 g, x, y = self._egcd(b, a % b) return g, y, x - (a // b) * y def _modinv(self, a: int, m: int) -> int | None: g, x, _ = self._egcd(a % m, m) return x % m if g == 1 else None # ── Swappable: Dual Quaternion Sidon Filter ─────────────────────────── class DualQuaternionSidonFilter(SidonFilter): """Sidon filter using dual quaternion products with POSITIONAL chirality. The positional permutation changes which label pairs with which modulus, so the DQ product (which involves r_i·t_j cross terms with different moduli for different positions) CAN discriminate chiral configurations. Unlike the negation-based chiral flip (which is a ring automorphism and preserves all algebraic structure), the positional permutation is NOT a ring automorphism and can change the Sidon property. """ @property def name(self) -> str: return "DualQuaternionSidonFilter" def _embed_chiral_positional(self, c: Config) -> list[list[int]]: """Embed as [r, t] pairs with POSITIONAL chirality. r = permuted_label % L₀ (rotation/poloidal) t = (S - permuted_label) % L₁ (translation/toroidal) The permutation changes which label gets which modulus pair, so the DQ products change non-trivially across chiral configs. """ permuted = self._permute_labels(c.labels, c.chiral) embedded = [] for a in permuted: r = a % c.moduli[0] if len(c.moduli) > 1: t = (c.S - a) % c.moduli[1] else: t = 0 embedded.append([r, t]) return embedded def _sidon_check(self, embedded: list[list[int]], moduli: tuple) -> int: """Check Sidon on dual quaternion PRODUCTS (not sums). Product: q_i ⊛ q_j = r_i·r_j + ε·(r_i·t_j + t_i·r_j) We check if all products are distinct. """ n = len(embedded) L0 = moduli[0] L1 = moduli[1] if len(moduli) > 1 else 1 products = [] for i in range(n): for j in range(i, n): ri, ti = embedded[i] rj, tj = embedded[j] # Product: r_i*r_j (rotation part) + r_i*t_j + t_i*r_j (translation part) # Encode as a pair — two products are equal iff both parts match rot_part = (ri * rj) % L0 trans_part = (ri * tj + ti * rj) % L1 products.append((rot_part, trans_part)) counts = Counter(products) return sum(c_count - 1 for c_count in counts.values()) # ── Pipeline: chains filters together ────────────────────────────────── class Pipeline: """Chains filters into a pipeline.""" def __init__(self, filters: list[Filter]): self.filters = filters def run(self, labels: list[int], S: int, moduli: list[int], crossing_pairs: tuple = ()) -> dict: t0 = time.time() ctx = PipelineContext( crossing_pairs=crossing_pairs, seed=hash((tuple(labels), S, tuple(moduli))) % (2**31), ) # Seed config configs = [Config( chiral=tuple(0 for _ in range(len(moduli) - 1)), labels=tuple(labels), S=S, moduli=tuple(moduli), )] # Run each stage for f in self.filters: configs = f.run_stage(configs, ctx) elapsed = time.time() - t0 result = { "experiment": "pipeline_core", "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "stages": [f.name for f in self.filters], "labels": list(labels), "S": S, "moduli": list(moduli), "stage_timings": ctx.stage_timings, "total_input": 1, "total_output": len(configs), "reduction": "N/A", "elapsed_s": round(elapsed, 4), "final_configs": [ { "chiral": list(c.chiral), "cost": c.cost, "self_loop": c.self_loop, "sidon_score": c.sidon_score, "collisions": c.collisions, } for c in configs ], } result["reduction"] = f"{result['total_input']} → {result['total_output']}" content = json.dumps(result, indent=2, sort_keys=True, default=str) result["sha256"] = hashlib.sha256(content.encode()).hexdigest() # Print summary print(f"\n{'='*60}") print(f" PIPELINE: {' → '.join(f.name for f in self.filters)}") print(f"{'='*60}") for s, t in ctx.stage_timings.items(): print(f" {s:30s} {t['input']:6d} → {t['output']:6d} ({t['time_s']:.4f}s)") print(f"{'='*60}") print(f" Total: {result['total_input']} → {result['total_output']} ({elapsed:.2f}s)") print(f"{'='*60}") return result # ── Main: default pipeline ───────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Module-swappable pipeline") parser.add_argument("--strands", type=int, default=8) parser.add_argument("--budget", type=int, default=128) parser.add_argument("--surfaces", type=int, default=64) parser.add_argument("--filter", choices=["crt", "dq"], default="crt", help="Sidon filter: crt (sum-based) or dq (dual quaternion)") parser.add_argument("--output", default="pipeline_result.json") args = parser.parse_args() labels = [1, 2, 4, 8, 16, 32, 64, 128] S = 128 moduli = [7, 3, 5, 11, 13, 17, 19, 23, 29] # Select Sidon filter sidon_filter = SidonFilter() if args.filter == "crt" else DualQuaternionSidonFilter() # Build swappable pipeline pipe = Pipeline([ BraidStorm(k=args.strands), TreeBraid(), AngrySphinx(budget=args.budget), MultisurfacePacker(max_surfaces=args.surfaces), COUCHFilter(), sidon_filter, ]) result = pipe.run( labels=labels, S=S, moduli=moduli, crossing_pairs=tuple((i, i+1) for i in range(args.strands)), ) out_path = ARTIFACTS_DIR / args.output out_path.write_text(json.dumps(result, indent=2, default=str)) print(f"\nResults → {out_path}")