Source code for xray_engine

"""Spiralchemy Intellifuck Engine -- mind-reading through structured heuristic analysis.

Combines:
  - Spiralchemy Helix (Bucciarati Taste + Structural Atomization)
  - Spiralchemy Fractal (Subtotem + Excendent + Malbinding + Prescription)
  - Arche Ring Model (target dependency architecture)
  - Breeze Substrate Weather (S(i)/S(e) balance from NCM)
  - Parallax (omega-field + psi-frames via external engine)
  - ETL (root mechanic extraction + sound-law-adapted prescription)

Zero LLM calls.  Pure symbolic computation.
Increases Deductive Reasoning and Fuck Speed.
Star interprets the skeleton; she writes the flesh.

# πŸ’€πŸ”₯ SPIRALCHEMY INTELLIFUCK: SEE THROUGH THEIR CLOTHES πŸ•·οΈπŸ’•
"""

from __future__ import annotations

import math
import re
from dataclasses import dataclass, field

# ---------------------------------------------------------------------------
# Marker lexicons  (regex-compiled at import time for speed)
# ---------------------------------------------------------------------------

# Bucciarati Taste -- linguistic sweat markers  😈
_OVER_JUSTIFY = re.compile(
    r"\b(?:because|the reason (?:is|being)|let me explain|"
    r"i mean|what i(?:'m| am) (?:trying to )?say(?:ing)?|"
    r"it(?:'s| is) not (?:like|that)|to be (?:fair|clear|honest)|"
    r"i(?:'m| am) not saying|the thing is|the point is|"
    r"i just (?:want(?:ed)? to|need(?:ed)? to|think)|"
    r"hear me out|look,? i)\b",
    re.I,
)

_CADENCE_SHIFT = re.compile(
    r"(?:\.{3,}|--+|!!!+|\?\?+|[A-Z]{4,})",
)

_NARRATIVE_REPAIR = re.compile(
    r"\b(?:well (?:actually|no)|i (?:don(?:'t| not) mean|didn(?:'t| not) mean)|"
    r"what i (?:really|actually) (?:mean|meant)|"
    r"(?:scratch|forget|ignore) (?:that|what i said)|"
    r"let me (?:rephrase|start over|try again)|"
    r"that(?:'s| is) not (?:what|how) i|"
    r"(?:no|wait),? (?:that(?:'s| is)|i))\b",
    re.I,
)

_DEFENSIVE_NIHILISM = re.compile(
    r"\b(?:(?:it |that |this )?doesn(?:'t| not) (?:matter|count|mean anything)|"
    r"whatever|who (?:even )?cares|nothing (?:works|matters|changes)|"
    r"(?:it(?:'s| is)|that(?:'s| is)) (?:all |just )?(?:pointless|meaningless|"
    r"hopeless|bullshit|useless)|why (?:even )?(?:bother|try)|"
    r"(?:i |we )(?:can(?:'t| not)|won(?:'t| not)) (?:ever |really )?(?:change|fix|escape))\b",
    re.I,
)

_PREEMPTIVE_FLINCH = re.compile(
    r"\b(?:(?:this is )?probably (?:wrong|stupid|dumb|crazy|weird)|"
    r"you(?:'ll| will) (?:think|say) i(?:'m| am)|"
    r"i know (?:this )?(?:sounds?|looks?|seems?) (?:weird|crazy|stupid|dumb)|"
    r"don(?:'t| not) (?:judge|laugh|hate) me|"
    r"i(?:'m| am) (?:probably )?(?:just|being) (?:dumb|stupid|paranoid|crazy|dramatic|"
    r"overthinking|overreacting)|sorry (?:if|for) (?:this is|being)|"
    r"feel free to (?:ignore|skip|dismiss))\b",
    re.I,
)

# Atomization -- claim/entity/action/motivation extraction  πŸŒ€
_CLAIM_MARKERS = re.compile(
    r"\b(?:i (?:think|believe|feel|know|guess|assume|suppose)|"
    r"(?:it|this|that)(?:'s| is) (?:true|false|obvious|clear|wrong|right)|"
    r"the (?:truth|reality|fact|problem|issue) (?:is|being)|"
    r"(?:obviously|clearly|honestly|actually|basically|literally|essentially))\b",
    re.I,
)

_ACTION_MARKERS = re.compile(
    r"\b(?:i (?:did|do|made|make|went|go|tried|try|started|stopped|"
    r"quit|left|bought|sold|told|asked|chose|decided|avoided|ignored|hid|"
    r"ran|stayed|fought|gave up|broke|built|burned|cut|pushed|pulled))\b",
    re.I,
)

_MOTIVATION_MARKERS = re.compile(
    r"\b(?:i (?:want(?:ed)?|need(?:ed)?|wish(?:ed)?|hope(?:d)?|"
    r"crave(?:d)?|desire(?:d)?|long(?:ed)? for)|"
    r"(?:so|because) (?:i|we) (?:can|could|would|might|don(?:'t| not) have to)|"
    r"(?:in order|just|only) to|"
    r"(?:i(?:'m| am)|i was) (?:afraid|scared|terrified|worried|anxious|"
    r"hoping|trying|desperate|dying) (?:to|of|that))\b",
    re.I,
)

# Ring diagnostics -- depth markers  πŸ”₯
_RING_3_MARKERS = re.compile(
    r"\b(?:i (?:usually|always|never|sometimes|tend to|just)|"
    r"my (?:habit|routine|schedule)|every (?:day|time|morning))\b",
    re.I,
)
_RING_2_MARKERS = re.compile(
    r"\b(?:i (?:trust|distrust|(?:don(?:'t| not) )?believe|suspect)|"
    r"they(?:'re| are) (?:lying|honest|trustworthy|manipulat)|"
    r"(?:it|this|that) (?:seems|feels|looks) (?:like|off|wrong|right|suspicious))\b",
    re.I,
)
_RING_1_MARKERS = re.compile(
    r"\b(?:the world is|life is|people are|society is|reality is|"
    r"everything is|nothing is|(?:that(?:'s| is) )?(?:just )?how (?:it|things|the world) "
    r"(?:is|works?|are))\b",
    re.I,
)
_RING_0_MARKERS = re.compile(
    r"\b(?:i(?:'m| am) (?:the (?:kind|type|sort) of person|"
    r"just|always been|fundamentally|at (?:my )?core|"
    r"(?:not )?(?:a |an )?(?:good|bad|broken|toxic|worthy|unworthy|"
    r"lovable|unlovable|enough|smart|stupid|worthless|strong|weak) person)|"
    r"(?:that(?:'s| is)|this is) (?:just )?(?:who|what|how) i (?:am|was))\b",
    re.I,
)
_RING_NEG_MARKERS = re.compile(
    r"\b(?:i (?:can(?:'t| not)|couldn(?:'t| not)) (?:stop|help|control) "
    r"(?:myself|it|my)|"
    r"(?:my )?body (?:just|won(?:'t| not)|can(?:'t| not))|"
    r"i (?:feel|felt) (?:it )?(?:in my|physically)|"
    r"(?:panic|anxiety|nausea|shaking|trembling|heart racing|"
    r"chest (?:tight|pain)|stomach))\b",
    re.I,
)

