"""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,
)
# -- 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,
)