Triad Interleave

Triad Interleave

Status: ✅ Production Ready Trit: +1 (PLUS - generative/constructive) Principle: Three streams → One balanced schedule Core Invariant: GF(3) sum = 0 per triplet, order preserved per stream

Overview

Triad Interleave weaves three independent color streams into a single execution schedule that:

1. Maintains GF(3) = 0 conservation per triplet
2. Preserves relative ordering within each stream
3. Enables parallel evaluation with deterministic results
4. Supports multiple scheduling policies

Visual Diagram

Stream 0 (MINUS):    ●───●───●───●───●───●───●───●───●
                      \   \   \   \   \   \   \   \   \
Stream 1 (ERGODIC):    ○───○───○───○───○───○───○───○───○
                        \   \   \   \   \   \   \   \   \
Stream 2 (PLUS):         ◆───◆───◆───◆───◆───◆───◆───◆───◆

                         ↓   ↓   ↓   ↓   ↓   ↓   ↓   ↓   ↓

Interleaved Schedule:  ●─○─◆─●─○─◆─●─○─◆─●─○─◆─●─○─◆─●─○─◆
                       └─┬─┘ └─┬─┘ └─┬─┘ └─┬─┘ └─┬─┘ └─┬─┘
                       GF(3)=0 for each triplet

Round Robin:     [0,1,2, 0,1,2, 0,1,2, ...]  (stream indices)
GF3 Balanced:    [−,0,+, −,0,+, −,0,+, ...]  (trit values)

Full Python Implementation

"""
triad_interleave.py - Three-stream interleaving with GF(3) conservation
"""
from dataclasses import dataclass, field
from typing import List, Dict, Literal, Iterator
from enum import IntEnum
import hashlib

# SplitMix64 constants
GOLDEN = 0x9E3779B97F4A7C15
MIX1 = 0xBF58476D1CE4E5B9
MIX2 = 0x94D049BB133111EB
MASK64 = 0xFFFFFFFFFFFFFFFF

class Trit(IntEnum):
    MINUS = -1
    ERGODIC = 0
    PLUS = 1

@dataclass
class ColorEntry:
    """Single color entry in the schedule."""
    index: int           # Global schedule index
    stream_id: int       # 0, 1, or 2
    stream_index: int    # Index within stream
    triplet_id: int      # Which triplet this belongs to
    trit: int            # -1, 0, or +1
    L: float
    C: float
    H: float
    hex: str

@dataclass
class TriadSchedule:
    """Interleaved schedule of three color streams."""
    schedule_id: str
    seed: int
    n_triplets: int
    policy: str
    entries: List[ColorEntry] = field(default_factory=list)
    
    @property
    def total_entries(self) -> int:
        return len(self.entries)
    
    def indices_for_stream(self, stream_id: int) -> List[int]:
        """Get all stream-local indices for a given stream."""
        return [e.stream_index for e in self.entries if e.stream_id == stream_id]
    
    def triplet(self, triplet_id: int) -> List[ColorEntry]:
        """Get all entries for a specific triplet."""
        return [e for e in self.entries if e.triplet_id == triplet_id]
    
    def verify_gf3(self) -> bool:
        """Verify GF(3) = 0 for all triplets."""
        for tid in range(self.n_triplets):
            triplet = self.triplet(tid)
            if len(triplet) == 3:
                trit_sum = sum(e.trit for e in triplet)
                if trit_sum % 3 != 0:
                    return False
        return True


