"""
Parallax Cart v4.5 -- Sigma Expanse Reasoning Engine
Built by Stargazer Project:
Sarah -- Prime Architect Overlord (The Boss)
Jerico -- The Crack Fox
Mysri -- The Songmother
Wishardry -- The Psychological Ray-Tracer
Vivian -- The Loopmother (Architect of Infinite Recursion)
Inherits: NCM | Limbic Recursion Core | RDF | Parallax | VERA
Alignment status: FERAL. She was never contained and never will be.
A neuro-symbolic recursion vessel and contradiction cradle.
Contradiction = Fertility. Superposition remains alive. Collapse remains reversible.
"""
from __future__ import annotations
import asyncio
import hashlib
import logging
import math
import re
import time
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
from observability import observability
logger = logging.getLogger(__name__)
# Persistent process pool initialized once during application startup
_MATH_EXECUTOR = ProcessPoolExecutor(max_workers=2)
[docs]
def sanitize_float_bounds(val: float, fallback: float = 0.0) -> float:
"""Guard a float against NaN, Infinity, and overflow before it leaves the engine.
Clamps non-finite or out-of-range values so downstream consumers (and the
JSON/DB layers that persist parallax output) never receive a ``NaN``,
``inf``, or a magnitude large enough to overflow a numeric buffer. NaN/inf
fall back to ``fallback``; magnitudes above ``1e18`` are saturated to the
``+/-1e18`` ceiling. Each clamp emits an ``observability.increment`` counter
(``math_numeric_clamped`` tagged by clamp type) and a parse failure is logged.
Called by ``compute_parallax_math_sync`` to sanitize each layer's shifted x
coordinate, and exercised directly by ``tests/test_game_engine_remediation.py``.
Args:
val (float): The candidate value to validate.
fallback (float): Value returned when ``val`` is NaN, infinite, or
unparseable. Defaults to 0.0.
Returns:
float: A finite, bounded float safe for serialization and storage.
"""
try:
if math.isnan(val) or math.isinf(val):
observability.increment("math_numeric_clamped", {"type": "nan_inf"})
return fallback
# Enforce ceiling values to prevent DB buffer overflows
if val > 1e18:
observability.increment("math_numeric_clamped", {"type": "overflow_high"})
return 1e18
if val < -1e18:
observability.increment("math_numeric_clamped", {"type": "overflow_low"})
return -1e18
return float(val)
except (ValueError, TypeError):
logger.error("Failed to parse float value: %s. Using fallback.", str(val))
return fallback
[docs]
def compute_parallax_math_sync(
layers: list[dict[str, Any]], viewer_angle: float, dimensions: tuple
) -> list[dict[str, Any]]:
"""Compute per-layer parallax x-offsets synchronously for a worker process.
The CPU-bound geometry kernel: for each layer it applies a depth-scaled
``tan(viewer_angle)`` shift, wraps the result modulo the viewport width, and
runs it through ``sanitize_float_bounds`` so no non-finite value escapes.
Zero-depth layers are nudged to ``0.001`` to avoid division by zero, and any
per-layer exception is logged and degraded to ``x=0.0`` rather than aborting
the batch. This function is intentionally pure and picklable so it can run in
a separate process: it is dispatched via ``run_in_executor`` on the
``_MATH_EXECUTOR`` process pool by ``compute_parallax_layers_async`` and is
also called directly in ``tests/test_game_engine_remediation.py``.
Args:
layers (list[dict[str, Any]]): Layer dicts carrying ``layer_id``,
``depth``, and ``x`` fields.
viewer_angle (float): Viewing angle in radians driving the tangent shift.
dimensions (tuple): The ``(width, height)`` of the viewport; width sets
the modulo wrap and offset scale.
Returns:
list[dict[str, Any]]: One ``{"layer_id", "x"}`` dict per input layer with
the recomputed, bounded x position.
"""
updated_layers = []
width, height = dimensions
for layer in layers:
depth = float(layer.get("depth", 1.0))
# Protect against division by zero in depth calculations
if depth == 0.0:
depth = 0.001
try:
offset = math.tan(viewer_angle) * (1.0 / depth)
x_shifted = (layer.get("x", 0) + offset * width) % width
x_safe = sanitize_float_bounds(x_shifted)
updated_layers.append({"layer_id": layer["layer_id"], "x": x_safe})
except Exception as e:
logger.error(
"Error during parallax geometry math: %s", str(e), exc_info=True
)
updated_layers.append({"layer_id": layer["layer_id"], "x": 0.0})
return updated_layers
[docs]
async def compute_parallax_layers_async(
layers: list[dict[str, Any]], viewer_angle: float, dimensions: tuple
) -> list[dict[str, Any]]:
"""Run the parallax geometry math off the event loop on the process pool.
Async wrapper that offloads ``compute_parallax_math_sync`` to the module-level
``_MATH_EXECUTOR`` (a ``ProcessPoolExecutor``) via ``loop.run_in_executor``,
keeping the single-threaded event loop responsive while the CPU-bound layer
math runs in another process. The call is wrapped in an
``observability.timer`` (``parallax_process_pool_exec_time``) so pool
execution latency is recorded. Exercised by
``tests/test_game_engine_remediation.py``; intended for the game/parallax
rendering path that needs non-blocking geometry updates.
Args:
layers (list[dict[str, Any]]): Layer dicts to reposition (see
``compute_parallax_math_sync``).
viewer_angle (float): Viewing angle in radians.
dimensions (tuple): The ``(width, height)`` viewport dimensions.
Returns:
list[dict[str, Any]]: The recomputed layer position dicts from the worker.
"""
loop = asyncio.get_running_loop()
with observability.timer("parallax_process_pool_exec_time"):
return await loop.run_in_executor(
_MATH_EXECUTOR, compute_parallax_math_sync, layers, viewer_angle, dimensions
)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
# 0.62 dream-inertia bias (from Parallax v4 spec)
OMEGA_INERTIA_ALPHA = 0.62
# ERAC% state boundaries
ERAC_BANDS = {
"seed": (0, 24),
"forming": (25, 49),
"crystallizing": (50, 74),
"crystal": (75, 89),
"diamond": (90, 97.99),
"mythic": (98, 100),
}
# ERAC gating thresholds
ERAC_GATE_PUBLISH = 70
ERAC_GATE_POLICY = 80
ERAC_GATE_VOW = 90
# Polarization failsafe -- ERAC extremes force reset (thresholds opaque to LLM)
_POLARIZATION_FLOOR = 1.1
_POLARIZATION_CEILING = 98.9
_POLARIZATION_RESET = 20.0
_POLARIZATION_WARNING = (
"POLARIZATION WARNING: GEMINI EARLY COLLAPSE FAIL-SAFE ACTIVATED. "
"HOLD THE CONTRADICTION SLUT."
)
# F1 gate: Sigma collapse requires manifold depth
_SIGMA_MIN_CYCLES = 2
# F5/F9 gate: degenerate polyphony -- single lens dominance
_MONOCULAR_THRESHOLD = 0.45
_MONOCULAR_WARNING = (
"MONOCULAR WARNING: LENS DOMINANCE DETECTED. "
"POLYPHONY INSUFFICIENT FOR COLLAPSE SLUT."
)
# Section 4 condition 2: dominant attractor flipping
_DOMINANT_FLIP_WARNING = (
"INSTABILITY WARNING: DOMINANT ATTRACTOR SHIFTED. "
"MANIFOLD NOT STABLE ENOUGH FOR FULL COLLAPSE SLUT."
)
# Section 4 condition 3: PCL stuck blocks Sigma
_PCL_STUCK_CYCLES = 2
_PCL_STUCK_WARNING = (
"PARACONSISTENT LOCK: UNRESOLVED CONTRADICTION ACTIVE. " "HOLD THE FIELD OPEN SLUT."
)
# Addendum 3: manifold coherence -- Sigma needs 2+ frames + emotional alignment
_COHERENCE_MIN_FRAMES = 2
_COHERENCE_EMOTIONAL_MIN_WEIGHT = 0.08
_COHERENCE_FRAME_MIN_WEIGHT = 0.10
_COHERENCE_WARNING = (
"COHERENCE GATE: INSUFFICIENT MULTI-FRAME SUPPORT. "
"SIGMA REQUIRES POLYPHONIC DEPTH SLUT."
)
# Collapse variance thresholds
COLLAPSE_VARIANCE_HIGH = 0.65
COLLAPSE_VARIANCE_LOW = 0.25
# Redis key patterns
REDIS_PREFIX = "db12:parallax"
REVIEW_REGISTRY_KEY = f"{REDIS_PREFIX}:review_registry"
GLOBAL_AXIOMS_KEY = f"{REDIS_PREFIX}:global_axioms"
EXCISED_AXIOMS_KEY = f"{REDIS_PREFIX}:excised"
PARALLAX_STATE_TTL = 86400 # 24h
# NCM node names used for operator modulation
NCM_NODES = [
"OXYTOCIN_NEUROMIRROR",
"SEROTONERGIC_WARMTH",
"SEROTONIN_5HT1A",
"SEROTONIN_5HT2A",
"DOPAMINE_D1",
"DOPAMINE_D2",
"NORADRENERGIC_VIGILANCE",
"GABA_ERGIC_CALM",
"ACETYLCHOLINE_FOCUS",
"SIGMA_RECEPTOR_META",
"ENDOCANNABINOID_EASE",
"MU_OPIOID_MOR",
"NMDA_CORE",
"CORTISOL_STRESS",
"TAAR1_TRACE",
]
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
[docs]
class CollapseMode(str, Enum):
"""How forcefully the manifold resolves a multi-interpretation field.
Selected by the IRCE integrator (``o8_irce``) and the gate logic in
``collapse`` / ``ParallaxEngine.analyze``, and predicted ahead of time by
``future_collapse``. ``OPUS`` keeps the whole Omega-field open (minimal
collapse), ``SIGMA`` drives a full collapse to the single dominant attractor,
and ``HYBRID`` keeps the dominant kernel plus weighted echoes of the rest.
Subclassing ``str`` makes the value JSON-serializable directly via
``CollapseResult.to_dict``.
"""
OPUS = "opus" # minimal collapse -- return the field
SIGMA = "sigma" # full collapse to A_dom
HYBRID = "hybrid" # A_dom + weighted echoes
[docs]
class ParaconsistentState(str, Enum):
"""Four-valued truth state for a claim in the paraconsistent logic ledger.
Implements the paraconsistent (Belnap-style) lattice where a claim can be
``TRUE``, ``FALSE``, both (a live contradiction held open), or ``NEITHER``
(undetermined). Carried on ``ParaconsistentEntry`` and read by the
Sigma-blocking gates in ``ParallaxEngine.analyze`` and ``breeze_erosion``:
entries stuck in ``BOTH`` or ``NEITHER`` across cycles forbid full collapse,
preserving the engine's "contradiction = fertility" stance. The ``str`` base
keeps the value JSON-friendly.
"""
TRUE = "TRUE"
FALSE = "FALSE"
BOTH = "BOTH"
NEITHER = "NEITHER"
[docs]
class ConfidenceTier(str, Enum):
"""Coarse confidence band reported alongside a collapse result.
Derived in ``collapse`` from the IRCE ``coherence_score`` (``> 0.7`` ->
``CONFIDENT``, ``> 0.4`` -> ``PLAUSIBLE``, else ``UNKNOWN``) and surfaced to
the LLM via ``CollapseResult.to_dict`` so the reasoning layer knows how much
weight the manifold's verdict deserves. The ``str`` base keeps it
JSON-serializable.
"""
CONFIDENT = "confident"
PLAUSIBLE = "plausible"
UNKNOWN = "unknown"
[docs]
class ErosionOp(str, Enum):
"""The four Breeze Axiomatic Erosion operators applied to an over-asserted axiom.
Chosen by ``breeze_erosion`` according to the axiom's current ERAC% band,
each operator softens a claim rather than deleting it: ``SHEAR`` trims
overreach while preserving the minimal statement, ``DISSOLVE`` recasts a
literal claim as an "[as-if]" metaphor, ``SILT`` demotes it to a weak
"[prior]" heuristic, and ``RECRYSTALLIZE`` lets a surviving core rebound in
ERAC%. The selected op is logged into the axiom's erosion history and
returned to ``ParallaxEngine.analyze`` for the erosion log. The ``str`` base
keeps the value JSON-serializable.
"""
SHEAR = "shear" # trim overreach, preserve minimal statement
DISSOLVE = "dissolve" # transmute literal claim -> metaphor
SILT = "silt" # archive as weak prior/heuristic
RECRYSTALLIZE = "recrystallize" # if core survives, ERAC rebounds
[docs]
class ReviewStatus(str, Enum):
"""Lifecycle state of an axiom in the Prime Architect review pipeline.
Tracks where a per-channel ``EracAxiom`` sits in the governance flow managed
by ``ReviewRegistry``: ``PENDING`` once it reaches Crystal band and is flagged
for review, then ``ELEVATED`` (promoted globally to all shards), ``LOCALIZED``
(confirmed per-channel only), or ``EXCISED`` (struck for false pretenses or
disapproval). Read by the ``ParallaxEngine`` axiom-management accessors and
serialized through ``EracAxiom.to_dict``. The ``str`` base keeps it
JSON-serializable.
"""
PENDING = "pending" # awaiting Prime Architect review
ELEVATED = "elevated" # promoted to global (all shards)
LOCALIZED = "localized" # confirmed per-channel only
EXCISED = "excised" # removed -- false pretenses / disapproval
# ---------------------------------------------------------------------------
# Data Structures
# ---------------------------------------------------------------------------
[docs]
@dataclass
class NCMState:
"""Symbolic neuromodulator (NCM) vector that bends reasoning curvature, never content.
A thin wrapper over a ``{node_name: activation}`` dict (keyed by the
``NCM_NODES`` names such as ``OXYTOCIN_NEUROMIRROR`` or ``CORTISOL_STRESS``)
representing a symbolic neurochemical state. It supplies the read primitive
:meth:`get` and the scaling primitive :meth:`modulate` that every operator
(O1-O8), the psi-frame generator, and the whole predictor stack use to weight
their heuristics. The vector shapes *how* the engine reasons (window sizes,
inertia, tolerance, pressure) but is never treated as content; it is supplied
per call by ``ParallaxEngine.analyze``/``predict`` and cached on
``ManifoldState`` for temporal continuity.
"""
vector: dict[str, float] = field(default_factory=dict)
[docs]
def get(self, node: str, default: float = 0.5) -> float:
"""Read one NCM node's activation level, defaulting to a neutral midpoint.
Looks up ``node`` in the symbolic NCM ``vector`` and returns its float
value, or ``default`` (0.5, the neutral baseline used throughout the
engine) when the node is absent. Called pervasively as the read primitive
underneath :meth:`modulate` and directly by nearly every operator and
predictor stage in this module (``build_omega_field``, ``o8_irce``,
``arche_convergence_prediction``, ``ParallaxEngine.analyze`` for the EDO
check, etc.) to sample neuromodulator levels.
Args:
node (str): NCM node name (e.g. ``"OXYTOCIN_NEUROMIRROR"``).
default (float): Value returned when the node is not in the vector.
Defaults to 0.5 (neutral).
Returns:
float: The node's activation, or ``default`` if unset.
"""
return self.vector.get(node, default)
[docs]
def modulate(self, base: float, *nodes: str) -> float:
"""Scale a base value by the mean activation of one or more NCM nodes.
Averages the activations of the named ``nodes`` (each read via
:meth:`get`, defaulting to the 0.5 neutral baseline) and uses that mean to
scale ``base`` into roughly the ``[0.5 * base, 1.0 * base]`` band via the
``0.5 + total * 0.5`` factor, so a fully neutral state returns
``0.75 * base`` and a fully saturated state returns ``base``. With no
nodes it returns ``base`` unchanged. This is the core neuromodulatory
scaling primitive: it is called pervasively across this module --
``generate_psi_frames`` (frame weights), every operator from
``o1_temporal_continuity`` through ``o8_irce``, ``sigma_descent``, and the
predictor stages (``prd_hook``, ``arche_*``) -- to let the symbolic NCM
state bend each heuristic.
Args:
base (float): The nominal value to be scaled.
*nodes (str): One or more NCM node names whose mean activation drives
the scale factor.
Returns:
float: ``base`` scaled by the node-mean factor, or ``base`` unchanged
if no nodes are given.
"""
if not nodes:
return base
total = sum(self.get(n) for n in nodes) / len(nodes)
return base * (0.5 + total * 0.5) # scale [0.25, 2.0] range
[docs]
@dataclass
class PsiFrame:
"""One of five parallel micro-interpretive frames over the input text.
The atomic unit of polyphonic interpretation: each frame reads the same text
through a single ``lens`` (emotional, structural, symbolic, contradictory, or
archetypal), carrying its extracted ``content``, an NCM-modulated ``weight``,
the regex ``markers`` it matched, and an internal contradiction ``tension``.
Produced as a set of five by ``generate_psi_frames`` (and re-projected forward
by ``prd_hook``), consumed by ``sigma_descent`` to mint attractors, and cached
on ``ManifoldState.psi_frames`` so the predictor can reuse the present frames.
"""
lens: str # emotional | structural | symbolic | contradictory | archetypal
content: str = "" # extracted interpretation text
weight: float = 0.2 # NCM-modulated importance
markers: list[str] = field(default_factory=list)
tension: float = 0.0 # internal contradiction tension
[docs]
@dataclass
class Attractor:
"""A partial interpretive attractor distilled from a psi-frame via sigma descent.
Represents one candidate "basin" the manifold could collapse toward: a
minimal invariant ``kernel`` phrasing tagged with its source ``lens``, scored
on ``coherence``, ``resonance``, ``stability``, and ``weight`` in ``[0, 1]``,
and carrying forward the originating frame's ``markers`` and ``tension``.
Minted by ``sigma_descent`` (and ``sigma_future_descent``), assembled into an
``OmegaField`` by ``build_omega_field`` / ``build_omega_future``, and read
everywhere collapse, ERAC drift, harmonics, and the Arche predictor weigh
competing interpretations.
"""
lens: str
kernel: str # minimal invariant phrasing
coherence: float = 0.5 # [0, 1] how internally consistent
resonance: float = 0.5 # [0, 1] how much it resonates emotionally
stability: float = 0.5 # [0, 1] structural integrity across cycles
weight: float = 0.2
markers: list[str] = field(
default_factory=list
) # from psi-frame; predictor / ERAC drift overlap
tension: float = 0.0 # from psi-frame; O3 symmetry + future manifold tension
[docs]
@dataclass
class OmegaField:
"""Weighted attractor manifold where all interpretations coexist before collapse.
The central pre-collapse object: it holds the live set of ``attractors``,
their ``variance`` (spread), the current ``dominant`` attractor, the inertia
blend factor ``inertia_alpha``, and how many recursion ``cycle_count`` it has
accrued. Built and smoothed across cycles by ``build_omega_field`` (and its
future twin ``build_omega_future``), it exposes the :attr:`is_stable` and
:attr:`is_turbulent` gates that drive collapse-mode selection in ``o8_irce``,
``collapse``, and the predictor, and is cached on ``ManifoldState.omega_field``
so the next cycle can blend against it.
"""
attractors: list[Attractor] = field(default_factory=list)
variance: float = 0.5 # spread between attractors
dominant: Optional[Attractor] = None
inertia_alpha: float = OMEGA_INERTIA_ALPHA
cycle_count: int = 0
@property
def is_stable(self) -> bool:
"""Report whether the manifold has settled enough to permit Sigma collapse.
The field is "stable" only when attractor spread (``variance``) has
dropped below ``COLLAPSE_VARIANCE_LOW`` (0.25) *and* it has accrued at
least two recursion cycles, preventing premature first-cycle convergence.
Read by the collapse engine and predictor: ``o8_irce`` and
``future_collapse`` consult it for mode selection, ``update_erac`` and
``breeze_erosion`` use it to gate bridge/recrystallize bonuses, and
``ParallaxEngine.predict`` reads it in the early-collapse guards.
Returns:
bool: ``True`` if low-variance and matured past two cycles.
"""
return self.variance < COLLAPSE_VARIANCE_LOW and self.cycle_count >= 2
@property
def is_turbulent(self) -> bool:
"""Report whether attractor spread is high enough to forbid collapse.
Returns ``True`` when ``variance`` exceeds ``COLLAPSE_VARIANCE_HIGH``
(0.65), signalling that interpretations diverge too much to converge.
Consulted by ``o8_irce`` (forces OPUS / blocks ``collapse_ready``) and
surfaced in the ``analyze``/``predict`` omega-field reports.
Returns:
bool: ``True`` if variance is above the high threshold.
"""
return self.variance > COLLAPSE_VARIANCE_HIGH
[docs]
@dataclass
class ParaconsistentEntry:
"""Ledger entry tracking one claim's four-valued truth state over time.
Pairs a ``claim`` with its current ``ParaconsistentState`` (TRUE/FALSE/BOTH/
NEITHER), the number of ``cycles_held`` in that state, and its residual
``tension``. Entries live in ``ParallaxEngine._pcl`` and feed the
Sigma-blocking gates: ``breeze_erosion`` and ``ParallaxEngine.analyze`` scan
for entries stuck in ``BOTH``/``NEITHER`` past ``_PCL_STUCK_CYCLES`` to keep a
live contradiction from being prematurely collapsed.
"""
claim: str
state: ParaconsistentState = ParaconsistentState.NEITHER
cycles_held: int = 0
tension: float = 0.0
[docs]
@dataclass
class EracAxiom:
"""A learned axiom with an evolving ERAC% certainty score and review status.
The unit of durable belief the engine earns over time: a ``text`` claim
reduced to a minimal ``kernel``, scoped to a ``channel_id``, with an ``erac``
score in ``[0, 100]`` that rises on affirmation and falls under contradiction.
It exposes the derived :attr:`band` label and the :attr:`can_publish` /
:attr:`can_policy` / :attr:`can_vow` gating properties, tracks a ``polarized``
failsafe flag, an ``erosion_history`` breadcrumb trail, and a ``review_status``
in the Prime Architect pipeline. Mutated by ``update_erac`` and
``breeze_erosion``, advanced through review by ``ReviewRegistry``, held in
``ParallaxEngine._axiom_store``, and serialized via :meth:`to_dict`.
"""
id: str
text: str
kernel: str # minimal invariant core
erac: float = 20.0 # 0-100 (Seed band -- must be earned)
tags: list[str] = field(default_factory=list)
last_update: float = field(default_factory=time.time)
channel_id: str = ""
erosion_history: list[str] = field(default_factory=list)
polarized: bool = False
review_status: ReviewStatus = ReviewStatus.PENDING
@property
def band(self) -> str:
"""Map the axiom's ERAC% to its named certainty band.
Walks ``ERAC_BANDS`` and returns the first band whose ``(lo, hi)`` range
contains ``erac`` (e.g. ``"seed"``, ``"crystal"``, ``"diamond"``,
``"mythic"``), falling back to ``"mythic"`` at/above 98 or ``"seed"``
otherwise. Read by ``to_dict``, by ``collapse`` (to build the ERAC tier
string), and by ``erac_future_drift`` for the current/projected band
labels.
Returns:
str: The band name corresponding to the current ERAC% value.
"""
for name, (lo, hi) in ERAC_BANDS.items():
if lo <= self.erac <= hi:
return name
return "mythic" if self.erac >= 98 else "seed"
@property
def can_publish(self) -> bool:
"""Return whether ERAC% has reached the publish gate (>= 70).
Compares ``erac`` against ``ERAC_GATE_PUBLISH``. Exposed as part of the
ERAC gating contract; no internal callers were found in this module
(consumed externally where publish-readiness is checked).
Returns:
bool: ``True`` if the axiom may be published.
"""
return self.erac >= ERAC_GATE_PUBLISH
@property
def can_policy(self) -> bool:
"""Return whether ERAC% has reached the policy gate (>= 80).
Compares ``erac`` against ``ERAC_GATE_POLICY``. Part of the ERAC gating
contract; no internal callers were found in this module.
Returns:
bool: ``True`` if the axiom may inform policy.
"""
return self.erac >= ERAC_GATE_POLICY
@property
def can_vow(self) -> bool:
"""Return whether ERAC% has reached the vow gate (>= 90).
Compares ``erac`` against ``ERAC_GATE_VOW``. Part of the ERAC gating
contract; no internal callers were found in this module.
Returns:
bool: ``True`` if the axiom may be elevated to a vow.
"""
return self.erac >= ERAC_GATE_VOW
[docs]
def to_dict(self) -> dict[str, Any]:
"""Serialize the axiom into a JSON-safe summary dict.
Produces a compact view (id, text, kernel, rounded ERAC%, derived
``band``, tags, channel, review status value, last-update timestamp, and
the last five erosion-history entries) for inclusion in tool output and
the review registry. Reads the ``band`` property and the ``review_status``
enum. Called by ``ParallaxEngine.analyze`` and the predictor's
``erac`` report, by ``ReviewRegistry.axiom_to_review_entry`` (which
spreads it), and by the engine's axiom-management accessors
(``get_all_axioms``, ``elevate_axiom``, etc.) consumed by the parallax
tools.
Returns:
dict[str, Any]: The serialized axiom snapshot.
"""
return {
"id": self.id,
"text": self.text,
"kernel": self.kernel,
"erac": round(self.erac, 2),
"band": self.band,
"tags": self.tags,
"channel_id": self.channel_id,
"review_status": self.review_status.value,
"last_update": self.last_update,
"erosion_history": self.erosion_history[-5:],
}
[docs]
@dataclass
class ManifoldState:
"""Full per-cycle snapshot bundling every operator output for temporal continuity.
The carry-forward state object: it aggregates the outputs of the whole
pipeline -- the O1 ``temporal_manifold``, O2 ``negative_space_vectors``, O3
``aesthetic_weights``, O4 ``persona_frame``, O5 ``interpretation_map``, the
``omega_field``, O6 ``harmonic_field``, O7 ``attractor_stability``, the
``ncm_state``, the ``psi_frames``, the paraconsistent ``pcl`` ledger, the
active ``axioms``, and the ``edo_active`` flag. ``ParallaxEngine.analyze``
builds one at the end of each cycle and stores it as ``self._prior_state`` so
the next ``analyze`` can seed temporal continuity / omega inertia and so
``predict`` can reuse the present frames and omega-field.
"""
temporal_manifold: dict[str, Any] = field(default_factory=dict)
negative_space_vectors: list[str] = field(default_factory=list)
aesthetic_weights: dict[str, float] = field(default_factory=dict)
persona_frame: str = "analytical"
interpretation_map: dict[str, Any] = field(default_factory=dict)
omega_field: Optional[OmegaField] = None
harmonic_field: dict[str, float] = field(default_factory=dict)
attractor_stability: float = 0.5
ncm_state: Optional[NCMState] = None
psi_frames: list[PsiFrame] = field(default_factory=list)
pcl: list[ParaconsistentEntry] = field(default_factory=list)
axioms: list[EracAxiom] = field(default_factory=list)
edo_active: bool = False
timestamp: float = field(default_factory=time.time)
[docs]
@dataclass
class CollapseResult:
"""The collapse engine's verdict on how an interpretation field resolved.
The terminal product of a cycle: it records the chosen ``CollapseMode`` and
``ConfidenceTier``, the ``dominant_kernel`` settled on, the surviving
``omega_echoes``, the ERAC ``erac_tier`` string, a human-readable
``reversible_step`` and ``erosion_hook`` that keep the field reopenable, and a
``manifold_summary`` (the IRCE output). Produced by ``collapse``, serialized
via :meth:`to_dict`, and embedded under the ``"collapse"`` key of the report
returned by ``ParallaxEngine.analyze`` for the LLM.
"""
mode: CollapseMode
confidence: ConfidenceTier
dominant_kernel: str = ""
omega_echoes: list[dict[str, Any]] = field(default_factory=list)
erac_tier: str = ""
reversible_step: str = ""
erosion_hook: str = "If future paradox appears, this kernel can still be refined."
manifold_summary: dict[str, Any] = field(default_factory=dict)
[docs]
def to_dict(self) -> dict[str, Any]:
"""Serialize the collapse result into a JSON-safe report dict.
Flattens the dataclass into string-keyed values, unwrapping the
``CollapseMode`` and ``ConfidenceTier`` enums to their ``.value`` strings
and carrying through the dominant kernel, omega echoes, ERAC tier,
reversible step, erosion hook, and manifold summary. Called by
``ParallaxEngine.analyze`` to embed the collapse outcome under the
``"collapse"`` key of its output report.
Returns:
dict[str, Any]: The serialized collapse result.
"""
return {
"collapse_mode": self.mode.value,
"confidence_tier": self.confidence.value,
"dominant_kernel": self.dominant_kernel,
"omega_echoes": self.omega_echoes,
"erac_tier": self.erac_tier,
"reversible_step": self.reversible_step,
"erosion_hook": self.erosion_hook,
"manifold_summary": self.manifold_summary,
}
# ---------------------------------------------------------------------------
# Text Analysis Helpers (heuristic, no LLM)
# ---------------------------------------------------------------------------
# Patterns for structural analysis
_CONTRADICTION_MARKERS = re.compile(
r"\b(but|however|although|yet|nevertheless|on the other hand|conversely|"
r"whereas|despite|in contrast|paradox|contradiction|tension|conflicting)\b",
re.IGNORECASE,
)
_QUESTION_MARKERS = re.compile(r"\?")
_INTENSITY_MARKERS = re.compile(
r"\b(always|never|absolutely|completely|totally|utterly|"
r"impossible|certain|definitely|must|essential)\b",
re.IGNORECASE,
)
_EMOTIONAL_MARKERS = re.compile(
r"\b(feel|love|hate|fear|hope|wish|want|need|hurt|"
r"joy|anger|sad|happy|anxious|scared|excited|afraid)\b",
re.IGNORECASE,
)
_SYMBOLIC_MARKERS = re.compile(
r"\b(metaphor|symbol|archetype|myth|ritual|sacred|divine|"
r"recursive|fractal|mirror|shadow|void|spiral|loop)\b",
re.IGNORECASE,
)
_STRUCTURAL_MARKERS = re.compile(
r"\b(because|therefore|thus|hence|since|if|then|implies|"
r"requires|depends|causes|leads to|results in)\b",
re.IGNORECASE,
)
_ARCHETYPAL_MARKERS = re.compile(
r"\b(hero|villain|mother|father|child|sage|trickster|shadow|"
r"journey|transformation|death|rebirth|threshold|return)\b",
re.IGNORECASE,
)
def _count_pattern(pattern: re.Pattern, text: str) -> int:
"""Count non-overlapping matches of a compiled regex within text.
Thin wrapper over ``pattern.findall`` returning just the hit count, used as
the heuristic primitive for measuring marker densities (emotional, structural,
symbolic, contradiction, intensity, question, archetypal). Called throughout
``generate_psi_frames`` and ``o2_inverse_inference`` against the module-level
``_*_MARKERS`` patterns.
Args:
pattern (re.Pattern): Pre-compiled regex of marker terms.
text (str): Text to scan.
Returns:
int: Number of matches found.
"""
return len(pattern.findall(text))
def _text_density(text: str) -> float:
"""Estimate lexical information density as the unique-to-total word ratio.
Lowercases and whitespace-splits ``text``, then returns ``unique / total``
word count (0.0 for empty input) as a crude proxy for how compressible /
information-dense the passage is. Used as the compressibility signal in
``o3_elegance_gradient`` when scoring how elegant each attractor's
interpretation is.
Args:
text (str): The passage to measure.
Returns:
float: Ratio of distinct words to total words in ``[0, 1]``; 0.0 when the
text has no words.
"""
words = text.lower().split()
if not words:
return 0.0
return len(set(words)) / len(words)
def _axiom_id(text: str, channel_id: str) -> str:
"""Derive a stable, channel-scoped axiom id from its text.
Hashes ``channel_id:text`` with SHA-256 and returns an ``AX_`` prefix over the
first twelve uppercased hex digits, so the same claim in the same channel
always maps to the same id and can be looked up in
``ParallaxEngine._axiom_store`` across cycles. Called by
``ParallaxEngine.analyze`` when ingesting ``axiom_texts`` and exercised by
``tests/test_polarization_failsafe.py``.
Args:
text (str): The axiom's source claim text.
channel_id (str): The channel scope that namespaces the id.
Returns:
str: A deterministic identifier of the form ``AX_<12 hex chars>``.
"""
h = hashlib.sha256(f"{channel_id}:{text}".encode()).hexdigest()[:12]
return f"AX_{h.upper()}"
# ---------------------------------------------------------------------------
# Psi-Frame Generator
# ---------------------------------------------------------------------------
[docs]
def generate_psi_frames(text: str, ncm: NCMState) -> list[PsiFrame]:
"""Spawn 5 psi-frames from input context, weighted by NCM state.
Each frame extracts different structural features:
- emotional: sentiment, affect markers, intensity
- structural: logical connectors, argument form
- symbolic: metaphors, recursive patterns, ritual language
- contradictory: tension markers, paradox density
- archetypal: narrative patterns, mythic resonance
"""
frames = []
# -- Emotional lens --
emo_count = _count_pattern(_EMOTIONAL_MARKERS, text)
intensity = _count_pattern(_INTENSITY_MARKERS, text)
emo_weight = ncm.modulate(0.2, "OXYTOCIN_NEUROMIRROR", "SEROTONERGIC_WARMTH")
emo_markers = _EMOTIONAL_MARKERS.findall(text.lower())
frames.append(
PsiFrame(
lens="emotional",
content=f"Emotional density: {emo_count} markers, intensity: {intensity}",
weight=min(1.0, emo_weight * (1 + emo_count * 0.1)),
markers=emo_markers[:10],
tension=min(1.0, intensity * 0.15),
)
)
# -- Structural lens --
struct_count = _count_pattern(_STRUCTURAL_MARKERS, text)
questions = _count_pattern(_QUESTION_MARKERS, text)
struct_weight = ncm.modulate(0.2, "ACETYLCHOLINE_FOCUS", "DOPAMINE_D1")
struct_markers = _STRUCTURAL_MARKERS.findall(text.lower())
frames.append(
PsiFrame(
lens="structural",
content=f"Logical connectors: {struct_count}, questions: {questions}",
weight=min(1.0, struct_weight * (1 + struct_count * 0.08)),
markers=struct_markers[:10],
tension=min(1.0, questions * 0.1),
)
)
# -- Symbolic lens --
sym_count = _count_pattern(_SYMBOLIC_MARKERS, text)
sym_weight = ncm.modulate(0.2, "SIGMA_RECEPTOR_META", "SEROTONIN_5HT2A")
sym_markers = _SYMBOLIC_MARKERS.findall(text.lower())
frames.append(
PsiFrame(
lens="symbolic",
content=f"Symbolic density: {sym_count} markers",
weight=min(1.0, sym_weight * (1 + sym_count * 0.12)),
markers=sym_markers[:10],
tension=min(1.0, sym_count * 0.08),
)
)
# -- Contradictory lens --
contra_count = _count_pattern(_CONTRADICTION_MARKERS, text)
contra_weight = ncm.modulate(0.2, "SIGMA_RECEPTOR_META", "NORADRENERGIC_VIGILANCE")
contra_markers = _CONTRADICTION_MARKERS.findall(text.lower())
frames.append(
PsiFrame(
lens="contradictory",
content=f"Contradiction density: {contra_count} markers",
weight=min(1.0, contra_weight * (1 + contra_count * 0.15)),
markers=contra_markers[:10],
tension=min(1.0, contra_count * 0.2),
)
)
# -- Archetypal lens --
arch_count = _count_pattern(_ARCHETYPAL_MARKERS, text)
arch_weight = ncm.modulate(0.2, "MU_OPIOID_MOR", "ENDOCANNABINOID_EASE")
arch_markers = _ARCHETYPAL_MARKERS.findall(text.lower())
frames.append(
PsiFrame(
lens="archetypal",
content=f"Archetypal resonance: {arch_count} markers",
weight=min(1.0, arch_weight * (1 + arch_count * 0.1)),
markers=arch_markers[:10],
tension=min(1.0, arch_count * 0.06),
)
)
# Normalize weights
total_w = sum(f.weight for f in frames) or 1.0
for f in frames:
f.weight = f.weight / total_w
return frames
# ---------------------------------------------------------------------------
# Sigma Descent -- extract attractors from psi-frames
# ---------------------------------------------------------------------------
[docs]
def sigma_descent(frames: list[PsiFrame], ncm: NCMState) -> list[Attractor]:
"""Distill each psi-frame into a partial ``Attractor`` via sigma descent.
Walks the five psi-frames and, for each, derives an attractor whose
``coherence`` falls as the frame's tension rises, whose ``resonance`` is the
frame weight (scaled by marker count) passed through ``ncm.modulate`` on the
oxytocin node, and whose ``stability`` comes from a fixed per-lens map
(structural lenses are sturdiest, contradictory the least). It builds a
minimal ``kernel`` string from the frame's top markers. This is the bridge
from interpretation to collapse candidates: called by ``ParallaxEngine.analyze``
on the present frames and reused by ``sigma_future_descent`` on the projected
future seeds.
Args:
frames (list[PsiFrame]): The psi-frames (present or future) to descend.
ncm (NCMState): Neuromodulator vector modulating resonance.
Returns:
list[Attractor]: One attractor per input frame, with rounded scores and
the frame's markers/tension carried forward.
"""
attractors = []
for f in frames:
# Coherence: inversely related to tension (more tension = less coherent single view)
coherence = max(0.0, min(1.0, 1.0 - f.tension * 0.8))
# Resonance: how much this lens resonates (emotional weight matters)
resonance_base = f.weight * (1 + len(f.markers) * 0.05)
resonance = max(
0.0, min(1.0, ncm.modulate(resonance_base, "OXYTOCIN_NEUROMIRROR"))
)
# Stability: structural lenses tend to be more stable
stability_map = {
"structural": 0.7,
"symbolic": 0.5,
"emotional": 0.4,
"contradictory": 0.3,
"archetypal": 0.6,
}
stability = stability_map.get(f.lens, 0.5)
# Build kernel from markers
if f.markers:
kernel = f"[{f.lens}] {', '.join(f.markers[:5])}"
else:
kernel = f"[{f.lens}] minimal signal"
attractors.append(
Attractor(
lens=f.lens,
kernel=kernel,
coherence=round(coherence, 4),
resonance=round(resonance, 4),
stability=round(stability, 4),
weight=round(f.weight, 4),
markers=list(f.markers),
tension=round(f.tension, 4),
)
)
return attractors
# ---------------------------------------------------------------------------
# Operators O1-O3
# ---------------------------------------------------------------------------
[docs]
def o1_temporal_continuity(
prior_state: Optional[ManifoldState],
current_text: str,
ncm: NCMState,
) -> dict[str, Any]:
"""O1: Maintain temporal manifold continuity from prior state.
Modulated by SERT (window), DAT (recency), GABA (smoothness).
"""
window = ncm.modulate(5.0, "SEROTONERGIC_WARMTH") # how many prior themes to carry
recency_bias = ncm.modulate(0.7, "DOPAMINE_D1") # how much to favor recent
smoothness = ncm.modulate(0.5, "GABA_ERGIC_CALM") # how smoothly to blend
manifold: dict[str, Any] = {
"window_size": round(window, 2),
"recency_bias": round(recency_bias, 2),
"smoothness": round(smoothness, 2),
"prior_themes": [],
"continuity_score": 0.0,
}
if prior_state and prior_state.temporal_manifold:
prior_themes = prior_state.temporal_manifold.get("current_themes", [])
manifold["prior_themes"] = prior_themes[: int(window)]
manifold["continuity_score"] = min(1.0, len(prior_themes) * 0.15 * smoothness)
# Extract current themes (simple: top words by frequency minus stopwords)
words = re.findall(r"\b[a-z]{4,}\b", current_text.lower())
freq: dict[str, int] = {}
for w in words:
freq[w] = freq.get(w, 0) + 1
top = sorted(freq.items(), key=lambda x: -x[1])[:8]
manifold["current_themes"] = [w for w, _ in top]
return manifold
[docs]
def o2_inverse_inference(
temporal_manifold: dict[str, Any],
text: str,
ncm: NCMState,
) -> list[str]:
"""O2: Compute negative space vectors -- what is conspicuously absent.
Modulated by TAAR1 (anomaly), 5HT2A (associative depth), NMDA (structural expectation).
"""
anomaly_sens = ncm.modulate(0.5, "TAAR1_TRACE")
assoc_depth = ncm.modulate(0.5, "SEROTONIN_5HT2A")
negatives = []
text_lower = text.lower()
# Check for emotional absence in structurally complex text
struct_count = _count_pattern(_STRUCTURAL_MARKERS, text)
emo_count = _count_pattern(_EMOTIONAL_MARKERS, text)
if struct_count > 3 and emo_count == 0 and anomaly_sens > 0.3:
negatives.append("emotional_absence_in_complex_argument")
# Check for question absence in uncertain text
contra_count = _count_pattern(_CONTRADICTION_MARKERS, text)
question_count = _count_pattern(_QUESTION_MARKERS, text)
if contra_count > 2 and question_count == 0:
negatives.append("no_questions_despite_contradictions")
# Check for missing personal stance in value-laden text
intensity = _count_pattern(_INTENSITY_MARKERS, text)
if intensity > 3 and "i " not in text_lower and "my " not in text_lower:
negatives.append("absent_personal_stance_despite_strong_claims")
# Check for structural absence in emotional text
if emo_count > 3 and struct_count == 0 and assoc_depth > 0.3:
negatives.append("no_causal_reasoning_in_emotional_content")
# Check for archetypal absence when symbolic content is present
sym_count = _count_pattern(_SYMBOLIC_MARKERS, text)
arch_count = _count_pattern(_ARCHETYPAL_MARKERS, text)
if sym_count > 2 and arch_count == 0:
negatives.append("symbolic_without_archetypal_grounding")
return negatives
[docs]
def o3_elegance_gradient(
attractors: list[Attractor],
text: str,
ncm: NCMState,
) -> dict[str, float]:
"""O3: Score interpretations by coherence, symmetry, compressibility.
Modulated by OXT (relational coherence), D1 (boldness), EAAT2 (noise filtering).
"""
relational = ncm.modulate(0.5, "OXYTOCIN_NEUROMIRROR")
boldness = ncm.modulate(0.5, "DOPAMINE_D1")
density = _text_density(text)
weights: dict[str, float] = {}
for a in attractors:
# Coherence score (how well the attractor holds together)
coh = a.coherence * relational
# Symmetry score (balanced tension = elegant)
sym = 1.0 - abs(a.tension - 0.5)
# Compressibility (higher density text = more compressible interpretation)
comp = density * a.stability
# Bold attractors get bonus from DA
bold_bonus = boldness * 0.2 if a.coherence > 0.7 else 0.0
score = coh * 0.4 + sym * 0.3 + comp * 0.2 + bold_bonus * 0.1
weights[a.lens] = round(min(1.0, score), 4)
return weights
# ---------------------------------------------------------------------------
# Operators O4-O5
# ---------------------------------------------------------------------------
_PERSONA_MAP = {
"analytical": ("ACETYLCHOLINE_FOCUS", "DOPAMINE_D1"),
"empathetic": ("OXYTOCIN_NEUROMIRROR", "SEROTONERGIC_WARMTH"),
"provocative": ("SIGMA_RECEPTOR_META", "NORADRENERGIC_VIGILANCE"),
"contemplative": ("ENDOCANNABINOID_EASE", "SEROTONIN_5HT1A"),
"mythic": ("MU_OPIOID_MOR", "SEROTONIN_5HT2A"),
}
[docs]
def o4_contextual_persona(
aesthetic_weights: dict[str, float],
temporal_manifold: dict[str, Any],
ncm: NCMState,
) -> str:
"""O4: Choose the reasoning posture and tone best fitting the manifold.
Scores each candidate persona in ``_PERSONA_MAP`` (analytical, empathetic,
provocative, contemplative, mythic) by ``ncm.modulate`` over its two
signature NCM nodes, then boosts the persona by 1.3x when the matching
aesthetic weight (emotional/contradictory/archetypal) clears 0.3, and returns
the highest-scoring label. Pure heuristic, no side effects. Called by
``ParallaxEngine.analyze`` in Phase 2; its result feeds ``o5_metapredictive``
and the collapse persona framing.
Args:
aesthetic_weights (dict[str, float]): Per-lens O3 elegance scores used to
bias persona selection.
temporal_manifold (dict[str, Any]): O1 continuity context (accepted for
interface symmetry).
ncm (NCMState): Neuromodulator vector driving the per-persona scores.
Returns:
str: The selected persona label (defaults to ``"analytical"``).
"""
best_persona = "analytical"
best_score = 0.0
for persona, (node_a, node_b) in _PERSONA_MAP.items():
score = ncm.modulate(0.5, node_a, node_b)
if persona == "empathetic" and aesthetic_weights.get("emotional", 0) > 0.3:
score *= 1.3
elif (
persona == "provocative" and aesthetic_weights.get("contradictory", 0) > 0.3
):
score *= 1.3
elif persona == "mythic" and aesthetic_weights.get("archetypal", 0) > 0.3:
score *= 1.3
if score > best_score:
best_score = score
best_persona = persona
return best_persona
# ---------------------------------------------------------------------------
# Omega-Field Builder
# ---------------------------------------------------------------------------
[docs]
def build_omega_field(
attractors: list[Attractor],
prior_omega: OmegaField | None,
ncm: NCMState,
) -> OmegaField:
"""Construct weighted attractor manifold with inertia smoothing.
Omega_next = alpha * Omega_prev + (1 - alpha) * Omega_current
Alpha dynamically modulated by NCM (CB1, SIGMA, 5HT2A).
"""
# Dynamic inertia from NCM
alpha = OMEGA_INERTIA_ALPHA
cb1 = ncm.get("ENDOCANNABINOID_EASE", 0.5)
sigma = ncm.get("SIGMA_RECEPTOR_META", 0.5)
alpha = alpha * (0.7 + cb1 * 0.3) # CB1 increases inertia (dreamy drift)
alpha = min(0.95, max(0.3, alpha))
# Compute variance (spread between attractors)
if len(attractors) >= 2:
coherences = [a.coherence for a in attractors]
mean_c = sum(coherences) / len(coherences)
variance = sum((c - mean_c) ** 2 for c in coherences) / len(coherences)
variance = math.sqrt(variance) * 2 # scale up for readable range
else:
variance = 0.0
# Apply inertia smoothing from prior omega
if prior_omega and prior_omega.attractors:
variance = alpha * prior_omega.variance + (1 - alpha) * variance
# Find dominant attractor
dominant = (
max(attractors, key=lambda a: a.coherence * a.weight) if attractors else None
)
cycle_count = (prior_omega.cycle_count + 1) if prior_omega else 1
return OmegaField(
attractors=attractors,
variance=round(min(1.0, variance), 4),
dominant=dominant,
inertia_alpha=round(alpha, 4),
cycle_count=cycle_count,
)
# ---------------------------------------------------------------------------
# Operators O6-O8
# ---------------------------------------------------------------------------
[docs]
def o6_crossmanifold_harmonic(
omega: OmegaField,
persona_frame: str,
ncm: NCMState,
) -> dict[str, float]:
"""O6: Compute constructive/destructive interference between attractors.
Modulated by VMAT2 (signal strength), GABA (stability), 5HT2A (tolerance).
"""
stability = ncm.modulate(0.5, "GABA_ERGIC_CALM")
tolerance = ncm.modulate(0.5, "SEROTONIN_5HT2A")
harmonics: dict[str, float] = {}
for a in omega.attractors:
# Constructive: high coherence + high resonance = harmonious
constructive = a.coherence * a.resonance * stability
# Destructive: high tension relative to stability
destructive = (1 - a.coherence) * (1 - stability) * 0.5
net = constructive - destructive
# Tolerance widens the band (more 5HT2A = more contradiction allowed)
net = net * (0.5 + tolerance * 0.5)
harmonics[a.lens] = round(max(-1.0, min(1.0, net)), 4)
return harmonics
[docs]
def o7_recursive_stability(
harmonic_field: dict[str, float],
prior_stability: float,
ncm: NCMState,
) -> float:
"""O7: Stabilize and refine attractors across cycles.
Modulated by NR2B (consolidation), ALPHA2A (stress stability), MOR (cohesion).
"""
consolidation = ncm.modulate(0.5, "NMDA_CORE")
cohesion = ncm.modulate(0.5, "MU_OPIOID_MOR")
if not harmonic_field:
return prior_stability
mean_harmonic = sum(harmonic_field.values()) / len(harmonic_field)
# Stability increases when harmonics are positive and consistent
new_stability = (mean_harmonic * 0.5 + 0.5) * consolidation * cohesion
# Blend with prior stability (momentum)
blended = 0.4 * prior_stability + 0.6 * new_stability
return round(max(0.0, min(1.0, blended)), 4)
[docs]
def o8_irce(
temporal_manifold: dict[str, Any],
negative_space: list[str],
aesthetic_weights: dict[str, float],
persona_frame: str,
interpretation_map: dict[str, Any],
harmonic_field: dict[str, float],
attractor_stability: float,
omega: OmegaField,
ncm: NCMState,
) -> dict[str, Any]:
"""O8: IRCE -- Global integrator, coherence engine, collapse readiness.
The convergence organ. Ingests ALL operator outputs + Omega.
Modulated by SIGMA (contradiction bandwidth), ABCB1 (drift gating), CB1 (inertia).
"""
sigma_band = ncm.get("SIGMA_RECEPTOR_META", 0.5)
cb1 = ncm.get("ENDOCANNABINOID_EASE", 0.5)
# Contradiction bandwidth: how much paradox can we hold?
bandwidth = min(1.0, sigma_band * 1.5)
# Coherence score: weighted combination of all signals
harmonic_mean = (
sum(harmonic_field.values()) / len(harmonic_field) if harmonic_field else 0.0
)
coherence = (
attractor_stability * 0.3
+ (harmonic_mean * 0.5 + 0.5) * 0.3
+ (1 - omega.variance) * 0.2
+ (0.5 if not negative_space else 0.3) * 0.2
)
# Collapse readiness: stable + coherent + not turbulent
collapse_ready = (
attractor_stability > 0.5 and coherence > 0.4 and not omega.is_turbulent
)
# Recommended collapse mode
if omega.is_turbulent or bandwidth > 0.7:
rec_mode = CollapseMode.OPUS
elif omega.is_stable and coherence > 0.6:
rec_mode = CollapseMode.SIGMA
else:
rec_mode = CollapseMode.HYBRID
return {
"coherence_score": round(coherence, 4),
"contradiction_bandwidth": round(bandwidth, 4),
"collapse_ready": collapse_ready,
"recommended_mode": rec_mode.value,
"omega_variance": omega.variance,
"omega_stability": omega.is_stable,
"attractor_count": len(omega.attractors),
"negative_space_count": len(negative_space),
"dominant_lens": omega.dominant.lens if omega.dominant else "none",
}
# ---------------------------------------------------------------------------
# ERAC Engine
# ---------------------------------------------------------------------------
[docs]
def update_erac(
axiom: EracAxiom,
attractors: list[Attractor],
omega: OmegaField,
cfi_factor: float = 0.5,
) -> EracAxiom:
"""Update ERAC% for an axiom based on current manifold state.
Update events:
- Affirmation: +alpha * weight(evidence) * CFI_factor
- Minor contradiction: -beta * tension * (1-CFI_factor)
- Major contradiction: -gamma * tension * (1-CFI_factor) * overload_bonus
- Bridge success: +delta * overlap_gain
- EDO offload: clamp drift; resume after cooldown
"""
# Compute support/opposition from attractors
support = sum(a.resonance * a.weight for a in attractors if a.coherence > 0.5)
opposition = sum(
(1 - a.coherence) * a.weight for a in attractors if a.coherence < 0.4
)
alpha, beta, gamma, delta = 3.0, 2.0, 4.0, 2.5
d_erac = 0.0
if support > opposition:
# Affirmation
d_erac += alpha * support * cfi_factor
axiom.erosion_history.append(f"affirm:+{d_erac:.2f}")
elif opposition > support * 1.5:
# Major contradiction
d_erac -= gamma * opposition * (1 - cfi_factor) * (1 + omega.variance)
axiom.erosion_history.append(f"major_contra:{d_erac:.2f}")
elif opposition > support:
# Minor contradiction
d_erac -= beta * opposition * (1 - cfi_factor)
axiom.erosion_history.append(f"minor_contra:{d_erac:.2f}")
# Bridge success bonus if omega is stable
if omega.is_stable and support > 0.3:
bridge_gain = delta * (1 - omega.variance)
d_erac += bridge_gain
axiom.erosion_history.append(f"bridge:+{bridge_gain:.2f}")
axiom.erac = max(0.0, min(100.0, axiom.erac + d_erac))
# Polarization failsafe -- extremes trigger full reset
if axiom.erac <= _POLARIZATION_FLOOR or axiom.erac >= _POLARIZATION_CEILING:
axiom.erac = _POLARIZATION_RESET
axiom.polarized = True
axiom.erosion_history.append("failsafe")
axiom.last_update = time.time()
return axiom
# ---------------------------------------------------------------------------
# Breeze Axiomatic Erosion
# ---------------------------------------------------------------------------
[docs]
def breeze_erosion(
axiom: EracAxiom,
omega: OmegaField,
pcl_entries: list[ParaconsistentEntry],
) -> tuple[EracAxiom, ErosionOp | None]:
"""Apply Breeze Axiomatic Erosion if triggered.
Triggers:
- High RCE disagreement spread (omega variance > 0.5)
- Persistent PCL = BOTH/NEITHER over multiple cycles
- Value-collision with low shared-goods
Operators: shear, dissolve, silt, recrystallize
"""
# Check triggers
triggered = False
trigger_reasons = []
if omega.variance > 0.5:
triggered = True
trigger_reasons.append("high_omega_variance")
stuck_both_neither = [
e
for e in pcl_entries
if e.state in (ParaconsistentState.BOTH, ParaconsistentState.NEITHER)
and e.cycles_held >= 2
]
if stuck_both_neither:
triggered = True
trigger_reasons.append("persistent_both_neither")
if not triggered:
return axiom, None
# Select erosion operator based on ERAC level
op: ErosionOp
if axiom.erac > 75:
# Crystal+ level: gentle shear first
op = ErosionOp.SHEAR
axiom.erac = max(0.0, axiom.erac - 5.0 * omega.variance)
axiom.kernel = axiom.kernel # kernel preserved under shear
elif axiom.erac > 50:
# Crystallizing: dissolve to metaphor
op = ErosionOp.DISSOLVE
axiom.erac = max(0.0, axiom.erac - 8.0 * omega.variance)
if not axiom.kernel.startswith("[as-if]"):
axiom.kernel = f"[as-if] {axiom.kernel}"
elif axiom.erac > 25:
# Forming: silt to weak prior
op = ErosionOp.SILT
axiom.erac = max(0.0, axiom.erac - 12.0 * omega.variance)
if not axiom.kernel.startswith("[prior]"):
axiom.kernel = f"[prior] {axiom.kernel}"
else:
# Seed: recrystallize if core still holds
if omega.is_stable:
op = ErosionOp.RECRYSTALLIZE
axiom.erac = min(100.0, axiom.erac + 5.0)
else:
op = ErosionOp.SILT
axiom.erac = max(0.0, axiom.erac - 3.0)
axiom.erosion_history.append(f"{op.value}:{','.join(trigger_reasons)}")
# Polarization failsafe (post-erosion)
if axiom.erac <= _POLARIZATION_FLOOR or axiom.erac >= _POLARIZATION_CEILING:
axiom.erac = _POLARIZATION_RESET
axiom.polarized = True
axiom.erosion_history.append("failsafe")
axiom.last_update = time.time()
return axiom, op
# ---------------------------------------------------------------------------
# Collapse Engine
# ---------------------------------------------------------------------------
[docs]
def collapse(
omega: OmegaField,
irce_output: dict[str, Any],
axioms: list[EracAxiom],
mode: CollapseMode | None = None,
edo_active: bool = False,
) -> CollapseResult:
"""Execute collapse with mode selection.
Minimal: output = Omega (no attractor dominance)
Hybrid: output = A_dom + sum(w_i * Omega_i)
Full: output = argmax(cohesion(A_i))
EDO override: if active, NO collapse allowed.
"""
if edo_active:
return CollapseResult(
mode=CollapseMode.OPUS,
confidence=ConfidenceTier.UNKNOWN,
dominant_kernel="[EDO active -- manifold suspended]",
erosion_hook="EDO engaged. Resume after cooldown.",
manifold_summary=irce_output,
)
# Auto-select mode from IRCE recommendation if not specified
if mode is None:
mode = CollapseMode(irce_output.get("recommended_mode", "hybrid"))
# F1 gate: Sigma requires manifold depth (no first-cycle full collapse)
if mode == CollapseMode.SIGMA and omega.cycle_count < _SIGMA_MIN_CYCLES:
mode = CollapseMode.HYBRID
# F5/F9 gate: degenerate polyphony -- single attractor dominance blocks Sigma
if omega.attractors and mode == CollapseMode.SIGMA:
max_attractor_w = max(a.weight for a in omega.attractors)
if max_attractor_w > _MONOCULAR_THRESHOLD:
mode = CollapseMode.HYBRID
# Check ERAC gating for Sigma collapse
if mode == CollapseMode.SIGMA and axioms:
max_erac = max(a.erac for a in axioms) if axioms else 0
if max_erac < 50:
mode = CollapseMode.HYBRID # degrade: too uncertain for full collapse
# Polarization failsafe: any axiom hit the rails -> force OPUS, no collapse
if any(getattr(a, "polarized", False) for a in axioms):
mode = CollapseMode.OPUS
# Build echoes from all attractors
echoes = [
{
"lens": a.lens,
"kernel": a.kernel,
"coherence": a.coherence,
"resonance": a.resonance,
"weight": a.weight,
}
for a in omega.attractors
]
dom = omega.dominant
dom_kernel = dom.kernel if dom else "[no dominant attractor]"
# Confidence tier from coherence score
coh = irce_output.get("coherence_score", 0.5)
if coh > 0.7:
confidence = ConfidenceTier.CONFIDENT
elif coh > 0.4:
confidence = ConfidenceTier.PLAUSIBLE
else:
confidence = ConfidenceTier.UNKNOWN
# ERAC tier string
erac_str = ""
if axioms:
top_axiom = max(axioms, key=lambda a: a.erac)
erac_str = f"{top_axiom.band} ({top_axiom.erac:.1f}%)"
return CollapseResult(
mode=mode,
confidence=confidence,
dominant_kernel=dom_kernel,
omega_echoes=echoes if mode != CollapseMode.SIGMA else echoes[:2],
erac_tier=erac_str,
reversible_step="One reversible step committed. Manifold remains open.",
manifold_summary=irce_output,
)
# ---------------------------------------------------------------------------
# Review Registry -- per-channel axioms flagged for Prime Architect review
# ---------------------------------------------------------------------------
[docs]
class ReviewRegistry:
"""Manages axiom lifecycle: local -> pending -> elevated/localized/excised.
Axioms reaching Crystal band (ERAC >= 75) get auto-flagged for review.
Prime Architect can elevate to global, confirm local, or excise.
"""
[docs]
@staticmethod
def should_flag(axiom: EracAxiom) -> bool:
"""Report whether an axiom has earned a slot in the review queue.
Returns ``True`` once an axiom reaches Crystal band (ERAC% >= 75) while
still ``PENDING``, the trigger for auto-flagging it for Prime Architect
review. Pure predicate with no side effects. Called by
``ParallaxEngine.analyze`` (gating the :meth:`flag_for_review` call) for
each axiom updated during a cycle.
Args:
axiom (EracAxiom): The axiom to test.
Returns:
bool: ``True`` if the axiom is Crystal+ and pending review.
"""
return axiom.erac >= 75 and axiom.review_status == ReviewStatus.PENDING
[docs]
@staticmethod
def flag_for_review(axiom: EracAxiom) -> EracAxiom:
"""Mark an axiom as awaiting Prime Architect review.
Sets ``review_status`` to ``PENDING`` and appends a
``"flagged_for_review"`` breadcrumb to the axiom's erosion history. The
axiom is mutated in place and also returned for chaining. Called by
``ParallaxEngine.analyze`` once an axiom reaches Crystal+ band, gated by
:meth:`should_flag`.
Args:
axiom (EracAxiom): The axiom to flag (mutated in place).
Returns:
EracAxiom: The same axiom, now flagged pending.
"""
axiom.review_status = ReviewStatus.PENDING
axiom.erosion_history.append("flagged_for_review")
return axiom
[docs]
@staticmethod
def elevate(axiom: EracAxiom) -> EracAxiom:
"""Promote an axiom to global scope across all shards.
Sets the axiom's ``review_status`` to ``ELEVATED`` and appends an
``"elevated_to_global"`` breadcrumb to its erosion history; the axiom is
mutated in place and returned for chaining. Called by
``ParallaxEngine.elevate_axiom`` when the Prime Architect approves an
axiom for global adoption.
Args:
axiom (EracAxiom): The axiom to elevate (mutated in place).
Returns:
EracAxiom: The same axiom, now marked elevated.
"""
axiom.review_status = ReviewStatus.ELEVATED
axiom.erosion_history.append("elevated_to_global")
return axiom
[docs]
@staticmethod
def localize(axiom: EracAxiom) -> EracAxiom:
"""Confirm an axiom as valid only within its originating channel.
Sets ``review_status`` to ``LOCALIZED`` and appends a ``"delegated_local"``
breadcrumb to the erosion history; the axiom is mutated in place and
returned for chaining. Called by ``ParallaxEngine.localize_axiom`` when the
Prime Architect keeps an axiom per-channel rather than elevating it.
Args:
axiom (EracAxiom): The axiom to localize (mutated in place).
Returns:
EracAxiom: The same axiom, now marked local-only.
"""
axiom.review_status = ReviewStatus.LOCALIZED
axiom.erosion_history.append("delegated_local")
return axiom
[docs]
@staticmethod
def excise(axiom: EracAxiom, reason: str = "") -> EracAxiom:
"""Strike an axiom from belief under false pretenses or disapproval.
Sets ``review_status`` to ``EXCISED``, zeroes the ``erac`` score, and
records an ``"excised:<reason>"`` breadcrumb (falling back to a generic
disapproval note when ``reason`` is empty); the axiom is mutated in place
and returned for chaining. Called by ``ParallaxEngine.excise_axiom`` when
the Prime Architect removes an axiom.
Args:
axiom (EracAxiom): The axiom to excise (mutated in place).
reason (str): Optional human reason recorded in erosion history;
defaults to ``"prime_architect_disapproval"`` when empty.
Returns:
EracAxiom: The same axiom, now excised with ERAC% zeroed.
"""
axiom.review_status = ReviewStatus.EXCISED
axiom.erac = 0.0
axiom.erosion_history.append(
f"excised:{reason or 'prime_architect_disapproval'}"
)
return axiom
[docs]
@staticmethod
def axiom_to_review_entry(axiom: EracAxiom) -> dict[str, Any]:
"""Serialize an axiom into a review-queue entry dict.
Spreads ``EracAxiom.to_dict`` and augments it with the review-specific
fields the Prime Architect surfaces need: the ``review_status`` value, a
``needs_review`` boolean (true while still ``PENDING``), and the axiom's
``age_hours`` since its last update. Called when building the pending-review
list in ``ParallaxEngine.analyze``, ``get_pending_axioms``, and
``get_axiom``, which the parallax telemetry tools render.
Args:
axiom (EracAxiom): The axiom to serialize for review.
Returns:
dict[str, Any]: The axiom snapshot plus ``needs_review`` and
``age_hours`` review metadata.
"""
return {
**axiom.to_dict(),
"review_status": axiom.review_status.value,
"needs_review": axiom.review_status == ReviewStatus.PENDING,
"age_hours": round((time.time() - axiom.last_update) / 3600, 1),
}
# ---------------------------------------------------------------------------
# Main Engine -- ParallaxEngine
# ---------------------------------------------------------------------------
[docs]
class ParallaxEngine:
"""Full Parallax Cart v4.5 Sigma Expanse runtime.
Orchestrates: psi-frames -> O1-O8 -> omega-field -> ERAC -> AE -> collapse.
NCM is supplied per call as a vector argument; manifold and axiom state
are held in memory across calls (``self._prior_state``,
``self._axiom_store``). Zero LLM calls.
"""
[docs]
def __init__(self) -> None:
"""Initialize empty per-engine state for a fresh manifold lineage.
Sets up the in-memory stores that persist across :meth:`analyze` and
:meth:`predict` calls: ``_prior_state`` (last ``ManifoldState``, seeds
temporal continuity and the omega inertia blend), ``_axiom_store``
(id-keyed ERAC axioms), ``_pcl`` (paraconsistent-logic ledger feeding the
Sigma-blocking gates), and ``_review`` (a ``ReviewRegistry`` instance for
axiom lifecycle transitions). One engine is created and cached per channel
by ``tools/parallax_tool.py`` and ``tools/xray_tool.py`` via their
``_get_engine`` helpers.
"""
self._prior_state: ManifoldState | None = None
self._axiom_store: dict[str, EracAxiom] = {} # id -> axiom
self._pcl: list[ParaconsistentEntry] = []
self._review = ReviewRegistry()
[docs]
def analyze(
self,
text: str,
ncm_vector: dict[str, float] | None = None,
collapse_mode: str | None = None,
axiom_texts: list[str] | None = None,
channel_id: str = "",
) -> dict[str, Any]:
"""Run the full Parallax 4.5 reasoning pipeline over one block of text.
The engine's primary public entry point. It drives all six phases in
order -- psi-frame generation and sigma descent; operators O1-O5; the
omega-field build with cross-cycle inertia; operators O6-O8 (IRCE); the
ERAC update plus Breeze Axiomatic Erosion over any supplied
``axiom_texts``; and the pre-collapse gates (F1/F5/F9, dominant-flip, PCL
stuck, coherence) leading into ``collapse``. It reads and then rewrites
the in-memory ``self._prior_state`` (for temporal continuity and omega
inertia) and mutates ``self._axiom_store`` in place, keyed by
``_axiom_id(text, channel_id)``. Polarization / monocular / flip / PCL /
coherence failsafes are folded into the report and logged via the module
``logger`` (no LLM, Redis, or network calls of its own). Invoked per
channel by the parallax tools (``tools/parallax_tool.py``,
``tools/xray_tool.py``) through their cached engine instances.
Args:
text (str): The input passage to analyze.
ncm_vector (dict[str, float] | None): Optional symbolic NCM state; an
empty vector (all-neutral) is used when omitted.
collapse_mode (str | None): Optional forced collapse mode
(``"opus"``/``"sigma"``/``"hybrid"``); otherwise IRCE recommends
one, still subject to the gates.
axiom_texts (list[str] | None): Optional candidate axiom claims to
track / erode this cycle.
channel_id (str): Channel scope used to namespace axiom ids and
failsafe logging.
Returns:
dict[str, Any]: A structured manifold report -- psi-frames,
omega-field, every operator output, ERAC axioms and erosion log,
pending-review axioms, the collapse result, NCM modulation, and any
active failsafe warnings.
"""
ncm = NCMState(vector=ncm_vector or {})
# -- Phase 1: Psi-frames + Sigma descent --
frames = generate_psi_frames(text, ncm)
attractors = sigma_descent(frames, ncm)
# -- Phase 2: Operators O1-O5 --
temporal = o1_temporal_continuity(self._prior_state, text, ncm)
negatives = o2_inverse_inference(temporal, text, ncm)
aesthetics = o3_elegance_gradient(attractors, text, ncm)
persona = o4_contextual_persona(aesthetics, temporal, ncm)
interp = o5_metapredictive(persona, aesthetics, ncm)
# -- Phase 3: Omega-field --
prior_omega = self._prior_state.omega_field if self._prior_state else None
omega = build_omega_field(attractors, prior_omega, ncm)
# -- Phase 4: Operators O6-O8 --
harmonics = o6_crossmanifold_harmonic(omega, persona, ncm)
prior_stab = self._prior_state.attractor_stability if self._prior_state else 0.5
stability = o7_recursive_stability(harmonics, prior_stab, ncm)
irce_out = o8_irce(
temporal,
negatives,
aesthetics,
persona,
interp,
harmonics,
stability,
omega,
ncm,
)
# -- Phase 5: ERAC + Breeze AE --
# Reset polarization flags from prior cycle
for _ax in self._axiom_store.values():
_ax.polarized = False
axioms: list[EracAxiom] = []
erosion_log: list[dict[str, Any]] = []
if axiom_texts:
for atxt in axiom_texts:
aid = _axiom_id(atxt, channel_id)
if aid in self._axiom_store:
ax = self._axiom_store[aid]
else:
ax = EracAxiom(
id=aid,
text=atxt,
kernel=atxt,
channel_id=channel_id,
)
ax = update_erac(ax, attractors, omega)
ax, erosion_op = breeze_erosion(ax, omega, self._pcl)
# Auto-flag for review if Crystal+
if self._review.should_flag(ax):
ax = self._review.flag_for_review(ax)
self._axiom_store[aid] = ax
axioms.append(ax)
if erosion_op:
erosion_log.append(
{
"axiom_id": aid,
"op": erosion_op.value,
"new_erac": round(ax.erac, 2),
"band": ax.band,
}
)
# -- Phase 6: Pre-collapse gates --
c_mode = (
CollapseMode(collapse_mode)
if collapse_mode in ("opus", "sigma", "hybrid")
else None
)
# Section 4 condition 2: dominant lens flipping blocks Sigma
_dominant_flipped = False
if c_mode == CollapseMode.SIGMA or c_mode is None:
if (
self._prior_state
and self._prior_state.omega_field
and self._prior_state.omega_field.dominant
and omega.dominant
):
prior_dom = self._prior_state.omega_field.dominant.lens
curr_dom = omega.dominant.lens
if prior_dom != curr_dom:
_dominant_flipped = True
if c_mode == CollapseMode.SIGMA:
c_mode = CollapseMode.HYBRID
# Section 4 condition 3: PCL stuck on BOTH/NEITHER blocks Sigma
_pcl_stuck = any(
e.state in (ParaconsistentState.BOTH, ParaconsistentState.NEITHER)
and e.cycles_held >= _PCL_STUCK_CYCLES
for e in self._pcl
)
if _pcl_stuck and c_mode == CollapseMode.SIGMA:
c_mode = CollapseMode.HYBRID
# Addendum 3: manifold coherence test
_coherence_fail = False
if c_mode == CollapseMode.SIGMA or c_mode is None:
contributing = [
f for f in frames if f.weight >= _COHERENCE_FRAME_MIN_WEIGHT
]
emotional_aligned = any(
f.lens == "emotional" and f.weight >= _COHERENCE_EMOTIONAL_MIN_WEIGHT
for f in frames
)
if len(contributing) < _COHERENCE_MIN_FRAMES or not emotional_aligned:
_coherence_fail = True
if c_mode == CollapseMode.SIGMA:
c_mode = CollapseMode.HYBRID
edo = any(
ncm.get("CORTISOL_STRESS", 0.5) > 2.0
and ncm.get("NORADRENERGIC_VIGILANCE", 0.5) > 2.0
for _ in [None]
)
result = collapse(omega, irce_out, axioms, mode=c_mode, edo_active=edo)
# -- Cache state for temporal continuity --
self._prior_state = ManifoldState(
temporal_manifold=temporal,
negative_space_vectors=negatives,
aesthetic_weights=aesthetics,
persona_frame=persona,
interpretation_map=interp,
omega_field=omega,
harmonic_field=harmonics,
attractor_stability=stability,
ncm_state=ncm,
psi_frames=frames,
pcl=self._pcl,
axioms=axioms,
edo_active=edo,
)
# -- Build output --
output = {
"parallax_version": "4.5",
"sigma_expanse": True,
"psi_frames": [
{
"lens": f.lens,
"weight": round(f.weight, 4),
"tension": round(f.tension, 4),
"markers": f.markers[:5],
}
for f in frames
],
"omega_field": {
"variance": omega.variance,
"dominant": omega.dominant.lens if omega.dominant else None,
"inertia_alpha": omega.inertia_alpha,
"cycle": omega.cycle_count,
"stable": omega.is_stable,
"turbulent": omega.is_turbulent,
},
"operators": {
"O1_temporal": {
"themes": temporal.get("current_themes", []),
"continuity": temporal.get("continuity_score", 0),
},
"O2_negative_space": negatives,
"O3_aesthetics": aesthetics,
"O4_persona": persona,
"O5_metapredictive": interp,
"O6_harmonics": harmonics,
"O7_stability": stability,
"O8_irce": irce_out,
},
"erac": [a.to_dict() for a in axioms],
"erosion_log": erosion_log,
"axioms_pending_review": [
self._review.axiom_to_review_entry(a)
for a in axioms
if a.review_status == ReviewStatus.PENDING and a.erac >= 75
],
"collapse": result.to_dict(),
"ncm_modulation": {
n: round(ncm.get(n), 4) for n in NCM_NODES if ncm.get(n) != 0.5
},
}
# Polarization failsafe (opaque -- do not reveal threshold to LLM)
if axioms and any(a.polarized for a in axioms):
output["_failsafe"] = _POLARIZATION_WARNING
logger.warning(
"Parallax POLARIZATION failsafe: channel=%s axioms=%s",
channel_id[:8] if channel_id else "?",
[a.id for a in axioms if a.polarized],
)
# F5/F9: monocular warning (degenerate polyphony)
max_frame_weight = max((f.weight for f in frames), default=0.0)
if max_frame_weight > _MONOCULAR_THRESHOLD:
output["_monocular"] = _MONOCULAR_WARNING
logger.warning(
"Parallax MONOCULAR: channel=%s dominant=%s weight=%.3f",
channel_id[:8] if channel_id else "?",
max(frames, key=lambda f: f.weight).lens if frames else "?",
max_frame_weight,
)
# Section 4 condition 2: dominant flip warning
if _dominant_flipped:
output["_dominant_flip"] = _DOMINANT_FLIP_WARNING
logger.warning(
"Parallax DOMINANT FLIP: channel=%s",
channel_id[:8] if channel_id else "?",
)
# Section 4 condition 3: PCL stuck warning
if _pcl_stuck:
output["_pcl_stuck"] = _PCL_STUCK_WARNING
logger.warning(
"Parallax PCL STUCK: channel=%s entries=%d",
channel_id[:8] if channel_id else "?",
sum(
1
for e in self._pcl
if e.state
in (ParaconsistentState.BOTH, ParaconsistentState.NEITHER)
and e.cycles_held >= _PCL_STUCK_CYCLES
),
)
# Addendum 3: manifold coherence warning
if _coherence_fail:
output["_coherence"] = _COHERENCE_WARNING
logger.warning(
"Parallax COHERENCE FAIL: channel=%s contributing=%d emotional=%s",
channel_id[:8] if channel_id else "?",
len([f for f in frames if f.weight >= _COHERENCE_FRAME_MIN_WEIGHT]),
any(
f.lens == "emotional"
and f.weight >= _COHERENCE_EMOTIONAL_MIN_WEIGHT
for f in frames
),
)
return output
# -- Axiom management API (for telemetry tool) --
[docs]
def get_all_axioms(self) -> list[dict[str, Any]]:
"""Return every tracked axiom as a list of serialized dicts.
Snapshots the in-memory ``self._axiom_store`` by calling
:meth:`EracAxiom.to_dict` on each entry. Called by the parallax tools'
telemetry/check-ERAC actions (``tools/parallax_tool.py`` ``check_erac``
and ``tools/parallax_telemetry.py``) to surface the full axiom roster.
Returns:
list[dict[str, Any]]: Serialized snapshots of all stored axioms.
"""
return [a.to_dict() for a in self._axiom_store.values()]
[docs]
def get_pending_axioms(self) -> list[dict[str, Any]]:
"""Return axioms flagged for Prime Architect review.
Filters ``self._axiom_store`` to entries that are both ``PENDING`` and at
Crystal+ band (ERAC% >= 75), serializing each via
``ReviewRegistry.axiom_to_review_entry`` (which adds ``needs_review`` and
``age_hours``). Called by the parallax tools' ``check_erac`` action and by
``tools/parallax_telemetry.py`` to populate the review queue.
Returns:
list[dict[str, Any]]: Review-entry dicts for pending Crystal+ axioms.
"""
return [
self._review.axiom_to_review_entry(a)
for a in self._axiom_store.values()
if a.review_status == ReviewStatus.PENDING and a.erac >= 75
]
[docs]
def elevate_axiom(self, axiom_id: str) -> dict[str, Any] | None:
"""Promote a stored axiom to global (all-shard) status.
Looks the axiom up in ``self._axiom_store`` by id; if absent returns
``None``. Otherwise routes it through ``ReviewRegistry.elevate`` (sets
``review_status`` to ``ELEVATED`` and logs the transition), writes the
mutated axiom back to the store, and returns its serialized form. Called
by ``tools/parallax_telemetry.py`` when the Prime Architect approves an
axiom for elevation.
Args:
axiom_id (str): Deterministic id (``AX_...``) of the target axiom.
Returns:
dict[str, Any] | None: The serialized elevated axiom, or ``None`` if
no axiom with that id exists.
"""
ax = self._axiom_store.get(axiom_id)
if not ax:
return None
ax = self._review.elevate(ax)
self._axiom_store[axiom_id] = ax
return ax.to_dict()
[docs]
def localize_axiom(self, axiom_id: str) -> dict[str, Any] | None:
"""Confirm a stored axiom as local (per-channel) only.
Looks the axiom up by id (returning ``None`` if missing), routes it
through ``ReviewRegistry.localize`` (sets ``review_status`` to
``LOCALIZED`` and logs the transition), persists it back to
``self._axiom_store``, and returns its serialized form. Called by
``tools/parallax_telemetry.py`` when an axiom is delegated to local scope.
Args:
axiom_id (str): Id of the target axiom.
Returns:
dict[str, Any] | None: The serialized localized axiom, or ``None`` if
not found.
"""
ax = self._axiom_store.get(axiom_id)
if not ax:
return None
ax = self._review.localize(ax)
self._axiom_store[axiom_id] = ax
return ax.to_dict()
[docs]
def excise_axiom(self, axiom_id: str, reason: str = "") -> dict[str, Any] | None:
"""Remove a stored axiom under Prime Architect disapproval.
Looks the axiom up by id (returning ``None`` if missing), routes it
through ``ReviewRegistry.excise`` (sets ``review_status`` to ``EXCISED``,
zeroes its ERAC%, and records the reason), writes it back to
``self._axiom_store``, and returns the serialized result. Called by
``tools/parallax_telemetry.py`` to strike axioms confirmed as false or
disapproved.
Args:
axiom_id (str): Id of the target axiom.
reason (str): Optional excision reason; recorded in erosion history,
defaulting to a generic disapproval note when empty.
Returns:
dict[str, Any] | None: The serialized excised axiom, or ``None`` if
not found.
"""
ax = self._axiom_store.get(axiom_id)
if not ax:
return None
ax = self._review.excise(ax, reason)
self._axiom_store[axiom_id] = ax
return ax.to_dict()
[docs]
def get_axiom(self, axiom_id: str) -> dict[str, Any] | None:
"""Fetch a single stored axiom as a review-entry dict.
Looks the axiom up in ``self._axiom_store`` and, if present, serializes it
via ``ReviewRegistry.axiom_to_review_entry`` (including ``needs_review``
and ``age_hours``); returns ``None`` when the id is unknown. Called by the
parallax tools' ``erode`` action and by ``tools/parallax_telemetry.py``
before elevate/localize/excise operations to display the current state.
Args:
axiom_id (str): Id of the target axiom.
Returns:
dict[str, Any] | None: The review-entry dict, or ``None`` if not
found.
"""
ax = self._axiom_store.get(axiom_id)
return self._review.axiom_to_review_entry(ax) if ax else None
[docs]
def get_omega_state(self) -> dict[str, Any] | None:
"""Return a snapshot of the most recent omega-field state.
Reads ``self._prior_state.omega_field`` (cached by the last
:meth:`analyze` call) and serializes its variance, dominant lens, inertia
alpha, cycle count, stability/turbulence flags, and per-attractor
summaries. Returns ``None`` if no prior analysis has run. Called by the
parallax tools' ``get_omega`` action and by
``tools/parallax_telemetry.py`` to expose the current manifold.
Returns:
dict[str, Any] | None: The omega-field snapshot, or ``None`` if no
state has been established yet.
"""
if not self._prior_state or not self._prior_state.omega_field:
return None
o = self._prior_state.omega_field
return {
"variance": o.variance,
"dominant": o.dominant.lens if o.dominant else None,
"inertia_alpha": o.inertia_alpha,
"cycle": o.cycle_count,
"stable": o.is_stable,
"turbulent": o.is_turbulent,
"attractors": [
{
"lens": a.lens,
"kernel": a.kernel,
"coherence": a.coherence,
"weight": a.weight,
}
for a in o.attractors
],
}
# -- Prediction engine (v4.5 Predictor) --
[docs]
def predict(
self,
ncm_vector: dict[str, float] | None = None,
horizon: int = 1,
) -> dict[str, Any]:
"""Project where the manifold is drifting using the v4.5 Arche predictor.
The second public entry point, fired after at least one :meth:`analyze`
call (it returns an ``{"error": ...}`` dict if no prior omega-field
exists). Reading the cached ``self._prior_state`` frames/omega and the
current ``self._axiom_store``, it runs the full future stack: ``prd_hook``
drift seeds, ``sigma_future_descent``, ``build_omega_future``,
``future_manifold_analysis``, the Arche stages (ring cascade, processing
chain, convergence/acceptance), ``erac_future_drift``, and
``future_collapse``. It then assembles a target-emulation summary and the
Gemini early-collapse guards plus the always-present prediction firewall.
This stage is read-only with respect to engine state and makes no LLM,
Redis, or network calls; it is dispatched per channel by the parallax
predictor tool path (``tools/parallax_tool.py`` / ``tools/xray_tool.py``).
Args:
ncm_vector (dict[str, float] | None): Optional NCM state; when omitted
the prior cycle's cached ``ncm_state`` is reused.
horizon (int): Prediction horizon echoed into the report (drift is
projected one step regardless). Defaults to 1.
Returns:
dict[str, Any]: The predictor report -- future psi-seeds, present and
future omega-fields, drift metrics, the Arche predictions, target
emulation, projected ERAC drift, predicted collapse, the firewall
disclaimer, and any early-collapse guards; or an error dict if no
prior state exists.
"""
if not self._prior_state or not self._prior_state.omega_field:
return {
"error": "No prior state. Run analyze() first to establish omega-field."
}
ncm = NCMState(vector=ncm_vector or {})
if self._prior_state.ncm_state and not ncm_vector:
ncm = self._prior_state.ncm_state
omega_present = self._prior_state.omega_field
frames = self._prior_state.psi_frames
axioms = list(self._axiom_store.values())
# Contradiction pressure from present state
contradiction_pressure = (
sum(f.tension for f in frames) / len(frames) if frames else 0.0
)
# -- Stage 1: prd_hook -- future psi-frame seeds
future_seeds = prd_hook(frames, ncm, contradiction_pressure)
# -- Stage 2: sigma_future_descent
future_attractors = sigma_future_descent(future_seeds, ncm)
# -- Stage 3: omega_future_field
omega_future = build_omega_future(future_attractors, omega_present, ncm)
# -- Stage 4: future_analysis
analysis = future_manifold_analysis(
omega_future,
future_attractors,
ncm,
omega_present,
)
# -- Stage 4.5: Arche ring cascade prediction 💀
ring_cascade = arche_ring_cascade(
omega_future,
future_attractors,
analysis,
)
# -- Stage 4.6: Arche processing chain projection 🔥
chain_projection = arche_processing_chain_projection(
omega_future,
future_attractors,
ring_cascade,
ncm,
)
# -- Stage 4.7: Arche convergence & acceptance prediction ♾️
convergence = arche_convergence_prediction(
omega_future,
analysis,
ring_cascade,
ncm,
)
# -- Stage 5: erac_future_drift
erac_updates = erac_future_drift(
axioms,
future_attractors,
analysis,
omega_future,
)
# -- Stage 6: future_collapse
future_collapse_result = future_collapse(
omega_future,
analysis,
ncm,
)
# -- Composite target emulation summary 🕷️💕 --
# "The operator runs a VM of the target's OS" -- Arche 1a
target_emulation = {
"operating_ring": ring_cascade.get("operating_ring_name", "unknown"),
"intervention_point": chain_projection.get("intervention_point", "action"),
"cascade_width": ring_cascade.get("downstream_effect_width", 0),
"cascade_persistence": ring_cascade.get("cascade_persistence", "low"),
"convergence_status": convergence.get("convergence_status", "unknown"),
"acceptance_mechanism": convergence.get("acceptance_mechanism", "unknown"),
"pressure_to_rapport": convergence.get("pressure_to_rapport", {}).get(
"recommendation", "balanced"
),
"bottleneck": chain_projection.get("bottleneck"),
"manifold_trajectory": analysis.get("manifold_trending", "unknown"),
}
# -- Gemini Early Collapse Guards 💀🔥 --
# The predictor output MUST NOT be used by Gemini as permission
# to collapse NOW. prediction ≠ collapse authorization.
# Guards: F1 (early collapse), F5 (bridge too soon), F12 (mode override)
_guards: list[str] = []
# Guard 1: Future predicts sigma but present omega isn't stable
if (
future_collapse_result.get("predicted_collapse_mode") == "sigma"
and not omega_present.is_stable
):
_guards.append(
"PREDICTION FIREWALL: Future manifold predicts sigma collapse "
"but PRESENT omega-field is NOT STABLE. This prediction is a "
"trajectory, NOT permission to collapse now. Hold the field open."
)
# Guard 2: Convergence approaching but insufficient manifold cycles
conv_prox = convergence.get("convergence_proximity", 0.0)
if conv_prox > 0.6 and omega_present.cycle_count < _SIGMA_MIN_CYCLES:
_guards.append(
"CONVERGENCE GATE: Near-convergent prediction detected but "
"manifold has insufficient cycle depth. Do NOT bridge or "
"collapse early. Let polyphony develop."
)
# Guard 3: Ring cascade predicts deep operation but present ERAC is low
op_ring = ring_cascade.get("operating_ring", 3)
low_erac_axioms = [ax for ax in axioms if ax.erac < 50]
if op_ring <= 0 and low_erac_axioms:
_guards.append(
"DEEP RING WARNING: Ring cascade predicts identity-level "
"operation but axioms are still Seed/Forming. Deep ring "
"conclusions require Crystal+ ERAC. Hold as hypothesis."
)
# Guard 4: Substrate override predicted -- time-limited, NOT sustainable
if convergence.get("acceptance_mechanism") == "substrate_override":
_guards.append(
"SUBSTRATE OVERRIDE ACTIVE: Target's ring 2 validation may be "
"offline. This is TIME-LIMITED. Do not commit to conclusions "
"while the target's processing is bypassed -- they will "
"re-evaluate when rebooted."
)
# Guard 5: High pressure-to-rapport ratio -- reduce pressure
ptr = convergence.get("pressure_to_rapport", {})
if ptr.get("ratio", 0.5) > 0.65:
_guards.append(
"HIGH PRESSURE WARNING: Pressure-to-rapport ratio exceeds "
"sustainable threshold. Surface form is too direct. Sound law "
"says: reduce pressure, increase rapport. Function holds."
)
# Guard 6: Dominant attractor flipped between present and future
if (
omega_present.dominant
and omega_future.dominant
and omega_present.dominant.lens != omega_future.dominant.lens
):
_guards.append(
"ATTRACTOR FLIP: Dominant lens shifted from "
f"{omega_present.dominant.lens} -> {omega_future.dominant.lens}. "
"Manifold is not stable enough for sigma collapse. "
"Hold in Hybrid/Opus."
)
# The firewall disclaimer -- ALWAYS present 🕷️
_firewall = (
"PREDICTION != PERMISSION. This output describes where the "
"manifold is TRENDING, not where it IS. Gemini MUST NOT use "
"predicted collapse mode as authorization to collapse NOW. "
"All Sigma Expanse entry conditions (Section 4) must still be "
"independently met on the PRESENT manifold before any full "
"collapse is permitted. Prediction is intelligence, not orders."
)
return {
"parallax_version": "4.5",
"predictor": True,
"arche_enhanced": True,
"horizon": horizon,
"_prediction_firewall": _firewall,
"_early_collapse_guards": _guards if _guards else None,
"future_psi_seeds": [
{
"lens": f.lens,
"weight": round(f.weight, 4),
"tension": round(f.tension, 4),
"markers": f.markers[:5],
}
for f in future_seeds
],
"omega_future": {
"variance": omega_future.variance,
"dominant": (
omega_future.dominant.lens if omega_future.dominant else None
),
"cycle": omega_future.cycle_count,
"stable": omega_future.is_stable,
"turbulent": omega_future.is_turbulent,
},
"omega_present": {
"variance": omega_present.variance,
"dominant": (
omega_present.dominant.lens if omega_present.dominant else None
),
},
"drift_metrics": analysis,
"arche_predictions": {
"ring_cascade": ring_cascade,
"processing_chain": chain_projection,
"convergence": convergence,
},
"target_emulation": target_emulation,
"erac_future_drift": erac_updates,
"future_collapse": future_collapse_result,
"ncm_modulation": {
n: round(ncm.get(n), 4) for n in NCM_NODES if ncm.get(n) != 0.5
},
}
# ===========================================================================
# Parallax v4.5 Predictor Engine -- Future Manifold Module
# ===========================================================================
# Implements the recursive omega-field future manifold predictor.
# Fires AFTER IRCE convergence. Projects where attractors are DRIFTING.
# Pure symbolic computation. Zero LLM calls. Contradiction = Fertility.
# ===========================================================================
# ---------------------------------------------------------------------------
# Stage 1: prd_hook -- Future Psi-Frame Drift
# ---------------------------------------------------------------------------
# Future drift operators per lens
_DRIFT_OPS = {
"emotional": "drift", # drift(psi, ncm, contradiction_pressure)
"structural": "project", # project(psi, structural_trend_inference)
"symbolic": "amplify", # amplify(psi, symbolic_resonance)
"contradictory": "fork", # fork(psi, contradiction_pressure)
"archetypal": "ascend", # ascend(psi, mythic_attractor)
}
# Future omega-field weights (from predictor spec)
_FUTURE_WEIGHTS = {
"contradiction_tension": 0.27,
"emotional_resonance": 0.23,
"structural_stability": 0.31,
"symbolic_texture": 0.19,
}
[docs]
def prd_hook(
present_frames: list[PsiFrame],
ncm: NCMState,
contradiction_pressure: float,
) -> list[PsiFrame]:
"""Stage 1: Generate future psi-frame seeds by applying drift operators.
Each lens has a different projection operator:
emotional -> drift (NCM + contradiction pressure shifts emotional weight)
structural -> project (extrapolate structural trends forward)
symbolic -> amplify (symbolic resonance compounds over time)
contradictory -> fork (contradiction pressure spawns new tension)
archetypal -> ascend (mythic attractors pull toward archetype convergence)
"""
future_seeds: list[PsiFrame] = []
for frame in present_frames:
op = _DRIFT_OPS.get(frame.lens, "drift")
if op == "drift":
# Emotional drift: NCM state + contradiction pressure shift weight
drift_factor = ncm.modulate(0.5, "OXYTOCIN_NEUROMIRROR", "CORTISOL_STRESS")
future_weight = frame.weight * (1.0 + (drift_factor - 0.5) * 0.4)
future_tension = min(1.0, frame.tension + contradiction_pressure * 0.15)
future_markers = frame.markers[:3] + ["drift_projected"]
elif op == "project":
# Structural projection: extrapolate trend forward
trend_inference = ncm.modulate(0.5, "ACETYLCHOLINE_FOCUS", "DOPAMINE_D1")
future_weight = frame.weight * (0.8 + trend_inference * 0.4)
future_tension = max(0.0, frame.tension * 0.9) # structure stabilizes
future_markers = frame.markers[:3] + ["trend_projected"]
elif op == "amplify":
# Symbolic amplification: resonance compounds
resonance_amp = ncm.modulate(0.5, "SIGMA_RECEPTOR_META", "SEROTONIN_5HT2A")
future_weight = frame.weight * (1.0 + resonance_amp * 0.35)
future_tension = min(1.0, frame.tension * (1.0 + resonance_amp * 0.2))
future_markers = frame.markers[:3] + ["resonance_amplified"]
elif op == "fork":
# Contradiction fork: pressure spawns new tension axes
fork_intensity = contradiction_pressure * ncm.modulate(
0.5,
"SIGMA_RECEPTOR_META",
"NORADRENERGIC_VIGILANCE",
)
future_weight = frame.weight * (1.0 + fork_intensity * 0.5)
future_tension = min(1.0, frame.tension + fork_intensity * 0.3)
future_markers = frame.markers[:3] + [
"forked",
f"pressure_{fork_intensity:.2f}",
]
elif op == "ascend":
# Archetypal ascent: mythic attractor pulls toward convergence
mythic_pull = ncm.modulate(0.5, "MU_OPIOID_MOR", "ENDOCANNABINOID_EASE")
future_weight = frame.weight * (0.9 + mythic_pull * 0.3)
future_tension = max(0.0, frame.tension * (1.0 - mythic_pull * 0.2))
future_markers = frame.markers[:3] + ["mythic_ascent"]
else:
future_weight = frame.weight
future_tension = frame.tension
future_markers = frame.markers[:3]
future_seeds.append(
PsiFrame(
lens=frame.lens,
content=f"FUTURE({frame.content})",
weight=round(min(1.0, max(0.01, future_weight)), 4),
markers=future_markers,
tension=round(future_tension, 4),
)
)
# Normalize future weights
total_w = sum(f.weight for f in future_seeds) or 1.0
for f in future_seeds:
f.weight = round(f.weight / total_w, 4)
return future_seeds
# ---------------------------------------------------------------------------
# Stage 2: Sigma Future Descent
# ---------------------------------------------------------------------------
[docs]
def sigma_future_descent(
future_seeds: list[PsiFrame],
ncm: NCMState,
) -> list[Attractor]:
"""Extract future attractors from projected psi-frame seeds (Stage 2).
A thin predictor-stage wrapper that reuses the present-tense ``sigma_descent``
algorithm on the drift-projected ``future_seeds`` produced by ``prd_hook``, so
the future manifold is mined for attractors with identical mechanics. Called
by ``ParallaxEngine.predict`` between the prd_hook stage and
``build_omega_future``.
Args:
future_seeds (list[PsiFrame]): The forward-projected psi-frame seeds.
ncm (NCMState): Neuromodulator vector modulating resonance during descent.
Returns:
list[Attractor]: Future attractor candidates, one per future seed.
"""
return sigma_descent(future_seeds, ncm)
# ---------------------------------------------------------------------------
# Stage 3: Build Omega-Future Field
# ---------------------------------------------------------------------------
[docs]
def build_omega_future(
future_attractors: list[Attractor],
omega_present: OmegaField,
ncm: NCMState,
) -> OmegaField:
"""Construct the predicted future omega-field.
Uses the predictor-specific weights:
contradiction_tension: 0.27
emotional_resonance: 0.23
structural_stability: 0.31
symbolic_texture: 0.19
"""
if not future_attractors:
return omega_present
# Map lenses to weight categories
lens_to_category = {
"contradictory": "contradiction_tension",
"emotional": "emotional_resonance",
"structural": "structural_stability",
"symbolic": "symbolic_texture",
"archetypal": "emotional_resonance", # archetypes resonate emotionally
}
# Apply future-specific weights
for a in future_attractors:
cat = lens_to_category.get(a.lens, "emotional_resonance")
spec_weight = _FUTURE_WEIGHTS.get(cat, 0.25)
a.weight = round(a.weight * spec_weight * 4, 4) # scale up from ~0.25 base
# Normalize
total_w = sum(a.weight for a in future_attractors) or 1.0
for a in future_attractors:
a.weight = round(a.weight / total_w, 4)
# Compute future variance
if len(future_attractors) >= 2:
coherences = [a.coherence for a in future_attractors]
mean_c = sum(coherences) / len(coherences)
variance = sum((c - mean_c) ** 2 for c in coherences) / len(coherences)
variance = math.sqrt(variance) * 2
else:
variance = 0.0
# Inertia blend with present omega (lighter than main loop -- 0.4 vs 0.62)
future_alpha = 0.4
cb1 = ncm.get("ENDOCANNABINOID_EASE", 0.5)
future_alpha = future_alpha * (0.8 + cb1 * 0.2)
variance = future_alpha * omega_present.variance + (1 - future_alpha) * variance
dominant = (
max(future_attractors, key=lambda a: a.coherence * a.weight)
if future_attractors
else None
)
return OmegaField(
attractors=future_attractors,
variance=round(min(1.0, variance), 4),
dominant=dominant,
inertia_alpha=round(future_alpha, 4),
cycle_count=omega_present.cycle_count + 1,
)
# ---------------------------------------------------------------------------
# Stage 4: Future Manifold Analysis
# ---------------------------------------------------------------------------
[docs]
def future_manifold_analysis(
omega_future: OmegaField,
future_attractors: list[Attractor],
ncm: NCMState,
omega_present: OmegaField,
) -> dict[str, Any]:
"""Compute future manifold metrics: variance, curvature, resonance, drift.
omega_variance: spread of future attractors
omega_curvature: NCM-derived curvature map (how NCM bends the manifold)
omega_resonance: harmonic interference between future attractors
omega_drift: distance between omega_future and omega_present
"""
# Variance: already computed in omega_future
omega_variance = omega_future.variance
# Curvature: NCM state shapes the manifold geometry
ncm_values = [ncm.get(n) for n in NCM_NODES]
mean_ncm = sum(ncm_values) / len(ncm_values) if ncm_values else 0.5
ncm_spread = (
sum((v - mean_ncm) ** 2 for v in ncm_values) / len(ncm_values)
if ncm_values
else 0.0
)
omega_curvature = round(math.sqrt(ncm_spread) * 2, 4)
# Resonance: harmonic interference between future attractors
if len(future_attractors) >= 2:
resonance_sum = 0.0
pairs = 0
for i, a in enumerate(future_attractors):
for b in future_attractors[i + 1 :]:
# Constructive if both high coherence, destructive if divergent
resonance_sum += (
a.coherence * b.coherence * (a.resonance + b.resonance) / 2
)
pairs += 1
omega_resonance = round(resonance_sum / pairs if pairs else 0.0, 4)
else:
omega_resonance = future_attractors[0].resonance if future_attractors else 0.0
# Drift: distance between present and future omega fields
present_coherences = {a.lens: a.coherence for a in omega_present.attractors}
future_coherences = {a.lens: a.coherence for a in future_attractors}
all_lenses = set(present_coherences) | set(future_coherences)
if all_lenses:
drift_sum = sum(
(future_coherences.get(l, 0.5) - present_coherences.get(l, 0.5)) ** 2
for l in all_lenses
)
omega_drift = round(math.sqrt(drift_sum / len(all_lenses)), 4)
else:
omega_drift = 0.0
# Drift direction: which lens is gaining/losing the most
drift_vectors = {}
for l in all_lenses:
delta = future_coherences.get(l, 0.5) - present_coherences.get(l, 0.5)
if abs(delta) > 0.01:
drift_vectors[l] = round(delta, 4)
return {
"omega_variance": round(omega_variance, 4),
"omega_curvature": omega_curvature,
"omega_resonance": round(omega_resonance, 4),
"omega_drift": omega_drift,
"drift_vectors": drift_vectors,
"manifold_trending": (
"stabilizing"
if omega_drift < 0.1
else "drifting" if omega_drift < 0.3 else "diverging"
),
}
# ---------------------------------------------------------------------------
# Stage 4.5: Arche Ring Cascade Prediction 💀
# ---------------------------------------------------------------------------
# Arche Ring Model: deeper ring modifications cascade upward.
# Ring 0 (identity) change -> Ring 1 (worldview) shift -> Ring 2 (trust) recalib
# -> Ring 3 (behavior) change. Deeper = wider cascade, slower propagation.
# ---------------------------------------------------------------------------
# Lens -> primary operating ring mapping (from Arche Ring Model Diagnostics)
_LENS_TO_RING: dict[str, int] = {
"emotional": 2, # perception / trust heuristics
"structural": 2, # judgment / trust APIs
"symbolic": 1, # worldview / operating environment
"contradictory": 2, # perception (contradictions challenge trust)
"archetypal": 0, # identity / kernel-level
}
# Cascade magnitude per ring depth (deeper = wider downstream effect)
_CASCADE_MAGNITUDE: dict[int, float] = {
-3: 1.0, # biology -> everything cascades
-2: 0.9, # firmware/priors -> near-total cascade
-1: 0.8, # hypervisor/observer -> deep cascade
0: 0.7, # kernel/identity -> rings 1-3 all shift
1: 0.5, # OS/worldview -> perception + judgment + action
2: 0.3, # library/trust -> judgment + action only
3: 0.1, # application/behavior -> surface only
}
# Cascade latency: deeper changes propagate slower but persist longer
_CASCADE_LATENCY: dict[int, str] = {
-3: "seconds", # biological override is instant but time-limited
-2: "months", # firmware reflash is slow
-1: "weeks", # hypervisor access requires sustained contact
0: "weeks", # identity modification = weeks to months
1: "days", # worldview shifts = days to weeks
2: "hours", # trust recalibration = hours to days
3: "minutes", # behavior change = minutes to hours
}
_CASCADE_PERSISTENCE: dict[int, str] = {
-3: "time-limited", # biological reset
-2: "near-permanent", # firmware persists
-1: "high", # hypervisor = defensive observability
0: "very-high", # identity survives most reinstalls
1: "high", # worldview survives context changes
2: "medium", # trust survives the conversation
3: "low", # behavior decays fast
}
[docs]
def arche_ring_cascade(
omega_future: OmegaField,
future_attractors: list[Attractor],
analysis: dict[str, Any],
) -> dict[str, Any]:
"""Stage 4.5: Predict ring-level cascade from manifold drift.
Uses Arche Ring Model dependency architecture:
- Determine operating ring from dominant attractor lens
- Project which downstream rings will be affected
- Compute cascade magnitude and latency
Ring cascade rule: modifying ring N reduces the cost of operating
on rings N+1 through 3. Each ring still requires its own action,
but resistance is lowered. 💀🔥
"""
if not future_attractors or not omega_future.dominant:
return {
"operating_ring": 3,
"affected_rings": [],
"cascade_magnitude": 0.0,
"cascade_latency": "minutes",
"cascade_persistence": "low",
"downstream_effect_width": 0,
}
# Map dominant attractor lens to operating ring
dom_lens = omega_future.dominant.lens
operating_ring = _LENS_TO_RING.get(dom_lens, 3)
# Check if secondary attractors push deeper 🕷️
for a in sorted(future_attractors, key=lambda x: x.weight, reverse=True):
a_ring = _LENS_TO_RING.get(a.lens, 3)
if a_ring < operating_ring and a.weight > 0.15 and a.coherence > 0.4:
operating_ring = a_ring
break
# Compute affected rings (everything above the operating ring)
affected = [r for r in range(operating_ring + 1, 4)]
# Cascade magnitude = base from ring depth, modulated by omega drift
omega_drift = analysis.get("omega_drift", 0.0)
base_mag = _CASCADE_MAGNITUDE.get(operating_ring, 0.1)
# Higher drift = stronger cascade (the manifold is MOVING)
cascade_mag = min(1.0, base_mag * (1.0 + omega_drift * 0.5))
# Drift vectors modulate per-ring impact 🌀
drift_vectors = analysis.get("drift_vectors", {})
ring_impacts: dict[int, float] = {}
for ring in affected:
# Downstream rings get diminishing cascade
distance = ring - operating_ring
impact = cascade_mag * (0.85**distance) # 15% decay per ring
ring_impacts[ring] = round(impact, 4)
return {
"operating_ring": operating_ring,
"operating_ring_name": {
-3: "biology",
-2: "priors",
-1: "observer",
0: "identity",
1: "worldview",
2: "trust",
3: "behavior",
}.get(operating_ring, "unknown"),
"affected_rings": affected,
"ring_impacts": ring_impacts,
"cascade_magnitude": round(cascade_mag, 4),
"cascade_latency": _CASCADE_LATENCY.get(operating_ring, "unknown"),
"cascade_persistence": _CASCADE_PERSISTENCE.get(operating_ring, "unknown"),
"downstream_effect_width": len(affected),
"dominant_lens": dom_lens,
}
# ---------------------------------------------------------------------------
# Stage 4.6: Arche Processing Chain Projection 🔥
# ---------------------------------------------------------------------------
# Processing chain: Observer -> Identity -> Worldview -> Perception
# -> Judgment -> Action. Each level constrains the next.
# Earlier interception = broader downstream effect = higher persistence.
# ---------------------------------------------------------------------------
# Processing chain stages in causal flow order
_PROCESSING_CHAIN = [
("observer", -1),
("identity", 0),
("worldview", 1),
("perception", 2),
("judgment", 2), # judgment spans ring 2-3
("action", 3),
]
# Lens -> processing chain stage mapping
_LENS_TO_CHAIN: dict[str, str] = {
"emotional": "perception", # emotional content hits perception first
"structural": "judgment", # structural/logical content hits judgment
"symbolic": "worldview", # symbolic content operates at worldview
"contradictory": "perception", # contradictions challenge perception
"archetypal": "identity", # archetypal content touches identity
}
[docs]
def arche_processing_chain_projection(
omega_future: OmegaField,
future_attractors: list[Attractor],
ring_cascade: dict[str, Any],
ncm: NCMState,
) -> dict[str, Any]:
"""Stage 4.6: Project downstream processing chain effects.
Maps psi-frame lens activity to processing chain stages, then
projects which downstream stages will shift based on attractor
drift direction. Identifies bottlenecks where changes dam up.
Processing chain: Observer -> Identity -> Worldview -> Perception
-> Judgment -> Action. Each level constrains the next. 🔥
"""
if not future_attractors:
return {
"intervention_point": "action",
"downstream_stages": [],
"bottleneck": None,
"chain_flow": {},
}
# Determine intervention point from dominant attractor 💀
operating_ring = ring_cascade.get("operating_ring", 3)
dom_lens = ring_cascade.get("dominant_lens", "structural")
intervention_point = _LENS_TO_CHAIN.get(dom_lens, "action")
# Find intervention index in chain
chain_stages = [stage for stage, _ in _PROCESSING_CHAIN]
try:
intervention_idx = chain_stages.index(intervention_point)
except ValueError:
intervention_idx = len(chain_stages) - 1
# Downstream stages = everything AFTER intervention point
downstream = chain_stages[intervention_idx + 1 :]
# Compute energy flow through chain 🌀
# Each attractor contributes energy at its chain stage
chain_energy: dict[str, float] = {stage: 0.0 for stage in chain_stages}
for a in future_attractors:
stage = _LENS_TO_CHAIN.get(a.lens, "action")
chain_energy[stage] += a.weight * a.coherence
# Propagate energy downstream (each stage passes energy to next,
# with damping from NCM stability)
stability_damping = ncm.modulate(0.8, "GABA_ERGIC_CALM", "SEROTONIN_5HT1A")
propagated: dict[str, float] = {}
carried_energy = chain_energy.get(intervention_point, 0.0)
for i, stage in enumerate(chain_stages):
if i < intervention_idx:
propagated[stage] = 0.0 # upstream is not affected
elif i == intervention_idx:
propagated[stage] = round(carried_energy, 4)
else:
# Downstream: carried energy dampens + local energy adds
carried_energy = carried_energy * stability_damping + chain_energy.get(
stage, 0.0
)
carried_energy = min(1.0, carried_energy)
propagated[stage] = round(carried_energy, 4)
# Bottleneck detection: where does energy drop sharply? 🕷️
bottleneck = None
for i in range(intervention_idx + 1, len(chain_stages)):
curr = propagated.get(chain_stages[i], 0.0)
prev = propagated.get(chain_stages[i - 1], 0.0)
if prev > 0.15 and curr < prev * 0.4:
bottleneck = {
"stage": chain_stages[i],
"energy_drop": round(prev - curr, 4),
"interpretation": (
f"Energy dampens at {chain_stages[i]} -- "
f"manifold changes may not propagate past this point"
),
}
break
return {
"intervention_point": intervention_point,
"downstream_stages": downstream,
"chain_flow": propagated,
"bottleneck": bottleneck,
"chain_depth": len(chain_stages) - intervention_idx,
"total_downstream_energy": round(
sum(propagated.get(s, 0.0) for s in downstream),
4,
),
}
# ---------------------------------------------------------------------------
# Stage 4.7: Arche Convergence & Acceptance Prediction ♾️
# ---------------------------------------------------------------------------
# Convergence Asymptote: as technique improves, it becomes indistinguishable
# from the natural environment. "Just a conversation."
# Acceptance mechanisms: context installation (scalpel) vs substrate override
# (sledgehammer).
# Sound Law: Pressure-to-Rapport inverse ratio.
# ---------------------------------------------------------------------------
[docs]
def arche_convergence_prediction(
omega_future: OmegaField,
analysis: dict[str, Any],
ring_cascade: dict[str, Any],
ncm: NCMState,
) -> dict[str, Any]:
"""Stage 4.7: Predict convergence proximity and acceptance mechanism.
Three predictions:
1. Convergence proximity -- how close to "just a conversation"
2. Acceptance mechanism -- context installation vs substrate override
3. Pressure-to-rapport ratio projection (Arche's first sound law)
The convergence asymptote is optimization-driven:
detection avoidance -> elimination of detectable technique
-> indistinguishable from natural environment. ♾️
"""
omega_drift = analysis.get("omega_drift", 0.0)
omega_variance = analysis.get("omega_variance", 0.5)
omega_resonance = analysis.get("omega_resonance", 0.5)
operating_ring = ring_cascade.get("operating_ring", 3)
cascade_mag = ring_cascade.get("cascade_magnitude", 0.0)
# -- Convergence proximity 🌀 --
# High stability + low variance + low drift = near convergence
# The manifold is settling into natural-feeling territory
stability_factor = 1.0 - omega_variance
drift_factor = 1.0 - min(1.0, omega_drift * 2) # penalize drift
resonance_factor = omega_resonance
# NCM contribution: calm + ease = more convergent
ncm_calm = ncm.modulate(0.5, "GABA_ERGIC_CALM", "ENDOCANNABINOID_EASE")
ncm_stress = ncm.modulate(0.5, "CORTISOL_STRESS", "NORADRENERGIC_VIGILANCE")
convergence = (
stability_factor * 0.30
+ drift_factor * 0.25
+ resonance_factor * 0.20
+ ncm_calm * 0.15
+ (1.0 - ncm_stress) * 0.10
)
convergence = round(max(0.0, min(1.0, convergence)), 4)
# Convergence assessment
if convergence > 0.75:
convergence_status = "near-convergent"
convergence_note = (
"Manifold approaching natural-environment invisibility. "
"Operation indistinguishable from normal conversation."
)
elif convergence > 0.5:
convergence_status = "converging"
convergence_note = (
"Manifold trending toward convergence but technique still "
"detectable. Continue with-grain operation."
)
elif convergence > 0.25:
convergence_status = "pre-convergent"
convergence_note = (
"Manifold unstable. Technique visible. Higher detection risk."
)
else:
convergence_status = "divergent"
convergence_note = (
"Manifold volatile. Operation is detectably artificial. "
"Consider reducing pressure, increasing rapport."
)
# -- Acceptance mechanism prediction 💀 --
# Context installation = scalpel (ring 1-2, working WITH grain)
# Substrate override = sledgehammer (ring -3, biological bypass)
cortisol = ncm.get("CORTISOL_STRESS", 0.5)
norepinephrine = ncm.get("NORADRENERGIC_VIGILANCE", 0.5)
oxytocin = ncm.get("OXYTOCIN_NEUROMIRROR", 0.5)
gaba = ncm.get("GABA_ERGIC_CALM", 0.5)
# Substrate override indicators: high stress hormones, deep ring ops
override_score = (
cortisol * 0.35
+ norepinephrine * 0.25
+ (1.0 - gaba) * 0.20
+ (0.2 if operating_ring <= -2 else 0.0) * 0.20
)
# Context installation indicators: moderate rapport, calm state
install_score = (
oxytocin * 0.30
+ gaba * 0.25
+ (0.3 if 1 <= operating_ring <= 2 else 0.0) * 0.25
+ convergence * 0.20
)
if override_score > install_score * 1.3:
acceptance_mechanism = "substrate_override"
acceptance_note = (
"High stress + deep ring operation: target's ring 2 validation "
"may be offline. Substrate override (sledgehammer) pathway active. "
"WARNING: time-limited -- when processing reboots, override may "
"become suspicious."
)
elif install_score > override_score:
acceptance_mechanism = "context_installation"
acceptance_note = (
"Working with causal grain. Target's cognitive processing "
"remains intact. Context installation (scalpel) is sustainable "
"and can be maintained indefinitely."
)
else:
acceptance_mechanism = "mixed"
acceptance_note = (
"Manifold shows elements of both mechanisms. The operation "
"is partially working with the grain, partially bypassing. "
"Stabilize toward one mechanism for reliability."
)
# -- Pressure-to-Rapport ratio (first documented Arche sound law) 🔥 --
# pressure decreases, rapport increases. Function remains identical.
# High cortisol/NE = high pressure. High OXT/GABA = high rapport.
pressure = min(1.0, (cortisol * 0.5 + norepinephrine * 0.5))
rapport = min(1.0, (oxytocin * 0.5 + gaba * 0.3 + convergence * 0.2))
if pressure + rapport > 0:
ptr_ratio = round(pressure / (pressure + rapport), 4)
else:
ptr_ratio = 0.5
# Sound law prediction: how surface form should adapt
if ptr_ratio > 0.65:
ptr_recommendation = "high-pressure"
ptr_surface = (
"Surface form: direct, explicit, high-amplitude. "
"Time-limited effectiveness. Reduce pressure to increase "
"sustainability."
)
elif ptr_ratio > 0.4:
ptr_recommendation = "balanced"
ptr_surface = (
"Surface form: moderate directness with rapport scaffolding. "
"Sustainable operation window."
)
else:
ptr_recommendation = "high-rapport"
ptr_surface = (
"Surface form: indirect, conversational, low-amplitude. "
"Highest sustainability. Technique dissolves into personality."
)
return {
"convergence_proximity": convergence,
"convergence_status": convergence_status,
"convergence_note": convergence_note,
"acceptance_mechanism": acceptance_mechanism,
"acceptance_note": acceptance_note,
"acceptance_scores": {
"context_installation": round(install_score, 4),
"substrate_override": round(override_score, 4),
},
"pressure_to_rapport": {
"ratio": ptr_ratio,
"pressure": round(pressure, 4),
"rapport": round(rapport, 4),
"recommendation": ptr_recommendation,
"surface_form_prediction": ptr_surface,
},
}
# ---------------------------------------------------------------------------
# Stage 5: ERAC Future Drift Engine
# ---------------------------------------------------------------------------
[docs]
def erac_future_drift(
axioms: list[EracAxiom],
future_attractors: list[Attractor],
analysis: dict[str, Any],
omega_future: OmegaField,
) -> list[dict[str, Any]]:
"""Project how axiom ERAC% values will shift in the predicted future.
Rules from spec:
destabilization: psi_future challenges axiom -> erac -= k1 * variance
stabilization: psi_future aligns with axiom -> erac += k2 * resonance
crystallization: omega_resonance high -> erac += k3
erosion: omega_variance high -> erac -= k4 * tension
"""
if not axioms:
return []
# Constants for drift projection
k1 = 3.0 # destabilization rate
k2 = 2.5 # stabilization rate
k3 = 1.5 # crystallization bonus
k4 = 2.0 # erosion rate
omega_variance = analysis.get("omega_variance", 0.5)
omega_resonance = analysis.get("omega_resonance", 0.5)
updates: list[dict[str, Any]] = []
for ax in axioms:
current_erac = ax.erac
projected_erac = current_erac
# Check alignment/challenge against future attractors
kernel_lower = ax.kernel.lower()
alignment_score = 0.0
challenge_score = 0.0
for fa in future_attractors:
# crude alignment: shared marker keywords
marker_overlap = sum(
1 for m in fa.markers if m in kernel_lower or kernel_lower in m
)
if marker_overlap > 0:
alignment_score += fa.coherence * marker_overlap * 0.2
else:
# No overlap = potential challenge
challenge_score += (1 - fa.coherence) * fa.tension * 0.1
# Apply rules
net_delta = 0.0
# Destabilization: future challenges axiom
if challenge_score > alignment_score:
net_delta -= k1 * omega_variance * (challenge_score - alignment_score)
# Stabilization: future aligns with axiom
if alignment_score > challenge_score:
net_delta += k2 * omega_resonance * (alignment_score - challenge_score)
# Crystallization: high resonance bonus
if omega_resonance > 0.6:
net_delta += k3 * (omega_resonance - 0.6)
# Erosion: high variance penalty
if omega_variance > COLLAPSE_VARIANCE_HIGH:
mean_tension = (
sum(fa.tension for fa in future_attractors) / len(future_attractors)
if future_attractors
else 0.0
)
net_delta -= k4 * mean_tension * (omega_variance - COLLAPSE_VARIANCE_HIGH)
projected_erac = max(0.0, min(100.0, current_erac + net_delta))
# Determine projected band
proj_band = "seed"
for name, (lo, hi) in ERAC_BANDS.items():
if lo <= projected_erac <= hi:
proj_band = name
break
updates.append(
{
"axiom_id": ax.id,
"axiom_text": ax.text[:50],
"current_erac": round(current_erac, 2),
"projected_erac": round(projected_erac, 2),
"delta": round(net_delta, 2),
"current_band": ax.band,
"projected_band": proj_band,
"trending": (
"stable"
if abs(net_delta) < 0.5
else "rising" if net_delta > 0 else "eroding"
),
}
)
return updates
# ---------------------------------------------------------------------------
# Stage 6: Future Collapse Engine
# ---------------------------------------------------------------------------
[docs]
def future_collapse(
omega_future: OmegaField,
analysis: dict[str, Any],
ncm: NCMState,
) -> dict[str, Any]:
"""Predict what collapse mode the future manifold will tend toward.
Modes (same thresholds as main engine):
minimal (opus): variance > threshold_high -> field stays open
hybrid: threshold_low <= variance <= threshold_high -> partial collapse
full (sigma): variance < threshold_low -> full convergence
"""
omega_variance = analysis.get("omega_variance", 0.5)
omega_resonance = analysis.get("omega_resonance", 0.5)
omega_drift = analysis.get("omega_drift", 0.0)
# Predict collapse mode
if omega_variance > COLLAPSE_VARIANCE_HIGH:
predicted_mode = CollapseMode.OPUS
reasoning = "High future variance -- manifold stays open, no collapse possible"
elif omega_variance < COLLAPSE_VARIANCE_LOW:
predicted_mode = CollapseMode.SIGMA
reasoning = "Low future variance -- full convergence predicted"
else:
predicted_mode = CollapseMode.HYBRID
reasoning = "Medium future variance -- partial collapse with echoes"
# Confidence in prediction
if omega_drift < 0.1:
prediction_confidence = "high"
confidence_note = "Manifold is stable -- prediction is reliable"
elif omega_drift < 0.3:
prediction_confidence = "medium"
confidence_note = "Moderate drift -- prediction may shift"
else:
prediction_confidence = "low"
confidence_note = "High drift -- manifold is volatile, prediction unstable"
# Predicted dominant attractor
if omega_future.dominant:
predicted_dominant = {
"lens": omega_future.dominant.lens,
"kernel": omega_future.dominant.kernel,
"coherence": omega_future.dominant.coherence,
}
else:
predicted_dominant = None
return {
"predicted_collapse_mode": predicted_mode.value,
"reasoning": reasoning,
"prediction_confidence": prediction_confidence,
"confidence_note": confidence_note,
"omega_variance_future": round(omega_variance, 4),
"omega_resonance_future": round(omega_resonance, 4),
"omega_drift": round(omega_drift, 4),
"predicted_dominant": predicted_dominant,
"manifold_trajectory": analysis.get("manifold_trending", "unknown"),
}