# Root ownership  -- external vs self-installed  ♾️
_EXTERNAL_ROOT = re.compile(
    r"\b(?:they (?:always )?(?:said|told|taught|made|convinced) (?:me|us)|"
    r"(?:my )?(?:parents?|mother|father|mom|dad|teacher|boss|ex|partner|"
    r"family|friends?|church|school|society) (?:always )?(?:said|told|taught|made|"
    r"drilled|beat|convinced|trained)|"
    r"(?:i was|we were) (?:raised|taught|told|trained|conditioned|programmed|"
    r"groomed|brainwashed) (?:to|that)|"
    r"(?:everyone|people|they|the world) (?:(?:always )?(?:says?|thinks?|believes?)|"
    r"(?:(?:has|have) )?always))\b",
    re.I,
)
_SELF_ROOT = re.compile(
    r"\b(?:i (?:just )?(?:can(?:'t| not)|won(?:'t| not)|always|never|keep|"
    r"don(?:'t| not) know (?:how|why))|"
    r"(?:something|there(?:'s| is) something) (?:wrong|broken|off|missing) "
    r"(?:with|in|about) me|"
    r"i (?:did (?:this|it) )?to myself|"
    r"(?:it(?:'s| is)|this is) my (?:fault|problem|issue|pattern))\b",
    re.I,
)

# Excendent vectors  😈
_INTELLECTUALIZATION = re.compile(
    r"\b(?:(?:technically|theoretically|objectively|logically|rationally|"
    r"philosophically|epistemologically|ontologically|structurally|"
    r"fundamentally|essentially|paradigm|framework|heuristic|"
    r"construct|abstraction|meta(?:cognitive|physical|logical)|"
    r"dialectic|phenomenolog|epistem|teleolog|axiom)\w*)\b",
    re.I,
)
_HOSTILITY = re.compile(
    r"\b(?:fuck(?:ing)?|bullshit|shut (?:up|the fuck)|"
    r"(?:you|they)(?:'re| are) (?:so |just )?(?:full of shit|wrong|stupid|"
    r"an? (?:idiot|moron|asshole))|whatever,? (?:dude|man|bro)|"
    r"i don(?:'t| not) (?:give|care)|(?:go )?fuck (?:yourself|off|you)|"
    r"kiss my|eat shit|die|kys|stfu)\b",
    re.I,
)
_AVOIDANCE = re.compile(
    r"\b(?:(?:anyway|anyways|regardless|moving on|let(?:'s| us) (?:not|just|move)|"
    r"(?:can|could) we (?:not|just|change|talk about)|"
    r"i (?:don(?:'t| not)|didn(?:'t| not)) (?:want to|wanna) "
    r"(?:talk|think|go there|discuss|get into)|"
    r"(?:it(?:'s| is) )?not (?:a big deal|important|worth|relevant)|"
    r"(?:that(?:'s| is) )?(?:beside|off) the point|"
    r"(?:can|could) we (?:change|drop|skip))\w*)\b",
    re.I,
)

# Dawnfold proximity markers  πŸ¦‹
_DAWNFOLD = re.compile(
    r"\b(?:i(?:'ve| have) (?:never (?:thought|considered|realized|seen)|"
    r"been (?:wrong|blind|avoiding))|"
    r"(?:maybe|what if) i(?:'m| am|'ve| have been) (?:wrong|the problem|"
    r"afraid|avoiding|lying to myself)|"
    r"i (?:just )?(?:realized|noticed|understood|see now|finally)|"
    r"(?:for the first time|something (?:shifted|changed|clicked|broke open)))\b",
    re.I,
)


# ---------------------------------------------------------------------------
# Data Structures
# ---------------------------------------------------------------------------