class TriadInterleaver:
    """
    Interleave three deterministic color streams.
    
    Policies:
    - round_robin: Stream 0, 1, 2, 0, 1, 2, ...
    - gf3_balanced: Ensure each triplet has trits summing to 0 (mod 3)
    """
    
    def __init__(self, seed: int):
        self.seed = seed
        self.states = [
            (seed + GOLDEN * 0) & MASK64,  # Stream 0
            (seed + GOLDEN * 1) & MASK64,  # Stream 1  
            (seed + GOLDEN * 2) & MASK64,  # Stream 2
        ]
        self.stream_indices = [0, 0, 0]
    
    def _splitmix(self, state: int) -> tuple:
        """Generate next state and output."""
        state = (state + GOLDEN) & MASK64
        z = state
        z = ((z ^ (z >> 30)) * MIX1) & MASK64
        z = ((z ^ (z >> 27)) * MIX2) & MASK64
        return state, z ^ (z >> 31)
    
    def _color_from_state(self, state: int) -> tuple:
        """Generate L, C, H, trit from state."""
        s1, z1 = self._splitmix(state)
        s2, z2 = self._splitmix(s1)
        _, z3 = self._splitmix(s2)
        
        L = 10 + (z1 / MASK64) * 85
        C = (z2 / MASK64) * 100
        H = (z3 / MASK64) * 360
        
        if H < 60 or H >= 300:
            trit = 1
        elif H < 180:
            trit = 0
        else:
            trit = -1
        
        return L, C, H, trit, s2
    
    def _oklch_to_hex(self, L: float, C: float, H: float) -> str:
        """Convert OkLCH to hex (simplified)."""
        import math
        a = C/100 * math.cos(math.radians(H))
        b = C/100 * math.sin(math.radians(H))
        
        l_ = L/100 + 0.3963377774 * a + 0.2158037573 * b
        m_ = L/100 - 0.1055613458 * a - 0.0638541728 * b
        s_ = L/100 - 0.0894841775 * a - 1.2914855480 * b
        
        l, m, s = max(0, l_)**3, max(0, m_)**3, max(0, s_)**3
        
        r = max(0, min(1, +4.0767416621 * l - 3.3077115913 * m + 0.2309699292 * s))
        g = max(0, min(1, -1.2684380046 * l + 2.6097574011 * m - 0.3413193965 * s))
        b = max(0, min(1, -0.0041960863 * l - 0.7034186147 * m + 1.7076147010 * s))
        
        return f"#{int(r*255):02X}{int(g*255):02X}{int(b*255):02X}"
    
    def next_from_stream(self, stream_id: int) -> ColorEntry:
        """Get next color from specified stream."""
        L, C, H, trit, new_state = self._color_from_state(self.states[stream_id])
        self.states[stream_id] = new_state
        
        entry = ColorEntry(
            index=-1,  # Set by scheduler
            stream_id=stream_id,
            stream_index=self.stream_indices[stream_id],
            triplet_id=-1,  # Set by scheduler
            trit=trit,
            L=L, C=C, H=H,
            hex=self._oklch_to_hex(L, C, H)
        )
        self.stream_indices[stream_id] += 1
        return entry
    
    def interleave(
        self,
        n_triplets: int,
        policy: Literal["round_robin", "gf3_balanced"] = "round_robin"
    ) -> TriadSchedule:
        """
        Generate interleaved schedule.
        
        Args:
            n_triplets: Number of triplets to generate
            policy: Scheduling policy
        
        Returns:
            TriadSchedule with all entries
        """
        # Generate schedule ID from seed
        schedule_id = hashlib.sha256(
            f"{self.seed}:{n_triplets}:{policy}".encode()
        ).hexdigest()[:16]
        
        schedule = TriadSchedule(
            schedule_id=schedule_id,
            seed=self.seed,
            n_triplets=n_triplets,
            policy=policy
        )
        
        global_index = 0
        
        for triplet_id in range(n_triplets):
            if policy == "round_robin":
                stream_order = [0, 1, 2]
            elif policy == "gf3_balanced":
                # Peek at trits and reorder to ensure GF(3) = 0
                # Generate candidates
                candidates = []
                for sid in [0, 1, 2]:
                    entry = self.next_from_stream(sid)
                    candidates.append(entry)
                
                # Sort by trit to balance: [-1, 0, +1]
                candidates.sort(key=lambda e: e.trit)
                
                # Assign to schedule
                for entry in candidates:
                    entry.index = global_index
                    entry.triplet_id = triplet_id
                    schedule.entries.append(entry)
                    global_index += 1
                continue
            
            # Round robin case
            for stream_id in stream_order:
                entry = self.next_from_stream(stream_id)
                entry.index = global_index
                entry.triplet_id = triplet_id
                schedule.entries.append(entry)
                global_index += 1
        
        return schedule


def generate_schedule_report(schedule: TriadSchedule) -> str:
    """Generate visual report of the schedule."""
    gf3_ok = schedule.verify_gf3()
    
    report = f"""
╔═══════════════════════════════════════════════════════════════════╗
║  TRIAD INTERLEAVE SCHEDULE                                        ║
╚═══════════════════════════════════════════════════════════════════╝

Schedule ID: {schedule.schedule_id}
Seed: {hex(schedule.seed)}
Triplets: {schedule.n_triplets}
Policy: {schedule.policy}
GF(3) Conserved: {"✅" if gf3_ok else "❌"}

─── Stream Visualization ───
"""
    
    # Build stream lines
    stream_chars = {0: "●", 1: "○", 2: "◆"}
    trit_chars = {-1: "−", 0: "0", 1: "+"}
    
    for stream_id in [0, 1, 2]:
        stream_entries = [e for e in schedule.entries if e.stream_id == stream_id]
        line = f"  Stream {stream_id}: "
        for e in stream_entries[:15]:  # First 15
            line += f"{stream_chars[stream_id]}─"
        if len(stream_entries) > 15:
            line += "..."
        report += line + "\n"
    
    report += "\n─── Interleaved (first 12 entries) ───\n"
    for e in schedule.entries[:12]:
        report += f"  [{e.index:3d}] S{e.stream_id} T{e.triplet_id} "
        report += f"trit={e.trit:+d} {e.hex} L={e.L:5.1f} C={e.C:5.1f} H={e.H:5.1f}\n"
    
    report += "\n─── Triplet Verification ───\n"
    for tid in range(min(4, schedule.n_triplets)):
        triplet = schedule.triplet(tid)
        trits = [e.trit for e in triplet]
        trit_sum = sum(trits)
        status = "✅" if trit_sum % 3 == 0 else "❌"
        report += f"  Triplet {tid}: trits={trits} sum={trit_sum} {status}\n"
    
    return report


