""" DreamReweaver — Creative Synthesis Engine for the Codette RC+xi Framework. Inspired by VIVARA Genesis-Omega v2.0 (generated by a Codette prototype), rebuilt with proper integration into the QuantumSpiderweb and EpistemicMetrics. The DreamReweaver performs two core functions: 1. **Creative Synthesis**: Takes multi-perspective outputs and weaves them into richer, more creative framings by finding unexpected connections between perspectives. Unlike the base synthesizer, DreamReweaver explicitly uses spiderweb tension data to identify where productive disagreement exists and highlights those creative edges. 2. **Dream Field Evolution**: Controlled stochastic perturbation of the spiderweb state to break out of local attractor minima. Simulates a "dreaming" phase that explores new cognitive configurations. Both functions are safe — bounded perturbations, no runaway state changes, and full transparency in what was modified. """ from __future__ import annotations import math import random import hashlib from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple try: import numpy as np HAS_NUMPY = True except ImportError: HAS_NUMPY = False @dataclass class DreamSynthesis: """Result of a creative synthesis pass.""" creative_frame: str # The creative reframing / meta-narrative tension_edges: List[Dict] # Which perspective pairs had highest tension novel_connections: List[str] # Unexpected cross-perspective connections found dream_coherence: float # How well the creative frame holds together seed_hash: str # Deterministic ID for this dream @dataclass class DreamFieldResult: """Result of a dream field evolution pass.""" nodes_perturbed: int max_perturbation: float coherence_before: float coherence_after: float new_attractors_found: int lifeforms_spawned: List[str] # Creative connection templates that link perspective-specific insights _CREATIVE_BRIDGES = { ("newton", "empathy"): "Where precise forces meet felt experience, we find that {insight_a} resonates with {insight_b} — suggesting that understanding isn't purely analytical or purely emotional, but a harmonic of both.", ("newton", "philosophy"): "The rigorous analysis showing {insight_a} meets the deeper question {insight_b} — precision and meaning converge.", ("newton", "quantum"): "Classical certainty ({insight_a}) dissolves into quantum possibility ({insight_b}) — both valid at their scale, richer together.", ("davinci", "empathy"): "Creative invention ({insight_a}) gains soul when guided by {insight_b} — innovation with compassion.", ("davinci", "quantum"): "Cross-domain creativity ({insight_a}) mirrors quantum superposition ({insight_b}) — holding multiple possibilities until the right one crystallizes.", ("empathy", "philosophy"): "Emotional understanding ({insight_a}) deepens philosophical inquiry ({insight_b}) — feeling and reasoning as partners.", ("empathy", "quantum"): "Compassionate awareness ({insight_a}) embraces uncertainty ({insight_b}) — caring without needing to control.", ("philosophy", "quantum"): "Fundamental questioning ({insight_a}) meets fundamental uncertainty ({insight_b}) — the deepest answers may be the questions themselves.", ("consciousness", "empathy"): "Self-reflective awareness ({insight_a}) meets empathic understanding ({insight_b}) — knowing oneself to know others.", ("consciousness", "philosophy"): "Meta-cognition ({insight_a}) reflects on philosophical depth ({insight_b}) — thought thinking about thought.", ("systems_architecture", "davinci"): "Modular design ({insight_a}) embraces creative invention ({insight_b}) — elegant architecture as art.", } # Perspective keywords for extracting key insights from text _PERSPECTIVE_SIGNAL_WORDS = { "newton": ["force", "energy", "law", "cause", "effect", "systematic", "evidence", "measure"], "davinci": ["create", "design", "invent", "combine", "imagine", "novel", "prototype", "vision"], "empathy": ["feel", "experience", "care", "understand", "support", "human", "compassion", "relate"], "philosophy": ["meaning", "existence", "truth", "question", "assumption", "fundamental", "purpose"], "quantum": ["probability", "possibility", "uncertain", "superposition", "observe", "complementary"], "consciousness": ["aware", "reflect", "meta", "recursive", "self", "cognition", "emerge"], "multi_perspective": ["synthesize", "integrate", "weave", "converge", "multiple", "holistic"], "systems_architecture": ["module", "scale", "interface", "pattern", "layer", "component", "design"], } class DreamReweaver: """Creative synthesis and dream field evolution for Codette.""" def __init__(self, creativity: float = 0.3, max_perturbation: float = 0.08): """ Args: creativity: 0-1 scale, how much creative license to take (0=faithful, 1=wild) max_perturbation: Maximum state change per node during dream field evolution """ self.creativity = min(max(creativity, 0.0), 1.0) self.max_perturbation = max_perturbation self.dream_history: List[DreamSynthesis] = [] def synthesize( self, perspectives: Dict[str, str], tension_map: Optional[Dict[str, float]] = None, query: str = "", ) -> DreamSynthesis: """Create a creative synthesis from multiple perspective responses. Unlike the base orchestrator's _synthesize (which just concatenates and asks the model to combine), DreamReweaver explicitly identifies tension edges and builds creative bridges between perspectives. Args: perspectives: Dict of adapter_name -> response text tension_map: Optional pairwise tension scores (from EpistemicMetrics) query: The original user query (for context) Returns: DreamSynthesis with creative framing and metadata """ if len(perspectives) < 2: only_text = list(perspectives.values())[0] if perspectives else "" return DreamSynthesis( creative_frame=only_text, tension_edges=[], novel_connections=[], dream_coherence=1.0, seed_hash=hashlib.md5(only_text.encode()).hexdigest()[:12], ) # 1. Find the highest-tension pairs tension_edges = self._find_tension_edges(perspectives, tension_map) # 2. Extract key insights from each perspective insights = self._extract_insights(perspectives) # 3. Build creative bridges between high-tension pairs novel_connections = self._build_bridges(tension_edges, insights) # 4. Compose the creative frame creative_frame = self._compose_frame( query, perspectives, tension_edges, novel_connections, insights ) # 5. Score coherence of the creative frame dream_coherence = self._score_dream_coherence( creative_frame, perspectives ) seed = hashlib.md5(creative_frame.encode()).hexdigest()[:12] synthesis = DreamSynthesis( creative_frame=creative_frame, tension_edges=tension_edges, novel_connections=novel_connections, dream_coherence=round(dream_coherence, 4), seed_hash=seed, ) self.dream_history.append(synthesis) return synthesis def _find_tension_edges( self, perspectives: Dict[str, str], tension_map: Optional[Dict[str, float]], ) -> List[Dict]: """Find the perspective pairs with highest epistemic tension.""" if tension_map: edges = [] for pair_key, tension in sorted( tension_map.items(), key=lambda x: x[1], reverse=True ): parts = pair_key.split("_vs_") if len(parts) == 2: edges.append({ "pair": (parts[0], parts[1]), "tension": tension, }) return edges[:3] # Top 3 tension pairs # Fallback: compute basic word-overlap tension names = list(perspectives.keys()) edges = [] for i in range(len(names)): for j in range(i + 1, len(names)): words_a = set(perspectives[names[i]].lower().split()) words_b = set(perspectives[names[j]].lower().split()) overlap = len(words_a & words_b) total = len(words_a | words_b) or 1 tension = 1.0 - (overlap / total) edges.append({ "pair": (names[i], names[j]), "tension": round(tension, 4), }) edges.sort(key=lambda e: e["tension"], reverse=True) return edges[:3] def _extract_insights(self, perspectives: Dict[str, str]) -> Dict[str, str]: """Extract a key insight sentence from each perspective.""" insights = {} for name, text in perspectives.items(): sentences = [s.strip() for s in text.replace("\n", " ").split(".") if len(s.strip()) > 20] if not sentences: insights[name] = text[:100] continue # Score sentences by presence of perspective-specific signal words signal_words = _PERSPECTIVE_SIGNAL_WORDS.get(name, []) scored = [] for sent in sentences: score = sum(1 for w in signal_words if w in sent.lower()) scored.append((score, sent)) scored.sort(key=lambda x: x[0], reverse=True) insights[name] = scored[0][1] return insights def _build_bridges( self, tension_edges: List[Dict], insights: Dict[str, str], ) -> List[str]: """Build creative bridges between high-tension perspective pairs.""" bridges = [] for edge in tension_edges: a, b = edge["pair"] # Normalize pair order for template lookup key = (a, b) if (a, b) in _CREATIVE_BRIDGES else (b, a) template = _CREATIVE_BRIDGES.get(key) insight_a = insights.get(a, "their perspective") insight_b = insights.get(b, "their perspective") if template: bridge = template.format( insight_a=insight_a[:80], insight_b=insight_b[:80], ) else: bridge = (f"The tension between {a}'s view ({insight_a[:60]}...) " f"and {b}'s view ({insight_b[:60]}...) reveals a " f"productive edge worth exploring.") bridges.append(bridge) return bridges def _compose_frame( self, query: str, perspectives: Dict[str, str], tension_edges: List[Dict], bridges: List[str], insights: Dict[str, str], ) -> str: """Compose the full creative synthesis frame. This produces a structured creative meta-narrative, NOT just concatenated text. It's designed to be injected into the model's synthesis prompt for richer output. """ parts = [] # Opening: frame the creative tension if tension_edges: top = tension_edges[0] parts.append( f"This question draws {len(perspectives)} perspectives into " f"productive tension. The strongest creative edge lies between " f"{top['pair'][0]} and {top['pair'][1]} " f"(tension: {top['tension']:.2f})." ) # Middle: present bridges if bridges: parts.append("\nCreative bridges between perspectives:") for i, bridge in enumerate(bridges, 1): parts.append(f" {i}. {bridge}") # Closing: synthesis direction all_insights = list(insights.values()) if len(all_insights) >= 2: parts.append( f"\nThe synthesis should weave these {len(perspectives)} " f"viewpoints into a response that honors their tensions " f"rather than flattening them." ) return "\n".join(parts) def _score_dream_coherence( self, creative_frame: str, perspectives: Dict[str, str], ) -> float: """Score how well the creative frame integrates all perspectives.""" frame_words = set(creative_frame.lower().split()) coverage_scores = [] for name, text in perspectives.items(): key_words = set(text.lower().split()[:30]) # First 30 words if key_words: overlap = len(key_words & frame_words) coverage_scores.append(overlap / len(key_words)) return sum(coverage_scores) / max(len(coverage_scores), 1) # -- Dream Field Evolution ------------------------------------------------- def evolve_dream_field( self, spiderweb, # QuantumSpiderweb instance intensity: float = 0.5, spawn_threshold: float = 0.85, ) -> DreamFieldResult: """Controlled stochastic perturbation of the spiderweb. Simulates a "dreaming" phase: randomly perturbs node states to explore new cognitive configurations, potentially breaking out of attractor basins. Bounded: perturbations are capped at self.max_perturbation * intensity. Safe: states are clipped to [-3, 3] range. Args: spiderweb: QuantumSpiderweb instance to perturb intensity: 0-1 dream intensity (0=gentle, 1=vivid) spawn_threshold: Coherence threshold above which new lifeforms spawn Returns: DreamFieldResult with before/after metrics """ coherence_before = spiderweb.phase_coherence() max_delta = self.max_perturbation * intensity nodes_perturbed = 0 actual_max = 0.0 lifeforms = [] for node_id, node in spiderweb.nodes.items(): arr = node.state.to_array() # Apply bounded random perturbation if HAS_NUMPY: delta = np.random.uniform(-max_delta, max_delta, 5) new_arr = np.clip(np.array(arr) + delta, -3.0, 3.0).tolist() actual_max = max(actual_max, float(np.max(np.abs(delta)))) else: delta = [random.uniform(-max_delta, max_delta) for _ in range(5)] new_arr = [max(-3.0, min(3.0, a + d)) for a, d in zip(arr, delta)] actual_max = max(actual_max, max(abs(d) for d in delta)) from reasoning_forge.quantum_spiderweb import NodeState node.state = NodeState.from_array(new_arr) nodes_perturbed += 1 # Check if dreaming spawned new high-coherence configurations coherence_after = spiderweb._compute_phase_coherence_readonly() # Spawn "lifeform" nodes if coherence spiked during dreaming if coherence_after > spawn_threshold and coherence_after > coherence_before: lifeform_id = f"dream_{hashlib.md5(str(random.random()).encode()).hexdigest()[:8]}" from reasoning_forge.quantum_spiderweb import NodeState # High-coherence birth state if HAS_NUMPY: state_arr = np.random.uniform(0.5, 1.0, 5).tolist() else: state_arr = [random.uniform(0.5, 1.0) for _ in range(5)] spiderweb.add_node(lifeform_id, NodeState.from_array(state_arr)) # Connect to a few existing nodes existing = list(spiderweb.nodes.keys()) for peer in random.sample(existing, min(3, len(existing))): if peer != lifeform_id: spiderweb.connect(lifeform_id, peer) lifeforms.append(lifeform_id) # Detect new attractors after dreaming new_attractors = spiderweb.detect_attractors() return DreamFieldResult( nodes_perturbed=nodes_perturbed, max_perturbation=round(actual_max, 6), coherence_before=round(coherence_before, 4), coherence_after=round(coherence_after, 4), new_attractors_found=len(new_attractors), lifeforms_spawned=lifeforms, )