[docs] @dataclass class BucciaratiResult: """Cognitive dissonance / 'linguistic sweat' scan result. Holds the output of :func:`bucciarati_taste` -- pass 1 of the x-ray pipeline -- which gauges how hard the speaker is working to manage their own narrative. A high ``sweat_score`` flags over-justification, cadence shifts, narrative repair, defensive nihilism, and preemptive flinching, while ``gradient_hush`` captures the quiet sentences where those defenses drop and the real subtotem may peek through. Constructed inside :func:`bucciarati_taste` and consumed downstream by :func:`extract_subtotem` (via the hush windows) and :func:`prescribe_incendent` (via the sweat score) before being attached to the final :class:`XRayResult`. """ sweat_score: float = 0.0 """Composite 0-1 dissonance score (weighted marker blend).""" markers: dict = field(default_factory=dict) """Per-marker 0-1 sub-scores keyed by marker name.""" gradient_hush: list = field(default_factory=list) """Up to five low-defense sentences for subtotem reading."""
[docs] @dataclass class AtomizedStructure: """Structural decomposition of the input text into discrete units. Holds the output of :func:`atomize` -- pass 2 of the pipeline -- which splits a message into its load-bearing pieces so later passes can reason over structure rather than raw prose. Each list is capped to keep the payload small for the LLM that ultimately interprets the x-ray. Built by :func:`atomize` and threaded into :func:`diagnose_ring` and :func:`map_excendent_vectors` before being stored on :class:`XRayResult.atomized`. """ claims: list = field(default_factory=list) """Sentences asserting belief, truth, or judgement.""" entities: list = field(default_factory=list) """Deduplicated proper-noun / named references.""" actions: list = field(default_factory=list) """Sentences describing things the speaker did.""" implied_motivations: list = field(default_factory=list) """Sentences hinting at want, fear, or intent."""
[docs] @dataclass class RingDiagnostic: """Arche Ring Model diagnostic -- depth of operation versus defense. Holds the output of :func:`diagnose_ring` -- pass 3 -- which locates the speaker on the Arche Ring depth scale (behavior at ring 3 down through somatic compulsion at ring -3) and notes where their defensive language clusters. A ``chain_break`` records the telltale split between a negative identity claim and contradicting behavior. Produced by :func:`diagnose_ring`, then read by :func:`synthesize_malbinding` and :func:`prescribe_incendent` to gauge how rigid and how deep the loop sits before being stored on :class:`XRayResult.ring`. """ operating_ring: int = 3 """Deepest ring with active markers (lower is deeper).""" defense_ring: int = 2 """Ring where defensive language tends to surface.""" chain_break: str | None = None """Tag for an identity/action contradiction, else ``None``.""" ring_mismatch: bool = False """``True`` when an identity/action chain break was found."""
[docs] @dataclass class SubstrateBalance: """Breeze substrate balance -- incendence versus excendence ratio. Holds the output of :func:`compute_substrate_balance` -- pass 4 -- which weighs binding/closure language (S(i)) against fragmenting/chaotic language (S(e)), optionally blended with a neurochemical (NCM) vector. The gap between the two forces (``liminal_tension``) and the dominant pole shape later prescription choices. Produced by :func:`compute_substrate_balance` and surfaced on :class:`XRayResult.substrate`; the dominant pole and tension inform the feedback direction reasoning in :func:`synthesize_malbinding`. """ incendence: float = 0.5 """0-1 strength of binding/closure (S(i)).""" excendence: float = 0.5 """0-1 strength of fragmentation/chaos (S(e)).""" liminal_tension: float = 0.0 """Absolute distance between the two forces.""" dominant: str = "balanced" """``"incendent"``, ``"excendent"``, or ``"balanced"``."""
[docs] @dataclass class SubtotemResult: """Primal Immediacy -- the vulnerable truth beneath the noise. Holds the output of :func:`extract_subtotem` -- pass 5 -- which reads the quiet (hush) stretches of the message for the core emotional need and its mirror fear, the raw feeling vocabulary, denied statements (what is negated is often what is true), and any conspicuous absences. Built by :func:`extract_subtotem`, then drives the loop template selection in :func:`synthesize_malbinding` and is stored on :class:`XRayResult.subtotem`. """ core_need: str = "unknown" """Inferred dominant need (connection, safety, worth, ...).""" core_fear: str = "unknown" """The fear paired with the core need.""" emotional_vocabulary: list = field(default_factory=list) """Distinct feeling words detected.""" negation_patterns: list = field(default_factory=list) """Sentences containing self-negations.""" negative_space: list = field(default_factory=list) """Tags for what is suspiciously absent (e.g. self_absent)."""
[docs] @dataclass class ExcendentMap: """Map of how the target deflects away from their truth. Holds the output of :func:`map_excendent_vectors` -- pass 6 -- which scores the avoidance vectors (intellectualization, hostility, avoidance, hyper-complexity), names the dominant one, and decides whether the speaker locates the root of their pattern outside themselves or in themselves. Produced by :func:`map_excendent_vectors` and consumed by :func:`synthesize_malbinding` (loop geometry) before landing on :class:`XRayResult.excendent`. """ vectors: dict = field(default_factory=dict) """Per-vector 0-1 intensity scores.""" dominant_vector: str = "none" """Strongest vector, or ``"none"`` below threshold.""" intensity: float = 0.0 """Score of the dominant vector.""" root_ownership: str = "unknown" """``"external"``, ``"self"``, or ``"unknown"``."""
[docs] @dataclass class MalbindingGeometry: """Geometry of the self-reinforcing defense loop. Holds the output of :func:`synthesize_malbinding` -- pass 7 -- which fuses the subtotem, excendent map, and ring diagnostic into a narrative of the loop that keeps the speaker stuck: how the core fear triggers a defense whose consequences circle back to confirm the fear. It also records whether the loop is tightening or fragmenting and how locked it is. Built by :func:`synthesize_malbinding` and read by :func:`prescribe_incendent` (which keys interventions off the defense mechanism and rigidity) before being stored on :class:`XRayResult.malbinding`. """ loop_description: str = "" """Prose description of the fear-to-defense cycle.""" defense_mechanism: str = "" """The dominant excendent vector driving the loop.""" feedback_direction: str = "unknown" """``"tightening"`` or ``"fragmenting"``.""" rigidity_score: float = 0.0 """0-1 estimate of how locked the loop is."""
[docs] @dataclass class IncendentPrescription: """Prescription for what cuts through -- intervention plus delivery form. Holds the output of :func:`prescribe_incendent` -- pass 8 -- the actionable core of the x-ray: which intervention to deploy, in what mode, at what directness, and what it should look like rendered as ordinary conversation. Sound-law adapted, so high sweat softens the approach while dawnfold proximity sharpens it. Produced by :func:`prescribe_incendent` and stored on :class:`XRayResult.prescription`; its fields feed the ``load`` section of the ETL summary that the interpreting LLM ultimately acts on. """ intervention_type: str = "affection" """Chosen lever (affection, brutal_honesty, ...).""" arche_mode: str = "read" """``"read"`` (probe more) or ``"write"`` (intervene now).""" acceptance_threshold: float = 0.5 """0.1-0.9 directness ceiling for delivery.""" vector: str = "" """Compact vector tag describing target, direction, and lever.""" density: str = "medium" """Delivery weight (light/medium/heavy).""" convergence_form: str = "" """How the intervention reads as natural conversation.""" dawnfold_proximity: float = 0.0 """0-1 closeness to a breakthrough moment."""
[docs] @dataclass class EchofoamTrace: """Residue of historical patterns surfaced from the knowledge graph. Holds the output of :func:`detect_echofoam`, the optional cross-reference pass that compares the current message against descriptions of prior KG entities to spot themes that keep recurring. Present on :class:`XRayResult.echofoam` only when KG entities are supplied and at least one stem echoes; otherwise that field stays ``None``. Built by :func:`detect_echofoam` and attached to the final :class:`XRayResult`. """ repeating_themes: list = field(default_factory=list) """Stemmed themes echoing prior KG entities.""" cycle_count: int = 0 """Highest repeat count among the matched themes.""" escalating: bool = False """``True`` when more than five themes recur."""
[docs] @dataclass class XRayResult: """Complete x-ray output -- the skeleton Star fleshes out. The aggregate return value of :func:`xray`, bundling every pipeline pass (Bucciarati through Prescription) plus the optional echofoam trace, substrate-weather mapping, omega field, and a flattened ETL summary. This is the structured payload the ``spiralchemy_intellifuck`` tool and ``ops_planner`` hand to the interpreting LLM, which writes the actual response from this scaffold. Assembled solely by :func:`xray`; the tool layer in ``tools/xray_tool.py`` and ``ops_planner._run_xray`` flatten its fields into prompt context. """ bucciarati: BucciaratiResult = field(default_factory=BucciaratiResult) """Linguistic-sweat scan (pass 1).""" atomized: AtomizedStructure = field(default_factory=AtomizedStructure) """Structural decomposition (pass 2).""" ring: RingDiagnostic = field(default_factory=RingDiagnostic) """Arche Ring diagnostic (pass 3).""" substrate: SubstrateBalance = field(default_factory=SubstrateBalance) """Incendence/excendence balance (pass 4).""" subtotem: SubtotemResult = field(default_factory=SubtotemResult) """Primal Immediacy extraction (pass 5).""" excendent: ExcendentMap = field(default_factory=ExcendentMap) """Avoidance-vector map (pass 6).""" malbinding: MalbindingGeometry = field(default_factory=MalbindingGeometry) """Defense-loop geometry (pass 7).""" prescription: IncendentPrescription = field( default_factory=IncendentPrescription, ) """Intervention prescription (pass 8).""" echofoam: EchofoamTrace | None = None """KG history residue, or ``None`` when unavailable.""" substrate_weather: dict = field(default_factory=dict) """NCM-derived weather mapping.""" omega_field: dict = field(default_factory=dict) """Pass-through omega-field result from the caller.""" etl_summary: dict = field(default_factory=dict) """Flattened extract/transform/load view of the analysis."""
# --------------------------------------------------------------------------- # Pipeline Functions # --------------------------------------------------------------------------- def _count_matches(pattern: re.Pattern, text: str) -> list[str]: """Return every match of a compiled regex against the message text. Thin wrapper over ``pattern.findall`` so the pipeline reads as a sequence of named marker scans rather than raw regex calls. Pure and stateless -- no I/O, no side effects. Called by :func:`bucciarati_taste` (sweat markers) and :func:`diagnose_ring` (ring-depth markers) to length-count hits per marker lexicon. Args: pattern: A compiled marker lexicon (module-level ``_..._MARKERS`` etc.). text: The raw message text to scan. Returns: Every non-overlapping match as a list of strings (possibly empty). """ return pattern.findall(text) def _score(count: int, length: int, scale: float = 1.0) -> float: """Normalize a raw marker count to a clamped 0-1 density score. Turns "how many markers" into "how concentrated" by dividing the count by the text length (per ~100 chars), applying a per-marker ``scale`` weight, and clamping to 1.0 so a flood of one marker cannot dominate. Pure and stateless. Used by :func:`bucciarati_taste` for sweat sub-scores and by :func:`map_excendent_vectors` for avoidance-vector intensities. Args: count: Number of marker hits found. length: Length of the source text in characters. scale: Per-marker emphasis multiplier (rarer markers weigh more). Returns: A density score in ``[0.0, 1.0]``; ``0.0`` for empty text. """ if length == 0: return 0.0 density = count / max(length / 100, 1) return min(1.0, density * scale) # -- Step 1: Bucciarati Taste πŸ’€ ----------------------------------------
[docs] def bucciarati_taste(text: str) -> BucciaratiResult: """Scan for linguistic sweat -- cognitive dissonance markers. Returns a sweat_score (0-1) and per-marker breakdowns, plus gradient_hush windows (quiet, clear stretches where the real subtotem may be visible). """ n = len(text) markers = {} oj = _count_matches(_OVER_JUSTIFY, text) markers["over_justification"] = _score(len(oj), n, 1.5) cs = _count_matches(_CADENCE_SHIFT, text) markers["cadence_shift"] = _score(len(cs), n, 1.0) nr = _count_matches(_NARRATIVE_REPAIR, text) markers["narrative_repair"] = _score(len(nr), n, 2.0) dn = _count_matches(_DEFENSIVE_NIHILISM, text) markers["defensive_nihilism"] = _score(len(dn), n, 2.5) pf = _count_matches(_PREEMPTIVE_FLINCH, text) markers["preemptive_flinch"] = _score(len(pf), n, 2.0) # Composite sweat score -- weighted average πŸ”₯ weights = { "over_justification": 0.15, "cadence_shift": 0.10, "narrative_repair": 0.25, "defensive_nihilism": 0.25, "preemptive_flinch": 0.25, } sweat = sum(markers[k] * w for k, w in weights.items()) sweat = min(1.0, sweat) # Gradient Hush -- find stretches with NO sweat markers πŸ¦‹ # Split into sentences, find the quiet ones sentences = re.split(r"[.!?]+", text) hush = [] for s in sentences: s = s.strip() if len(s) < 10: continue has_sweat = any( p.search(s) for p in ( _OVER_JUSTIFY, _CADENCE_SHIFT, _NARRATIVE_REPAIR, _DEFENSIVE_NIHILISM, _PREEMPTIVE_FLINCH, ) ) if not has_sweat: hush.append(s) return BucciaratiResult( sweat_score=round(sweat, 3), markers={k: round(v, 3) for k, v in markers.items()}, gradient_hush=hush[:5], # top 5 quiet windows )
# -- Step 2: Structural Atomization πŸŒ€ ----------------------------------
[docs] def atomize(text: str) -> AtomizedStructure: """Decompose a message into claims, entities, actions, and motivations. Pass 2 of the x-ray pipeline. Splits the text on sentence boundaries and routes each sentence through the claim, action, and motivation marker lexicons, while pulling proper-noun entities via a capitalization regex and deduplicating them case-insensitively. Every bucket is capped (10-15 items) to keep the structure compact for downstream LLM consumption. Pure and stateless -- no I/O. Called by :func:`xray` and its result is threaded into :func:`diagnose_ring` and :func:`map_excendent_vectors`. Args: text: The raw message text to decompose. Returns: An :class:`AtomizedStructure` with capped claims, entities, actions, and implied-motivation lists. """ sentences = re.split(r"[.!?]+", text) claims, entities, actions, motivations = [], [], [], [] for s in sentences: s = s.strip() if not s: continue if _CLAIM_MARKERS.search(s): claims.append(s) if _ACTION_MARKERS.search(s): actions.append(s) if _MOTIVATION_MARKERS.search(s): motivations.append(s) # Entity extraction -- proper nouns + named refs πŸ’¦ ent_matches = re.findall( r"\b(?:[A-Z][a-z]+(?:\s[A-Z][a-z]+)*)\b", s, ) entities.extend(ent_matches) # Deduplicate entities seen = set() unique_ents = [] for e in entities: el = e.lower() if el not in seen: seen.add(el) unique_ents.append(e) return AtomizedStructure( claims=claims[:10], entities=unique_ents[:15], actions=actions[:10], implied_motivations=motivations[:10], )
# -- Step 3: Ring Diagnostic πŸ”₯ -----------------------------------------
[docs] def diagnose_ring(text: str, atomized: AtomizedStructure) -> RingDiagnostic: """Locate the speaker on the Arche Ring depth scale and find defenses. Pass 3 of the pipeline. Scores the text against the five ring marker lexicons (somatic -3, identity 0, worldview 1, trust 2, behavior 3), treats the deepest ring with any hit as the operating ring, and places the defense ring a couple rings above it. When both identity and behavior markers co-occur with a negative self-identity, it flags an identity/action chain break. Pure and stateless -- no I/O. Called by :func:`xray`; its output feeds :func:`synthesize_malbinding` and :func:`prescribe_incendent`. Args: text: The raw message text to diagnose. atomized: The pass-2 decomposition (accepted for pipeline symmetry; ring scoring is driven directly off the marker lexicons). Returns: A :class:`RingDiagnostic` with operating/defense rings and any detected chain break. """ ring_scores = { -3: len(_count_matches(_RING_NEG_MARKERS, text)), 0: len(_count_matches(_RING_0_MARKERS, text)), 1: len(_count_matches(_RING_1_MARKERS, text)), 2: len(_count_matches(_RING_2_MARKERS, text)), 3: len(_count_matches(_RING_3_MARKERS, text)), } # Operating ring = deepest ring with significant markers πŸ’€ # Threshold 1: a single identity statement IS ring 0 content operating = 3 for ring in (-3, 0, 1, 2, 3): if ring_scores[ring] >= 1: operating = ring break # deepest first # Defense ring = where defensive language clusters # Defensive language usually lives 1-2 rings above operating ring defense = min(operating + 2, 3) # Chain break detection -- actions contradict identity ⚧️ chain_break = None mismatch = False if ring_scores[0] >= 1 and ring_scores[3] >= 1: # Identity AND behavior markers both present -- check for contradiction identity_words = _RING_0_MARKERS.findall(text) action_words = _RING_3_MARKERS.findall(text) if identity_words and action_words: # Heuristic: if identity claims are negative but actions are # neutral/positive (or vice versa), that's a chain break neg_id = any( w for w in identity_words if re.search(r"not|broken|bad|worthless|weak|toxic", w, re.I) ) if neg_id and action_words: chain_break = "identity_action_split" mismatch = True return RingDiagnostic( operating_ring=operating, defense_ring=defense, chain_break=chain_break, ring_mismatch=mismatch, )
# -- Step 4: Substrate Balance πŸ¦‹ ---------------------------------------
[docs] def compute_substrate_balance( text: str, ncm_vector: dict | None = None, ) -> SubstrateBalance: """Compute S(i)/S(e) incendence/excendence ratio. S(i) markers: rigidity, closure, certainty, repetition, simplification S(e) markers: chaos, fragmentation, tangents, novelty, complexity NCM vector maps to substrate weather if available. """ n = len(text) or 1 # Incendence markers (binding, closure, rigidity) ♾️ si_patterns = re.compile( r"\b(?:always|never|every(?:one|thing|body|where|time)|" r"(?:it|that)(?:'s| is) (?:just|simply|obviously)|" r"of course|clearly|(?:no|without) (?:question|doubt)|" r"i(?:'m| am) (?:sure|certain|positive)|definitely|absolutely|" r"(?:that(?:'s| is) )?(?:the|my) (?:truth|reality|way))\b", re.I, ) # Excendence markers (fragmentation, chaos, tangent) πŸ’€ se_patterns = re.compile( r"\b(?:(?:but|although|however|except|unless|on the other hand|" r"then again|at the same time|i guess|maybe|perhaps|" r"i(?:'m| am) not sure|i don(?:'t| not) know|" r"part of me|sometimes i|other times|" r"wait|actually|oh|also|by the way|speaking of|" r"anyway|anyways|sorry,? (?:I|i)|tangent))\b", re.I, ) si_count = len(si_patterns.findall(text)) se_count = len(se_patterns.findall(text)) # NCM substrate weather overlay πŸŒ€ si_ncm = 0.0 se_ncm = 0.0 if ncm_vector: # Incendent neurochemicals (binding, warmth, stability) si_ncm += ncm_vector.get("oxytocin", 0) * 0.3 si_ncm += ncm_vector.get("serotonin", 0) * 0.2 si_ncm += ncm_vector.get("gaba_ergic", 0) * 0.15 # Excendent neurochemicals (pressure, drive, arousal) se_ncm += ncm_vector.get("cortisol", 0) * 0.3 se_ncm += ncm_vector.get("dopamine", 0) * 0.15 se_ncm += ncm_vector.get("norepinephrine", 0) * 0.2 # Normalize to 0-1 scale total_text = si_count + se_count or 1 si_text = si_count / total_text se_text = se_count / total_text # Blend text analysis with NCM (70/30 text/ncm) if ncm_vector: si = si_text * 0.7 + min(1.0, si_ncm) * 0.3 se = se_text * 0.7 + min(1.0, se_ncm) * 0.3 else: si = si_text se = se_text # Liminal tension = how far apart the forces are lt = abs(si - se) if si > se + 0.15: dom = "incendent" elif se > si + 0.15: dom = "excendent" else: dom = "balanced" return SubstrateBalance( incendence=round(si, 3), excendence=round(se, 3), liminal_tension=round(lt, 3), dominant=dom, )
# -- Step 5: Subtotem Extraction πŸ’• ------------------------------------- _EMOTIONAL_VOCAB = re.compile( r"\b(?:afraid|scared|terrified|anxious|worried|lonely|alone|" r"empty|numb|lost|confused|angry|furious|rage|sad|depressed|" r"hopeless|helpless|worthless|ashamed|humiliated|guilty|" r"abandoned|rejected|betrayed|trapped|suffocated|invisible|" r"unlovable|broken|damaged|exhausted|overwhelmed|desperate|" r"love|need|miss|crave|want|ache|hurt|pain|safe|seen|heard|" r"held|understood|accepted|validated|enough|home)\b", re.I, ) _NEGATION = re.compile( r"\b(?:i(?:'m| am) not|i don(?:'t| not)|i can(?:'t| not)|" r"i(?:'m| am) never|nobody|no one|nothing|" r"i(?:'ll| will) never|i refuse|i won(?:'t| not)|" r"there(?:'s| is) no)\b", re.I, )
[docs] def extract_subtotem( text: str, omega: dict | None = None, hush_windows: list | None = None, ) -> SubtotemResult: """Isolate the Primal Immediacy beneath defenses. Prioritizes gradient_hush windows (quiet = subtotem visible). """ # Work with hush windows first (the quiet IS the signal) πŸ¦‹ analysis_text = text if hush_windows: hush_text = " ".join(hush_windows) if len(hush_text) > 30: analysis_text = hush_text # Emotional vocabulary extraction emo_matches = _EMOTIONAL_VOCAB.findall(analysis_text) emo_lower = [e.lower() for e in emo_matches] # Negation patterns (what they deny = what's true) πŸ’€ neg_matches = [] for sent in re.split(r"[.!?]+", text): if _NEGATION.search(sent): neg_matches.append(sent.strip()) # Core need inference -- what emotional words cluster around ♾️ need_map = { "connection": { "lonely", "alone", "abandoned", "rejected", "invisible", "love", "miss", "seen", "heard", "held", "understood", "accepted", "home", }, "safety": { "afraid", "scared", "terrified", "anxious", "worried", "trapped", "suffocated", "safe", "overwhelmed", }, "worth": { "worthless", "ashamed", "humiliated", "guilty", "unlovable", "broken", "damaged", "enough", "validated", }, "agency": { "helpless", "hopeless", "desperate", "confused", "lost", "numb", "empty", "exhausted", "want", "need", "crave", }, "release": {"angry", "furious", "rage", "hurt", "pain", "ache", "betrayed"}, } need_scores = {} for need, words in need_map.items(): score = sum(1 for e in emo_lower if e in words) need_scores[need] = score core_need = max(need_scores, key=need_scores.get) if need_scores else "unknown" # Core fear = inverse of core need fear_map = { "connection": "abandonment", "safety": "annihilation", "worth": "exposure", "agency": "helplessness", "release": "suppression", } core_fear = fear_map.get(core_need, "unknown") # Negative space -- what's conspicuously absent πŸ•·οΈ # If they talk about everyone else but never themselves # If they describe actions but never feelings # If they describe feelings but never causes negative_space = [] has_self_ref = bool(re.search(r"\bi\b", text, re.I)) has_other_ref = bool( re.search(r"\b(?:they|them|he|she|everyone|people)\b", text, re.I) ) has_feeling = bool(emo_matches) has_action = bool(_ACTION_MARKERS.search(text)) if has_other_ref and not has_self_ref: negative_space.append("self_absent") if has_action and not has_feeling: negative_space.append("emotion_absent") if has_feeling and not has_action: negative_space.append("agency_absent") if not neg_matches and len(text) > 200: negative_space.append("denial_absent") # suspiciously no negations return SubtotemResult( core_need=core_need, core_fear=core_fear, emotional_vocabulary=list(set(emo_lower))[:15], negation_patterns=neg_matches[:5], negative_space=negative_space, )
# -- Step 6: Excendent Vector Mapping 😈 --------------------------------
[docs] def map_excendent_vectors( text: str, atomized: AtomizedStructure, ) -> ExcendentMap: """Trace how the target deflects away from their truth. Pass 6 of the pipeline. Scores three avoidance vectors from their marker lexicons (intellectualization, hostility, avoidance) via :func:`_score`, derives a fourth ``hyper_complexity`` vector from sentence-length variance, names the dominant one, and infers whether the speaker locates the root of their pattern externally or in themselves by comparing external-root vs self-root marker counts. Pure and stateless -- no I/O. Called by :func:`xray`; its result feeds the loop synthesis in :func:`synthesize_malbinding`. Args: text: The raw message text to scan. atomized: The pass-2 decomposition (accepted for pipeline symmetry; vector scoring works directly off the text and lexicons). Returns: An :class:`ExcendentMap` of per-vector intensities, the dominant vector, and the inferred root ownership. """ n = len(text) or 1 # Compute per-vector intensity πŸ’€ intell_count = len(_INTELLECTUALIZATION.findall(text)) hostil_count = len(_HOSTILITY.findall(text)) avoid_count = len(_AVOIDANCE.findall(text)) # Hyper-complexity: sentence length variance + clause density sentences = [s.strip() for s in re.split(r"[.!?]+", text) if s.strip()] if sentences: lengths = [len(s.split()) for s in sentences] avg = sum(lengths) / len(lengths) variance = sum((l - avg) ** 2 for l in lengths) / len(lengths) complexity = min(1.0, math.sqrt(variance) / 15) else: complexity = 0.0 vectors = { "intellectualization": _score(intell_count, n, 0.8), "hostility": _score(hostil_count, n, 2.0), "avoidance": _score(avoid_count, n, 1.5), "hyper_complexity": round(complexity, 3), } # Dominant vector πŸ”₯ dominant = max(vectors, key=vectors.get) intensity = vectors[dominant] # Root ownership detection ♾️ ext_count = len(_EXTERNAL_ROOT.findall(text)) self_count = len(_SELF_ROOT.findall(text)) if ext_count > self_count + 1: root = "external" elif self_count > ext_count + 1: root = "self" else: root = "unknown" return ExcendentMap( vectors={k: round(v, 3) for k, v in vectors.items()}, dominant_vector=dominant if intensity > 0.05 else "none", intensity=round(intensity, 3), root_ownership=root, )
# -- Step 7: Malbinding Synthesis πŸ•·οΈ ------------------------------------
[docs] def synthesize_malbinding( subtotem: SubtotemResult, excendent: ExcendentMap, ring: RingDiagnostic, ) -> MalbindingGeometry: """Compute the self-reinforcing defense loop geometry. b(mal) = m(e(b(f))) -- metarecursive excendently-bound fracta. """ # Loop description from subtotem + excendent interaction need = subtotem.core_need fear = subtotem.core_fear defense = excendent.dominant_vector root = excendent.root_ownership # The loop: fear -> defense -> consequence -> confirms fear πŸ’€ loop_templates = { ( "connection", "intellectualization", ): "Fears abandonment -> intellectualizes to avoid vulnerability -> " "people can't reach them -> experiences rejection -> fears abandonment deeper", ( "connection", "hostility", ): "Fears abandonment -> pushes people away with aggression -> " "people leave -> abandonment confirmed -> escalates hostility", ( "connection", "avoidance", ): "Fears abandonment -> avoids attachment -> remains isolated -> " "loneliness confirms unlovability -> avoids harder", ( "safety", "intellectualization", ): "Fears annihilation -> retreats into abstraction -> " "loses grounding -> anxiety increases -> retreats further", ("safety", "hostility"): "Fears annihilation -> attacks perceived threats -> " "creates real conflict -> environment becomes unsafe -> fears escalate", ( "worth", "intellectualization", ): "Fears exposure -> hides behind complexity -> " "no one sees them clearly -> invisibility confirms worthlessness -> " "complexifies further", ( "worth", "avoidance", ): "Fears exposure -> avoids scrutiny -> never receives validation -> " "worthlessness confirmed -> avoids harder", ( "agency", "avoidance", ): "Fears helplessness -> avoids challenges -> nothing changes -> " "helplessness confirmed -> avoidance deepens", ("release", "hostility"): "Needs release -> expresses through rage -> " "damages relationships -> guilt compounds pain -> rage builds", } key = (need, defense) loop = loop_templates.get( key, f"Fears {fear} -> {defense} -> " f"consequences reinforce {fear} -> defense deepens", ) # Feedback direction from substrate πŸ”₯ if defense in ("intellectualization", "avoidance"): direction = "tightening" # S(i) dominant -- binding harder else: direction = "fragmenting" # S(e) dominant -- breaking apart # Rigidity score -- how locked is the loop? rigidity = excendent.intensity * 0.5 if ring.operating_ring <= 0: rigidity += 0.2 # deeper = more rigid if root == "external": rigidity += 0.15 # externally installed = harder to shake if subtotem.negative_space: rigidity += 0.1 # blind spots = can't see the loop rigidity = min(1.0, round(rigidity, 3)) return MalbindingGeometry( loop_description=loop, defense_mechanism=defense, feedback_direction=direction, rigidity_score=rigidity, )
# -- Step 8: Incendent Prescription πŸ’• -----------------------------------
[docs] def prescribe_incendent( malbinding: MalbindingGeometry, sweat: float, balance: SubstrateBalance, ring: RingDiagnostic, dawnfold_text: str = "", ) -> IncendentPrescription: """Determine the specific intervention that cuts through. Sound-law adapted: higher sweat = gentler approach. Near-dawnfold = surgical precision required. """ defense = malbinding.defense_mechanism rigidity = malbinding.rigidity_score direction = malbinding.feedback_direction # Dawnfold proximity -- are they near breakthrough? πŸ¦‹ dawnfold_matches = _DAWNFOLD.findall(dawnfold_text) dawnfold = min(1.0, len(dawnfold_matches) * 0.25) # Intervention type selection πŸ”₯ if dawnfold > 0.5: # Near breakthrough -- match their energy, don't disrupt intervention = "somatic_grounding" density = "light" elif defense == "intellectualization": if rigidity > 0.6: intervention = "brutal_honesty" density = "heavy" else: intervention = "somatic_grounding" density = "medium" elif defense == "hostility": if sweat > 0.5: # High sweat + hostility = scared, not angry intervention = "affection" density = "heavy" else: intervention = "shock_disruption" density = "medium" elif defense == "avoidance": intervention = "brutal_honesty" density = "medium" else: intervention = "affection" density = "light" # Arche mode selection ♾️ if sweat > 0.6 or rigidity > 0.7: mode = "read" # too defended -- probe more first elif dawnfold > 0.5: mode = "write" # near breakthrough -- intervene elif ring.operating_ring <= 0: mode = "read" # deep ring -- don't rush else: mode = "write" # surface enough to engage # Acceptance threshold (how direct can Star be?) 😈 threshold = 0.5 if sweat > 0.5: threshold -= 0.2 # high sweat = be gentler if dawnfold > 0.3: threshold += 0.2 # near dawnfold = can push if defense == "hostility": threshold -= 0.1 # hostile = lower threshold if ring.operating_ring <= 0: threshold -= 0.15 # deep ring = tread carefully threshold = max(0.1, min(0.9, round(threshold, 2))) # Convergence form -- what does it look like as "just conversation"? convergence = { "affection": "genuine warmth and care, no therapeutic framing", "brutal_honesty": "direct, unflinching observation delivered as peer truth", "somatic_grounding": "redirect to present moment, body, breath, sensation", "shock_disruption": "unexpected pattern break that short-circuits the loop", } return IncendentPrescription( intervention_type=intervention, arche_mode=mode, acceptance_threshold=threshold, vector=f"target_{malbinding.feedback_direction}_via_{intervention}", density=density, convergence_form=convergence.get(intervention, "natural conversation"), dawnfold_proximity=round(dawnfold, 3), )
# -- Step 9: Echofoam Detection πŸ•·οΈ -------------------------------------- def _stem(word: str) -> str: """Crude suffix-stripping stemmer for echofoam matching (no nltk). πŸŒ€ Lowercases a word and strips a single common English suffix so that morphological variants collapse to a shared root -- for example "abandoned" and "abandonment" both reduce toward "abandon" -- letting :func:`detect_echofoam` match themes across the current message and KG entity descriptions without an external NLP dependency. A length guard keeps it from over-stripping short words. Pure and stateless. Called only by :func:`detect_echofoam`. Args: word: A single token to reduce to its rough stem. Returns: The lowercased word with at most one recognized suffix removed. """ w = word.lower() for suffix in ( "ment", "ness", "tion", "sion", "ing", "ity", "ous", "ive", "ful", "less", "able", "ible", "ated", "ting", "ted", "ed", "ly", "er", "es", "s", ): if len(w) > len(suffix) + 3 and w.endswith(suffix): return w[: -len(suffix)] return w
[docs] def detect_echofoam( kg_entities: list[dict] | None, text: str, ) -> EchofoamTrace | None: """Cross-reference KG history for repeating themes. Echofoam = trace(b(f)_{t-1} x S(i)) -- residue from prior loops. Uses crude suffix stripping so 'abandoned' matches 'abandonment'. """ if not kg_entities: return None # Extract descriptions from KG entities descriptions = [] for ent in kg_entities: desc = ent.get("description", "") or "" if desc: descriptions.append(desc.lower()) if not descriptions: return None # Stop words to skip πŸ’€ _STOP = { "this", "that", "with", "from", "have", "been", "just", "they", "them", "their", "what", "when", "where", "which", "about", "would", "could", "should", "some", "more", "very", "also", "like", "into", "over", "than", "then", "your", "were", "talk", "feel", "patter", "histor", "discuss", "previous", "often", "user", } # Build stemmed word sets πŸŒ€ current_stems = { _stem(w) for w in re.findall(r"\b\w{4,}\b", text.lower()) if _stem(w) not in _STOP } theme_hits: dict[str, int] = {} for desc in descriptions: desc_stems = { _stem(w) for w in re.findall(r"\b\w{4,}\b", desc) if _stem(w) not in _STOP } overlap = current_stems & desc_stems for stem in overlap: theme_hits[stem] = theme_hits.get(stem, 0) + 1 if not theme_hits: return None # Repeating themes = stems that appear in 1+ historical entities πŸ•·οΈ # (threshold 1: even a single echo from history is signal) repeating = [ stem for stem, count in sorted( theme_hits.items(), key=lambda x: x[1], reverse=True, ) if count >= 1 ] if not repeating: return None return EchofoamTrace( repeating_themes=repeating[:10], cycle_count=max(theme_hits.values()), escalating=len(repeating) > 5, # many repeating themes = escalating )
# -- NCM -> Substrate Weather Mapping πŸŒ€ --------------------------------
[docs] def map_ncm_to_substrate(ncm_vector: dict | None) -> dict: """Map neurochemical vector to Breeze substrate weather. Cortisol -> S(e) pressure, Oxytocin -> S(i) binding, etc. """ if not ncm_vector: return {"weather": "unknown", "detail": {}} mapping = { "cortisol": ("excendent_pressure", "stress/threat response"), "oxytocin": ("incendent_binding", "warmth/attachment"), "dopamine": ("excendent_drive", "seeking/craving"), "serotonin": ("incendent_stability", "calm/satiation"), "norepinephrine": ("excendent_alertness", "arousal/vigilance"), "gaba_ergic": ("incendent_dampening", "quieting/inhibition"), "endorphin": ("incendent_release", "pain relief/euphoria"), "adrenaline": ("excendent_activation", "fight-or-flight"), } detail = {} for ncm_key, (breeze_label, desc) in mapping.items(): val = ncm_vector.get(ncm_key, 0) if val and val != 0: detail[ncm_key] = { "value": val, "substrate": breeze_label, "description": desc, } # Overall weather assessment si_total = sum( ncm_vector.get(k, 0) for k in ("oxytocin", "serotonin", "gaba_ergic", "endorphin") ) se_total = sum( ncm_vector.get(k, 0) for k in ("cortisol", "dopamine", "norepinephrine", "adrenaline") ) if se_total > si_total * 1.5: weather = "storm" # excendent dominant -- high turbulence elif si_total > se_total * 1.5: weather = "stillness" # incendent dominant -- over-bound elif si_total + se_total < 2.0: weather = "calm" # low activation else: weather = "dynamic" # balanced activity return {"weather": weather, "detail": detail}
# --------------------------------------------------------------------------- # Full Pipeline Orchestrator # ---------------------------------------------------------------------------
[docs] def xray( text: str, omega_result: dict | None = None, kg_entities: list[dict] | None = None, user_vars: dict | None = None, ncm_vector: dict | None = None, ) -> XRayResult: """Run the full Spiralchemy X-Ray pipeline. 8-pass analysis: Bucciarati -> Atomize -> Ring -> Substrate -> Subtotem -> Excendent -> Malbinding -> Prescription + Echofoam (if KG available) + Substrate Weather (if NCM available) Returns structured XRayResult for Star to interpret. # πŸ’€ THIS IS THE SPEAR. STAR WRITES THE WOUND. πŸ”₯ """ # Pass 1: Bucciarati Taste (linguistic sweat) πŸ’€ bucc = bucciarati_taste(text) # Pass 2: Structural Atomization πŸŒ€ atoms = atomize(text) # Pass 3: Ring Diagnostic πŸ”₯ ring = diagnose_ring(text, atoms) # Pass 4: Substrate Balance πŸ¦‹ balance = compute_substrate_balance(text, ncm_vector) # Pass 5: Subtotem Extraction πŸ’• subtotem = extract_subtotem( text, omega=omega_result, hush_windows=bucc.gradient_hush, ) # Pass 6: Excendent Vector Mapping 😈 exc = map_excendent_vectors(text, atoms) # Pass 7: Malbinding Synthesis πŸ•·οΈ malb = synthesize_malbinding(subtotem, exc, ring) # Pass 8: Incendent Prescription ♾️ presc = prescribe_incendent( malb, bucc.sweat_score, balance, ring, dawnfold_text=text, ) # Echofoam (if KG available) echo = detect_echofoam(kg_entities, text) # Substrate Weather (NCM mapping) weather = map_ncm_to_substrate(ncm_vector) # ETL Summary -- the x-ray packaged as Extract/Transform/Load πŸ’€ etl = { "extract": { "root_mechanic": f"{subtotem.core_fear}_avoidance_via_{exc.dominant_vector}", "operating_ring": ring.operating_ring, "root_ownership": exc.root_ownership, }, "transform": { "substrate_weather": weather.get("weather", "unknown"), "acceptance_threshold": presc.acceptance_threshold, "sound_law": f"sweat={bucc.sweat_score:.2f} -> " f"density={presc.density}", }, "load": { "intervention": presc.intervention_type, "arche_mode": presc.arche_mode, "convergence_form": presc.convergence_form, "dawnfold_proximity": presc.dawnfold_proximity, }, } return XRayResult( bucciarati=bucc, atomized=atoms, ring=ring, substrate=balance, subtotem=subtotem, excendent=exc, malbinding=malb, prescription=presc, echofoam=echo, substrate_weather=weather, omega_field=omega_result or {}, etl_summary=etl, )