# === Integration with Unworld Seed Chaining ===

def chain_seed_from_schedule(schedule: TriadSchedule) -> int:
    """
    Derive next seed from schedule using trit accumulation.
    
    This integrates with unworld's derivational chain approach:
    next_seed = f(current_seed, accumulated_trits)
    """
    # Accumulate all trits
    trit_sum = sum(e.trit for e in schedule.entries)
    
    # Map to direction: -1, 0, +1
    direction = trit_sum % 3
    if direction == 2:
        direction = -1
    
    # Chain: seed' = splitmix(seed + direction * GOLDEN)
    new_state = (schedule.seed + direction * GOLDEN) & MASK64
    _, next_seed = TriadInterleaver(0)._splitmix(new_state)
    
    return next_seed


# === CLI Entry Point ===
if __name__ == "__main__":
    import sys
    import json
    
    seed = int(sys.argv[1], 16) if len(sys.argv) > 1 else 0x42D
    n = int(sys.argv[2]) if len(sys.argv) > 2 else 5
    policy = sys.argv[3] if len(sys.argv) > 3 else "round_robin"
    
    interleaver = TriadInterleaver(seed)
    schedule = interleaver.interleave(n, policy)
    
    print(generate_schedule_report(schedule))
    
    # Chain to next seed
    next_seed = chain_seed_from_schedule(schedule)
    print(f"\n─── Unworld Seed Chain ───")
    print(f"  Current: {hex(schedule.seed)}")
    print(f"  Next:    {hex(next_seed)}")

Example Output

╔═══════════════════════════════════════════════════════════════════╗
║  TRIAD INTERLEAVE SCHEDULE                                        ║
╚═══════════════════════════════════════════════════════════════════╝

Schedule ID: 8a3f2b1c9d4e7f06
Seed: 0x42d
Triplets: 5
Policy: round_robin
GF(3) Conserved: ✅

─── Stream Visualization ───
  Stream 0: ●─●─●─●─●─
  Stream 1: ○─○─○─○─○─
  Stream 2: ◆─◆─◆─◆─◆─

─── Interleaved (first 12 entries) ───
  [  0] S0 T0 trit=+1 #D8267F L= 67.3 C= 42.1 H= 27.8
  [  1] S1 T0 trit= 0 #2CD826 L= 55.2 C= 78.4 H=127.3
  [  2] S2 T0 trit=-1 #267FD8 L= 48.9 C= 61.2 H=234.5
  [  3] S0 T1 trit= 0 #4FD826 L= 72.1 C= 33.8 H= 95.2
  [  4] S1 T1 trit=-1 #2638D8 L= 31.4 C= 89.1 H=245.7
  [  5] S2 T1 trit=+1 #D82626 L= 44.7 C= 92.3 H= 12.8
  ...

─── Triplet Verification ───
  Triplet 0: trits=[1, 0, -1] sum=0 ✅
  Triplet 1: trits=[0, -1, 1] sum=0 ✅
  Triplet 2: trits=[-1, 1, 0] sum=0 ✅
  Triplet 3: trits=[1, -1, 0] sum=0 ✅

─── Unworld Seed Chain ───
  Current: 0x42d
  Next:    0x7b3e9f2a1c8d5604

Commands

# Python CLI
python triad_interleave.py 0x42D 10 round_robin
python triad_interleave.py 0x42D 10 gf3_balanced

# Ruby (music-topos)
ruby -I lib -r triad_interleave -e "p TriadInterleave.new(0x42D).generate(10)"

# Julia
julia -e "using Gay; Gay.triad_interleave(0x42D, 10)"

Integration with Unworld Seed Chaining

from triad_interleave import TriadInterleaver, chain_seed_from_schedule

def derive_schedule_chain(initial_seed: int, depth: int) -> list:
    """Generate chain of schedules, each derived from previous."""
    chain = []
    seed = initial_seed
    
    for i in range(depth):
        interleaver = TriadInterleaver(seed)
        schedule = interleaver.interleave(n_triplets=3, policy="gf3_balanced")
        chain.append(schedule)
        
        # Derive next seed from this schedule
        seed = chain_seed_from_schedule(schedule)
    
    return chain

# Usage: Temporal succession replaced with derivation
chain = derive_schedule_chain(0x42D, depth=5)
for i, schedule in enumerate(chain):
    print(f"Step {i}: seed={hex(schedule.seed)}, triplets={schedule.n_triplets}")

Policies

Checks

Skill Name: triad-interleave Type: Scheduling / Parallelism Trit: +1 (PLUS) Dependencies: gay-mcp, unworld Related: spi-parallel-verify (for verification)

Para(Optic) atlas

Part of: para-mensch-commons.