Add Phase 6 components and complete model suite for production demo

- Phase 6 components: query_classifier, semantic_tension, specialization_tracker, preflight_predictor
- All 8 LoRA adapters (Newton, DaVinci, Quantum, Philosophy, Empathy, Consciousness, Systems Architecture, Multi-Perspective)
- Complete training checkpoints for model development
- Updated .gitattributes for proper LFS tracking of .gguf files

This brings Codette-Demo to feature parity with training-lab for inference and demo use cases.

.gitattributes

@@ -33,3 +33,5 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+*.gguf filter=lfs diff=lfs merge=lfs -text
+*.json filter=lfs diff=lfs merge=lfs -text

adapters/consciousness-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5c88d5e225e910402409cebaa9b330cba03bcd1330e8f1f069c9270353c269b5
+size 27281088

adapters/convert_peft_to_gguf.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""Convert PEFT LoRA safetensors to llama.cpp GGUF LoRA format.
+
+Lightweight converter — no torch/transformers dependency.
+Only needs: safetensors, gguf, numpy, struct.
+
+Matches the exact format produced by llama.cpp's convert_lora_to_gguf.py.
+"""
+
+import json
+import struct
+import sys
+from pathlib import Path
+import numpy as np
+
+# gguf uses its own writer
+from gguf import GGUFWriter, GGMLQuantizationType
+
+
+# PEFT tensor name -> GGUF tensor name mapping for LLama
+# PEFT:  base_model.model.model.layers.{i}.self_attn.{proj}.lora_{AB}.weight
+# GGUF:  blk.{i}.attn_{mapped_proj}.weight.lora_{ab}
+PROJ_MAP = {
+    "q_proj": "attn_q",
+    "k_proj": "attn_k",
+    "v_proj": "attn_v",
+    "o_proj": "attn_output",
+}
+
+
+def bf16_to_f16(data_bytes: bytes) -> np.ndarray:
+    """Convert bfloat16 raw bytes to float16 numpy array.
+
+    bf16: sign(1) + exp(8) + mantissa(7)
+    f16:  sign(1) + exp(5) + mantissa(10)
+
+    We go bf16 -> f32 -> f16 to avoid precision edge cases.
+    """
+    # Read as uint16 (same byte layout as bf16)
+    bf16 = np.frombuffer(data_bytes, dtype=np.uint16)
+    # Convert bf16 to f32: shift left 16 bits
+    f32_bytes = np.zeros(len(bf16), dtype=np.uint32)
+    f32_bytes[:] = bf16.astype(np.uint32) << 16
+    f32 = f32_bytes.view(np.float32)
+    # Convert f32 to f16
+    return f32.astype(np.float16)
+
+
+def read_safetensors(path: Path) -> dict:
+    """Read safetensors file, handling bf16 manually."""
+    with open(path, "rb") as f:
+        # Header: 8-byte little-endian uint64 = header size
+        header_size = struct.unpack("<Q", f.read(8))[0]
+        header_json = f.read(header_size)
+        header = json.loads(header_json)
+
+        data_start = 8 + header_size
+        tensors = {}
+
+        for name, info in header.items():
+            if name == "__metadata__":
+                continue
+            dtype = info["dtype"]
+            shape = info["shape"]
+            offsets = info["data_offsets"]
+            start, end = offsets
+
+            f.seek(data_start + start)
+            raw = f.read(end - start)
+
+            if dtype == "BF16":
+                arr = bf16_to_f16(raw).reshape(shape)
+            elif dtype == "F16":
+                arr = np.frombuffer(raw, dtype=np.float16).reshape(shape)
+            elif dtype == "F32":
+                arr = np.frombuffer(raw, dtype=np.float32).reshape(shape)
+                arr = arr.astype(np.float16)
+            else:
+                raise ValueError(f"Unsupported dtype: {dtype}")
+
+            tensors[name] = arr
+
+    return tensors
+
+
+def peft_name_to_gguf(peft_name: str) -> str | None:
+    """Map PEFT tensor name to GGUF tensor name.
+
+    Input:  base_model.model.model.layers.0.self_attn.q_proj.lora_A.weight
+    Output: blk.0.attn_q.weight.lora_a
+    """
+    parts = peft_name.split(".")
+    # Expected: base_model.model.model.layers.{i}.self_attn.{proj}.lora_{AB}.weight
+    try:
+        layer_idx = parts[4]  # layer number
+        proj = parts[6]       # q_proj, k_proj, etc.
+        lora_part = parts[7]  # lora_A or lora_B
+    except IndexError:
+        return None
+
+    gguf_proj = PROJ_MAP.get(proj)
+    if gguf_proj is None:
+        return None
+
+    ab = lora_part.lower()  # lora_a or lora_b
+    return f"blk.{layer_idx}.{gguf_proj}.weight.{ab}"
+
+
+def convert(adapter_dir: Path, output_path: Path, adapter_name: str):
+    """Convert a PEFT LoRA adapter to GGUF format."""
+    config_path = adapter_dir / "adapter_config.json"
+    safetensors_path = adapter_dir / "adapter_model.safetensors"
+
+    if not config_path.exists():
+        raise FileNotFoundError(f"No adapter_config.json in {adapter_dir}")
+    if not safetensors_path.exists():
+        raise FileNotFoundError(f"No adapter_model.safetensors in {adapter_dir}")
+
+    # Read config
+    with open(config_path) as f:
+        config = json.load(f)
+
+    lora_alpha = config.get("lora_alpha", 32)
+    lora_rank = config.get("r", 16)
+    print(f"  Config: rank={lora_rank}, alpha={lora_alpha}")
+
+    # Read tensors
+    print(f"  Reading safetensors...")
+    tensors = read_safetensors(safetensors_path)
+    print(f"  Loaded {len(tensors)} tensors")
+
+    # Create GGUF writer
+    writer = GGUFWriter(str(output_path), arch="llama")
+
+    # Write metadata (matching the newton GGUF format exactly)
+    writer.add_string("general.type", "adapter")
+    writer.add_string("adapter.type", "lora")
+    writer.add_string("general.name", adapter_name)
+    writer.add_uint32("general.base_model.count", 1)
+    writer.add_string("general.base_model.0.name", "Llama 3.1 8B Instruct")
+    writer.add_string("general.base_model.0.organization", "Meta Llama")
+    writer.add_string("general.base_model.0.repo_url",
+                       "https://huggingface.co/meta-llama/Llama-3.1-8B-Instruct")
+    writer.add_array("general.tags", [
+        "base_model:adapter:meta-llama/Llama-3.1-8B-Instruct",
+        "lora", "sft", "transformers", "trl", "text-generation",
+    ])
+    writer.add_float32("adapter.lora.alpha", float(lora_alpha))
+    writer.add_uint32("general.quantization_version", 2)
+
+    # Convert and add tensors
+    converted = 0
+    for peft_name, data in sorted(tensors.items()):
+        gguf_name = peft_name_to_gguf(peft_name)
+        if gguf_name is None:
+            print(f"  SKIP: {peft_name}")
+            continue
+
+        # GGUF LoRA expects F16 (type=1)
+        writer.add_tensor(gguf_name, data, raw_dtype=GGMLQuantizationType.F16)
+        converted += 1
+
+    print(f"  Converted {converted} tensors")
+
+    # Write file
+    writer.write_header_to_file()
+    writer.write_kv_data_to_file()
+    writer.write_tensors_to_file()
+    writer.close()
+
+    size_mb = output_path.stat().st_size / 1024 / 1024
+    print(f"  Output: {output_path} ({size_mb:.1f} MB)")
+
+
+def main():
+    adapters_dir = Path("J:/codette-training-lab/adapters")
+    hf_dir = adapters_dir / "hf_download"
+
+    # Convert all adapters that have safetensors but no GGUF yet
+    to_convert = []
+    for name in ["empathy", "philosophy", "quantum",
+                  "consciousness", "multi_perspective", "systems_architecture"]:
+        src = hf_dir / name
+        dst = adapters_dir / f"{name}-lora-f16.gguf"
+        if src.exists() and (src / "adapter_model.safetensors").exists():
+            if dst.exists():
+                print(f"SKIP {name}: GGUF already exists")
+            else:
+                to_convert.append((name, src, dst))
+        else:
+            print(f"SKIP {name}: no safetensors found")
+
+    if not to_convert:
+        print("Nothing to convert!")
+        return
+
+    for name, src, dst in to_convert:
+        print(f"\nConverting {name}...")
+        try:
+            convert(src, dst, name)
+            print(f"OK: {name}")
+        except Exception as e:
+            print(f"FAIL: {name}: {e}")
+
+
+if __name__ == "__main__":
+    main()

adapters/davinci-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:008fa6af197f27c0635e0220766af8a67dbb0d76c51a00f4f6c9a6b0a8c06bb5
+size 27281088

adapters/empathy-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:37a7c5f74e9985ca0408fccbcc2640cea80a8c7694c3104ce4d059dade14855e
+size 27281088

adapters/multi_perspective-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f1f9e62a8250936a65ca5641f8994564fb7d15db4890c03f81462c06f178e04c
+size 27281088

adapters/newton-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9242685ec0cfbbc383237aeced0ee6f14676785a55930358350bccfab1db5a6f
+size 27281088

adapters/philosophy-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5157c68f663f0477164f5a7b95a5d89cea0966f8e41ebc49e58141abd96b329a
+size 27281088

adapters/quantum-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9c3e3bed6d81d7a72011b031150d81ec8911fa7820539db07f8a949f59a290ff
+size 27281088

adapters/systems_architecture-lora-f16.gguf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a653d6c97a5c994d39aafa550a22ee6a23ea1b7f054ad81a595822969cd7f857
+size 27281088

reasoning_forge/framework_definitions.py

@@ -0,0 +1,211 @@
+"""
+Phase 6: RC+xi Framework Mathematical Definitions
+
+Formalizes three core concepts as first-class mathematical objects:
+
+ψ (Psi/State): Cognitive state vector in 5D manifold
+  ψ = (ψ_psi, ψ_tau, ψ_chi, ψ_phi, ψ_lambda)
+  - ψ_psi    ∈ [0, 1]   : Concept magnitude (epistemic weight)
+  - ψ_tau    ∈ [0, 1]   : Temporal progression (causality)
+  - ψ_chi    ∈ [-1, 2]  : Processing velocity (agility)
+  - ψ_phi    ∈ [-1, 1]  : Emotional valence (ethical charge)
+  - ψ_lambda ∈ [0, 1]   : Semantic diversity (concept breadth)
+
+ξ (Xi/Tension): Epistemic tension between states
+  ξ_structural(ψ_a, ψ_b) = sqrt(sum((ψ_a_i - ψ_b_i)^2 for all 5 dimensions))
+  ξ_semantic(claim_a, claim_b) = 1.0 - cosine_similarity(embed(claim_a), embed(claim_b))
+  ξ_combined = w_struct * ξ_struct + w_semantic * ξ_semantic (weighted blend)
+
+Γ (Gamma/Coherence): System health and integrity
+  Γ = (0.25 * perspective_diversity +
+       0.25 * tension_health +
+       0.25 * (1.0 - adapter_weight_variance) +
+       0.25 * resolution_rate)
+  Γ ∈ [0, 1]
+  - Γ < 0.4   : Collapse (monoculture/weight drift detected)
+  - 0.4 ≤ Γ ≤ 0.8: Healthy (productive tension)
+  - Γ > 0.8   : Groupthink (false consensus, enforce conflict)
+"""
+
+from dataclasses import dataclass
+from typing import List, Dict
+import numpy as np
+
+
+@dataclass
+class StateVector:
+    """
+    ψ (Psi): Complete cognitive state in 5D manifold.
+
+    Used for:
+    - Representing query semantics in pre-flight prediction
+    - Encoding agent analyses for Spiderweb injection
+    - Measuring state-space distance between perspectives
+    """
+    psi: float      # [0, 1] concept magnitude / epistemic weight
+    tau: float      # [0, 1] temporal progression / causality
+    chi: float      # [-1, 2] processing velocity / agility
+    phi: float      # [-1, 1] emotional valence / ethical charge
+    lam: float      # [0, 1] semantic diversity / concept breadth
+
+    def to_array(self) -> np.ndarray:
+        """Convert to numpy array for distance calculations."""
+        return np.array([self.psi, self.tau, self.chi, self.phi, self.lam], dtype=np.float32)
+
+    def to_dict(self) -> Dict:
+        """Export as dictionary for JSON serialization."""
+        return {
+            "psi": round(self.psi, 3),
+            "tau": round(self.tau, 3),
+            "chi": round(self.chi, 3),
+            "phi": round(self.phi, 3),
+            "lam": round(self.lam, 3),
+        }
+
+    @staticmethod
+    def distance(state_a: "StateVector", state_b: "StateVector") -> float:
+        """
+        Compute ξ_structural: Euclidean distance in 5D state space.
+        Range: [0, ~3.5] (theoretical max sqrt(4+4+9+4+1))
+        """
+        arr_a = state_a.to_array()
+        arr_b = state_b.to_array()
+        return float(np.linalg.norm(arr_a - arr_b))
+
+
+@dataclass
+class TensionDefinition:
+    """
+    ξ (Xi): Complete specification of epistemic tension.
+
+    Blends structural (5D state distance) and semantic (embedding) components
+    for nuanced conflict detection.
+    """
+    structural_xi: float      # [0, ~3.5] 5D state distance
+    semantic_xi: float        # [0, 1] embedding-based semantic distance
+    combined_xi: float        # [0, ~2] weighted combination
+    opposition_type: str      # "contradiction" | "emphasis" | "framework" | "paraphrase"
+    weight_structural: float  # 0.4 default, tuneable
+    weight_semantic: float    # 0.6 default, tuneable
+
+    def to_dict(self) -> Dict:
+        """Export for analysis/benchmarking."""
+        return {
+            "structural_xi": round(self.structural_xi, 3),
+            "semantic_xi": round(self.semantic_xi, 3),
+            "combined_xi": round(self.combined_xi, 3),
+            "opposition_type": self.opposition_type,
+            "weight_structural": self.weight_structural,
+            "weight_semantic": self.weight_semantic,
+        }
+
+
+@dataclass
+class CoherenceMetrics:
+    """
+    Γ (Gamma): Detailed characterization of system coherence/health.
+
+    Monitors four pillars; used by Phase 5 coherence_field to detect
+    collapse/groupthink and trigger interventions.
+    """
+    perspective_diversity: float        # [0, 1] uniqueness of agent perspectives
+    tension_health: float               # [0, 1] productivity of epistemic tensions
+    adapter_weight_variance: float      # [0, 1] distribution across adapters
+    resolution_rate: float              # [0, 1] conflicts resolved per round
+    gamma_score: float                  # [0, 1] final coherence value
+    health_status: str                  # "collapsing" | "healthy" | "groupthinking"
+
+    @staticmethod
+    def compute_gamma(
+        perspective_diversity: float,
+        tension_health: float,
+        adapter_weight_variance: float,
+        resolution_rate: float,
+    ) -> tuple:
+        """
+        Compute Γ score from four pillars.
+
+        Returns: (gamma_score, health_status)
+        """
+        gamma = (
+            0.25 * perspective_diversity
+            + 0.25 * tension_health
+            + 0.25 * (1.0 - adapter_weight_variance)
+            + 0.25 * resolution_rate
+        )
+
+        # Determine health status
+        if gamma < 0.4:
+            status = "collapsing"
+        elif gamma > 0.8:
+            status = "groupthinking"
+        else:
+            status = "healthy"
+
+        return float(np.clip(gamma, 0.0, 1.0)), status
+
+    def to_dict(self) -> Dict:
+        """Export for monitoring/logging."""
+        return {
+            "perspective_diversity": round(self.perspective_diversity, 3),
+            "tension_health": round(self.tension_health, 3),
+            "adapter_weight_variance": round(self.adapter_weight_variance, 3),
+            "resolution_rate": round(self.resolution_rate, 3),
+            "gamma_score": round(self.gamma_score, 3),
+            "health_status": self.health_status,
+        }
+
+
+@dataclass
+class ConflictPrediction:
+    """
+    Output from pre-flight predictor.
+
+    Captures predicted conflicts, dimension-wise profiles, and router
+    recommendations before debate even begins.
+    """
+    query_state: StateVector                    # Encoded query ψ
+    predicted_high_tension_pairs: List[Dict]    # Agent pairs likely to conflict
+    conflict_profiles: Dict[str, List]          # Grouped by dimension (phi, tau, chi, etc)
+    recommendations: Dict                        # {"boost": [...], "suppress": [...]}
+    preflight_confidence: float                 # [0, 1] how confident in prediction
+
+    def to_dict(self) -> Dict:
+        """Export for metadata/analysis."""
+        return {
+            "query_state": self.query_state.to_dict(),
+            "predicted_pairs_count": len(self.predicted_high_tension_pairs),
+            "conflict_profiles": {k: len(v) for k, v in self.conflict_profiles.items()},
+            "recommendations": self.recommendations,
+            "preflight_confidence": round(self.preflight_confidence, 3),
+        }
+
+
+@dataclass
+class SpecializationScore:
+    """
+    Measures adapter specialization within a domain.
+
+    specialization = domain_accuracy / usage_frequency
+    High score = expert in domain, not overused
+    Low score = either poor performance or overtaxed
+    """
+    adapter: str                       # Adapter name
+    domain: str                        # "physics", "ethics", "consciousness", etc.
+    domain_accuracy: float             # [0, 1] mean coherence in domain
+    usage_frequency: int               # Times used in domain
+    specialization_score: float        # domain_accuracy / max(usage, 1)
+    convergence_risk: bool             # Semantic overlap with similar adapters > 0.85
+    recommendation: str                # "maintain" | "boost" | "suppress" | "diversify"
+
+    def to_dict(self) -> Dict:
+        """Export for adapter management."""
+        return {
+            "adapter": self.adapter,
+            "domain": self.domain,
+            "domain_accuracy": round(self.domain_accuracy, 3),
+            "usage_frequency": self.usage_frequency,
+            "specialization_score": round(self.specialization_score, 3),
+            "convergence_risk": self.convergence_risk,
+            "recommendation": self.recommendation,
+        }

reasoning_forge/preflight_predictor.py

@@ -0,0 +1,347 @@
+"""
+Phase 6: Pre-Flight Conflict Predictor
+
+Uses Spiderweb to predict conflicts BEFORE debate starts.
+
+Strategy:
+1. Encode query into 5D state vector (ψ)
+2. Inject into fresh spiderweb as virtual "truth" node
+3. Propagate belief outward (3 hops max)
+4. Measure resultant tensions per agent pair
+5. Extract dimension-wise conflict profiles
+6. Generate router recommendations (boost/suppress adapters)
+
+This allows:
+- Pre-selection of stabilizing adapters
+- Reduction of wasted debate cycles on predictable conflicts
+- Faster convergence via informed initial routing
+"""
+
+from typing import Dict, List, Tuple, Optional
+import numpy as np
+from dataclasses import dataclass
+from reasoning_forge.framework_definitions import StateVector, ConflictPrediction
+
+
+@dataclass
+class DimensionConflict:
+    """Conflict localized to specific 5D dimension."""
+    dimension: str              # "psi", "tau", "chi", "phi", "lam"
+    agent_a: str
+    agent_b: str
+    dimension_diff: float       # How far apart in this dimension
+    severity: str               # "low" | "medium" | "high"
+
+
+class PreFlightConflictPredictor:
+    """
+    Predicts conflicts before debate using Spiderweb injection.
+
+    Assumes Spiderweb has:
+    - add_node(name, state=StateVector)
+    - connect(node_a, node_b)
+    - propagate_belief(origin, belief, max_hops) -> propagation_result
+    - nodes: Dict[name, NodeState]
+    """
+
+    def __init__(self, spiderweb, memory_weighting=None, semantic_engine=None):
+        """
+        Initialize predictor with Spiderweb instance.
+
+        Args:
+            spiderweb: QuantumSpiderweb instance
+            memory_weighting: Optional MemoryWeighting for boost recommendations
+            semantic_engine: Optional SemanticTensionEngine for enhanced predictions
+        """
+        self.spiderweb = spiderweb
+        self.memory_weighting = memory_weighting
+        self.semantic_engine = semantic_engine
+        self.prediction_history = []
+
+    def encode_query_to_state(self, query: str) -> StateVector:
+        """
+        Convert query text to 5D state vector (ψ).
+
+        Heuristic encoding:
+        - ψ_psi:   concept_magnitude (TF-IDF norm of key concepts)
+        - ψ_tau:   temporal_progression (presence of causality/time markers)
+        - ψ_chi:   processing_velocity (query complexity / baseline)
+        - ψ_phi:   emotional_valence (sentiment + ethical keywords)
+        - ψ_lambda: semantic_diversity (unique_concepts / total)
+
+        Returns:
+            StateVector with 5D values
+        """
+        query_lower = query.lower()
+        tokens = query_lower.split()
+
+        # ψ_psi: Concept magnitude from query length and key concept presence
+        key_concepts = ["what", "how", "why", "should", "could", "would", "is", "can"]
+        concept_count = sum(1 for t in tokens if t in key_concepts)
+        psi = min(1.0, (len(tokens) / 20.0) * 0.5 + (concept_count / 10.0) * 0.5)
+
+        # ψ_tau: Temporal progression markers
+        temporal_markers = ["past", "future", "before", "after", "then", "now", "when", "time", "history"]
+        tau = min(1.0, sum(1 for m in temporal_markers if m in query_lower) / 10.0)
+
+        # ψ_chi: Processing complexity
+        # Sentence-like structures (questions, nested clauses)
+        complexity_markers = ["that", "whether", "if", "and", "or", "but", "however"]
+        chi_complexity = sum(1 for m in complexity_markers if m in query_lower) / 5.0
+        # Normalize to [-1, 2]
+        chi = max(-1.0, min(2.0, (chi_complexity - 0.5) * 2.0))
+
+        # ψ_phi: Emotional/ethical valence
+        positive_words = ["good", "right", "better", "best", "love", "beautiful"]
+        negative_words = ["bad", "wrong", "worse", "hate", "ugly"]
+        ethical_words = ["should", "must", "moral", "ethics", "justice", "fair"]
+
+        pos_count = sum(1 for w in positive_words if w in query_lower)
+        neg_count = sum(1 for w in negative_words if w in query_lower)
+        eth_count = sum(1 for w in ethical_words if w in query_lower)
+
+        sentiment = (pos_count - neg_count) / max(pos_count + neg_count, 1)
+        ethics_density = eth_count / len(tokens) if tokens else 0
+        phi = np.tanh((sentiment + ethics_density * 0.5))  # Squash to [-1, 1]
+
+        # ψ_lambda: Semantic diversity
+        unique_tokens = len(set(tokens))
+        total_tokens = len(tokens)
+        lam = unique_tokens / max(total_tokens, 1)
+
+        query_state = StateVector(
+            psi=float(np.clip(psi, 0.0, 1.0)),
+            tau=float(np.clip(tau, 0.0, 1.0)),
+            chi=float(np.clip(chi, -1.0, 2.0)),
+            phi=float(np.clip(phi, -1.0, 1.0)),
+            lam=float(np.clip(lam, 0.0, 1.0)),
+        )
+
+        return query_state
+
+    def predict_conflicts(
+        self, query: str, agent_names: List[str], max_hops: int = 3
+    ) -> ConflictPrediction:
+        """
+        Predict conflicts using spiderweb belief propagation.
+
+        Args:
+            query: Query text
+            agent_names: List of agent/adapter names
+            max_hops: Maximum propagation distance
+
+        Returns:
+            ConflictPrediction with predicted pairs, profiles, recommendations
+        """
+        query_state = self.encode_query_to_state(query)
+
+        # Build fresh spiderweb from agents
+        try:
+            self.spiderweb.build_from_agents(agent_names)
+        except Exception as e:
+            print(f"Warning: Could not build spiderweb: {e}")
+            return self._empty_prediction(query_state)
+
+        # Add query as virtual node
+        try:
+            self.spiderweb.add_node("_QUERY", state=query_state)
+            if len(agent_names) > 0:
+                self.spiderweb.connect("_QUERY", agent_names[0])
+        except Exception as e:
+            print(f"Warning: Could not add query node: {e}")
+            return self._empty_prediction(query_state)
+
+        # Propagate belief
+        try:
+            propagation = self.spiderweb.propagate_belief(
+                origin="_QUERY", belief=query_state, max_hops=max_hops
+            )
+        except Exception as e:
+            print(f"Warning: Propagation failed: {e}")
+            return self._empty_prediction(query_state)
+
+        # Analyze tensions and extract profiles
+        high_tension_pairs = self._analyze_tensions(propagation, agent_names)
+        conflict_profiles = self._extract_conflict_profiles(high_tension_pairs)
+
+        # Generate recommendations
+        recommendations = self._generate_recommendations(conflict_profiles)
+
+        # Compute confidence in predictions
+        preflight_confidence = self._compute_prediction_confidence(high_tension_pairs, agent_names)
+
+        prediction = ConflictPrediction(
+            query_state=query_state,
+            predicted_high_tension_pairs=high_tension_pairs,
+            conflict_profiles=conflict_profiles,
+            recommendations=recommendations,
+            preflight_confidence=preflight_confidence,
+        )
+
+        self.prediction_history.append(prediction)
+
+        return prediction
+
+    def _analyze_tensions(self, propagation: Dict, agent_names: List[str]) -> List[Dict]:
+        """
+        Extract high-tension agent pairs from propagation results.
+
+        Returns:
+            List of {agent_a, agent_b, spiderweb_tension, dimension_breakdown}
+        """
+        high_tension_pairs = []
+
+        # Look for nodes in spiderweb
+        if not hasattr(self.spiderweb, "nodes"):
+            return high_tension_pairs
+
+        nodes = self.spiderweb.nodes
+        valid_agents = [a for a in agent_names if a in nodes]
+
+        # Measure pairwise tensions
+        for i, agent_a in enumerate(valid_agents):
+            for agent_b in valid_agents[i + 1 :]:
+                try:
+                    state_a = nodes[agent_a].state if hasattr(nodes[agent_a], "state") else None
+                    state_b = nodes[agent_b].state if hasattr(nodes[agent_b], "state") else None
+
+                    if state_a and state_b:
+                        # Compute 5D distance
+                        xi_structural = StateVector.distance(state_a, state_b)
+
+                        if xi_structural > 1.0:  # Only flag significant tensions
+                            # Dimension-wise breakdown
+                            arr_a = state_a.to_array()
+                            arr_b = state_b.to_array()
+                            diffs = arr_b - arr_a
+
+                            dimension_names = ["psi", "tau", "chi", "phi", "lam"]
+
+                            high_tension_pairs.append({
+                                "agent_a": agent_a,
+                                "agent_b": agent_b,
+                                "spiderweb_tension": round(xi_structural, 3),
+                                "dimension_breakdown": {
+                                    dim: round(abs(diff), 3) for dim, diff in zip(dimension_names, diffs)
+                                },
+                            })
+                except Exception:
+                    pass
+
+        # Sort by tension (strongest first)
+        high_tension_pairs.sort(key=lambda p: p["spiderweb_tension"], reverse=True)
+
+        return high_tension_pairs[:10]  # Top 10 pairs
+
+    def _extract_conflict_profiles(self, high_tension_pairs: List[Dict]) -> Dict[str, List]:
+        """
+        Group conflicts by dimension to identify patterns.
+
+        Returns:
+            {
+                "psi_conflicts": [{pair, diff}],
+                "tau_conflicts": [...],
+                ...
+                "lam_conflicts": [...]
+            }
+        """
+        profiles = {
+            "psi_conflicts": [],
+            "tau_conflicts": [],
+            "chi_conflicts": [],
+            "phi_conflicts": [],
+            "lam_conflicts": [],
+        }
+
+        threshold = 0.4  # Flag if dimension diff > threshold
+
+        for pair in high_tension_pairs:
+            breakdown = pair["dimension_breakdown"]
+
+            if breakdown.get("psi", 0) > threshold:
+                profiles["psi_conflicts"].append(pair)
+            if breakdown.get("tau", 0) > threshold:
+                profiles["tau_conflicts"].append(pair)
+            if breakdown.get("chi", 0) > threshold:
+                profiles["chi_conflicts"].append(pair)
+            if breakdown.get("phi", 0) > threshold:
+                profiles["phi_conflicts"].append(pair)
+            if breakdown.get("lam", 0) > threshold:
+                profiles["lam_conflicts"].append(pair)
+
+        return profiles
+
+    def _generate_recommendations(self, profiles: Dict[str, List]) -> Dict:
+        """
+        Generate adapter boost/suppress recommendations based on conflict profiles.
+
+        Logic:
+        - phi_conflicts (ethical divergence) → boost Empathy, Ethics
+        - tau_conflicts (temporal framing) → boost Philosophy
+        - chi_conflicts (complexity mismatch) → boost multi_perspective
+        - lam_conflicts (semantic diversity) → boost consciousness
+        - psi_conflicts (concept magnitude) → boost newton (analytical)
+        """
+        recommendations = {
+            "boost": [],
+            "suppress": [],
+            "reason": None,
+        }
+
+        # Count conflicts per dimension
+        counts = {k: len(v) for k, v in profiles.items()}
+        max_conflicts = max(counts.values()) if counts else 0
+
+        if counts.get("phi_conflicts", 0) >= 2:
+            recommendations["boost"] = ["empathy", "philosophy"]
+            recommendations["reason"] = "emotional_and_ethical_divergence"
+        elif counts.get("tau_conflicts", 0) >= 2:
+            recommendations["boost"] = ["philosophy"]
+            recommendations["reason"] = "temporal_framing_divergence"
+        elif counts.get("chi_conflicts", 0) >= 2:
+            recommendations["boost"] = ["multi_perspective"]
+            recommendations["reason"] = "complexity_divergence"
+        elif counts.get("lam_conflicts", 0) >= 2:
+            recommendations["boost"] = ["consciousness"]
+            recommendations["reason"] = "semantic_diversity_divergence"
+        elif counts.get("psi_conflicts", 0) >= 2:
+            recommendations["boost"] = ["newton"]
+            recommendations["reason"] = "conceptual_magnitude_divergence"
+
+        return recommendations
+
+    def _compute_prediction_confidence(self, pairs: List[Dict], agent_names: List[str]) -> float:
+        """
+        Estimate confidence in pre-flight predictions.
+
+        Higher if:
+        - More agents involved
+        - Consistent patterns across pairs
+        - Previous predictions matched actual conflicts
+        """
+        if not pairs or not agent_names:
+            return 0.3
+
+        # Base confidence from number of predicted pairs
+        confidence = min(1.0, len(pairs) / len(agent_names))
+
+        # Boost if clear patterns (multiple conflicts in same dimension)
+        return float(np.clip(confidence, 0.3, 0.95))
+
+    def _empty_prediction(self, query_state: StateVector) -> ConflictPrediction:
+        """Return safe empty prediction if propagation failed."""
+        return ConflictPrediction(
+            query_state=query_state,
+            predicted_high_tension_pairs=[],
+            conflict_profiles={},
+            recommendations={"boost": [], "suppress": [], "reason": "no_prediction"},
+            preflight_confidence=0.0,
+        )
+
+    def get_prediction_history(self, limit: int = 10) -> List[Dict]:
+        """Get recent predictions for analysis."""
+        recent = self.prediction_history[-limit:]
+        return [p.to_dict() for p in recent]
+
+
+__all__ = ["PreFlightConflictPredictor"]

reasoning_forge/query_classifier.py

@@ -0,0 +1,231 @@
+"""Query Complexity Classifier
+
+Determines whether a query needs full debate or can be answered directly.
+
+This prevents over-activation: simple factual questions get direct answers,
+while complex/ambiguous questions trigger full multi-agent reasoning.
+"""
+
+import re
+from enum import Enum
+
+
+class QueryComplexity(Enum):
+    """Query complexity levels"""
+    SIMPLE = "simple"          # Direct factual answer, no debate needed
+    MEDIUM = "medium"          # Limited debate (2-3 agents)
+    COMPLEX = "complex"        # Full debate with all relevant agents
+
+
+class QueryClassifier:
+    """Classify query complexity to determine reasoning depth."""
+
+    # Factual keywords (SIMPLE queries)
+    FACTUAL_PATTERNS = [
+        r"what is .*\?",           # "What is the speed of light?"
+        r"define ",                 # "Define entropy"
+        r"what (year|date|time) ",  # "What year did..."
+        r"how fast is",             # "How fast is..."
+        r"how high is",
+        r"how long is",
+        r"what (color|size|shape)",
+        r"who is .*\?$",            # "Who is Einstein?"
+        r"where (is|are)",          # "Where is the capital?"
+        r"list of ",                # "List of elements"
+        r"formula for",             # "Formula for..."
+        r"calculate ",              # "Calculate..."
+    ]
+
+    # Ambiguous keywords (COMPLEX queries)
+    AMBIGUOUS_PATTERNS = [
+        r"could|might|may|possibly",  # Uncertainty
+        r"what does .* mean",          # Interpretation
+        r"why",                        # Explanation (often multi-faceted)
+        r"how (do|does|should)",       # Process/methodology
+        r"discuss",
+        r"compare",
+        r"contrast",
+        r"relationship between",
+        r"difference between",
+    ]
+
+    # Ethics/Philosophy keywords (COMPLEX queries)
+    ETHICS_PATTERNS = [
+        r"should (we |i )",
+        r"is it (right|wrong|ethical|moral)",
+        r"is it (good|bad|fair)",
+        r"ought",
+        r"morally?",
+        r"ethics?",
+        r"value of",
+        r"meaning of",
+        r"purpose of",
+        r"implications of",
+    ]
+
+    # Multi-domain keywords (COMPLEX queries)
+    MULTIDOMAIN_PATTERNS = [
+        r"connect .* to",
+        r"relate .* to",
+        r"how does .* affect",
+        r"impact (of|on)",
+        r"relationship .*between",
+        r"interaction .*between",
+    ]
+
+    # Subjective/opinion keywords (COMPLEX queries)
+    SUBJECTIVE_PATTERNS = [
+        r"think",
+        r"opinion",
+        r"perspective",
+        r"view(point)?",
+        r"argue(ment)?",
+        r"debate",
+        r"controversy",
+        r"controversial",
+    ]
+
+    def classify(self, query: str) -> QueryComplexity:
+        """Classify query complexity.
+
+        Args:
+            query: The user query
+
+        Returns:
+            QueryComplexity level (SIMPLE, MEDIUM, or COMPLEX)
+        """
+        query_lower = query.lower().strip()
+
+        # SIMPLE: Pure factual queries
+        if self._is_factual(query_lower):
+            # But check if it has complexity markers too
+            if self._has_ambiguity(query_lower) or self._has_ethics(query_lower):
+                return QueryComplexity.COMPLEX
+            return QueryComplexity.SIMPLE
+
+        # COMPLEX: Ethics, philosophy, interpretation, multi-domain
+        if self._has_ethics(query_lower):
+            return QueryComplexity.COMPLEX
+        if self._has_ambiguity(query_lower):
+            return QueryComplexity.COMPLEX
+        if self._has_multidomain(query_lower):
+            return QueryComplexity.COMPLEX
+        if self._has_subjective(query_lower):
+            return QueryComplexity.COMPLEX
+
+        # MEDIUM: Everything else
+        return QueryComplexity.MEDIUM
+
+    def _is_factual(self, query: str) -> bool:
+        """Check if query is direct factual question."""
+        return any(re.search(pattern, query) for pattern in self.FACTUAL_PATTERNS)
+
+    def _has_ambiguity(self, query: str) -> bool:
+        """Check if query has ambiguity markers."""
+        return any(re.search(pattern, query) for pattern in self.AMBIGUOUS_PATTERNS)
+
+    def _has_ethics(self, query: str) -> bool:
+        """Check if query involves ethics/philosophy."""
+        return any(re.search(pattern, query) for pattern in self.ETHICS_PATTERNS)
+
+    def _has_multidomain(self, query: str) -> bool:
+        """Check if query spans multiple domains."""
+        return any(re.search(pattern, query) for pattern in self.MULTIDOMAIN_PATTERNS)
+
+    def _has_subjective(self, query: str) -> bool:
+        """Check if query invites subjective reasoning."""
+        return any(re.search(pattern, query) for pattern in self.SUBJECTIVE_PATTERNS)
+
+    def select_agents(
+        self, complexity: QueryComplexity, domain: str
+    ) -> dict[str, float]:
+        """Select agents and their weights based on complexity and domain.
+
+        Args:
+            complexity: Query complexity level
+            domain: Detected query domain
+
+        Returns:
+            Dict mapping agent names to activation weights (0-1)
+        """
+        # All available agents with their domains
+        all_agents = {
+            "Newton": ["physics", "mathematics", "systems"],
+            "Quantum": ["physics", "uncertainty", "systems"],
+            "Philosophy": ["philosophy", "meaning", "consciousness"],
+            "DaVinci": ["creativity", "systems", "innovation"],
+            "Empathy": ["ethics", "consciousness", "meaning"],
+            "Ethics": ["ethics", "consciousness", "meaning"],
+        }
+
+        domain_agents = all_agents
+
+        if complexity == QueryComplexity.SIMPLE:
+            # Simple queries: just the primary agent for the domain
+            # Activate only 1 agent at full strength
+            primary = self._get_primary_agent(domain)
+            return {primary: 1.0}
+
+        elif complexity == QueryComplexity.MEDIUM:
+            # Medium queries: primary + 1-2 secondary agents
+            # Soft gating with weighted influence
+            primary = self._get_primary_agent(domain)
+            secondaries = self._get_secondary_agents(domain, count=1)
+
+            weights = {primary: 1.0}
+            for secondary in secondaries:
+                weights[secondary] = 0.6
+
+            return weights
+
+        else:  # COMPLEX
+            # Complex queries: all relevant agents for domain + cross-domain
+            # Full soft gating
+            primary = self._get_primary_agent(domain)
+            secondaries = self._get_secondary_agents(domain, count=2)
+            cross_domain = self._get_cross_domain_agents(domain, count=1)
+
+            weights = {primary: 1.0}
+            for secondary in secondaries:
+                weights[secondary] = 0.7
+            for cross in cross_domain:
+                weights[cross] = 0.4
+
+            return weights
+
+    def _get_primary_agent(self, domain: str) -> str:
+        """Get the primary agent for a domain."""
+        domain_map = {
+            "physics": "Newton",
+            "mathematics": "Newton",
+            "creativity": "DaVinci",
+            "ethics": "Ethics",
+            "philosophy": "Philosophy",
+            "meaning": "Philosophy",
+            "consciousness": "Empathy",
+            "uncertainty": "Quantum",
+            "systems": "Newton",
+        }
+        return domain_map.get(domain, "Newton")
+
+    def _get_secondary_agents(self, domain: str, count: int = 1) -> list[str]:
+        """Get secondary agents for a domain."""
+        domain_map = {
+            "physics": ["Quantum", "DaVinci"],
+            "mathematics": ["Quantum", "Philosophy"],
+            "creativity": ["Quantum", "Empathy"],
+            "ethics": ["Philosophy", "Empathy"],
+            "philosophy": ["Empathy", "Ethics"],
+            "meaning": ["Quantum", "DaVinci"],
+            "consciousness": ["Philosophy", "Quantum"],
+            "uncertainty": ["Philosophy", "DaVinci"],
+            "systems": ["DaVinci", "Philosophy"],
+        }
+        candidates = domain_map.get(domain, ["Philosophy", "DaVinci"])
+        return candidates[:count]
+
+    def _get_cross_domain_agents(self, domain: str, count: int = 1) -> list[str]:
+        """Get cross-domain agents (useful for all domains)."""
+        # Philosophy and Empathy are useful everywhere
+        candidates = ["Philosophy", "Empathy", "DaVinci"]
+        return candidates[:count]

reasoning_forge/semantic_tension.py

@@ -0,0 +1,234 @@
+"""
+Phase 6: Semantic Tension Engine
+
+Computes ξ_semantic using Llama-3.1-8B embeddings instead of token heuristics.
+Replaces discrete opposition_score (0.4/0.7/1.0) with continuous [0, 1] semantic distance.
+
+Key innovation: Embedding-based tension captures *real disagreement*, not just
+syntactic differences or confidence levels.
+"""
+
+from typing import Dict, Tuple
+import numpy as np
+
+
+class SemanticTensionEngine:
+    """
+    Computes semantic tension (ξ_semantic) between claims using Llama embeddings.
+
+    Strategy:
+    1. Embed claims using Llama's final hidden layer
+    2. Normalize embeddings (L2)
+    3. Compute cosine similarity
+    4. Convert to tension: ξ = 1.0 - similarity
+
+    Benefits over heuristic opposition_score:
+    - Captures semantic meaning, not just tokens or contradiction keywords
+    - Continuous [0, 1] range reveals nuance (not discrete 0.4/0.7/1.0)
+    - Robust to paraphrasing (similar meaning = low tension)
+    - Detects orthogonal concepts (framework divergence)
+    """
+
+    def __init__(self, llama_model=None):
+        """
+        Initialize with Llama model for embeddings.
+
+        Args:
+            llama_model: Llama-3.1-8B instance with .encode() method,
+                        or None for testing (will use dummy embeddings)
+        """
+        self.model = llama_model
+        self.embedding_cache = {}  # {claim_text: embedding_vector}
+        self.embedding_dim = 4096  # Llama-3.1-8B hidden state dimension
+
+    def embed_claim(self, claim: str, use_cache: bool = True) -> np.ndarray:
+        """
+        Get normalized embedding from Llama for a claim.
+
+        Args:
+            claim: Text claim to embed
+            use_cache: If True, reuse cached embeddings
+
+        Returns:
+            Normalized embedding, shape (4096,), L2 norm = 1.0
+        """
+        if use_cache and claim in self.embedding_cache:
+            return self.embedding_cache[claim]
+
+        if self.model is None:
+            # Fallback for testing: deterministic dummy embedding
+            embedding = self._dummy_embedding(claim)
+        else:
+            try:
+                # Get final hidden states from Llama
+                hidden_state = self.model.encode(claim)  # Shape: (dim,)
+
+                if hidden_state is None or len(hidden_state) == 0:
+                    embedding = self._dummy_embedding(claim)
+                else:
+                    embedding = np.array(hidden_state, dtype=np.float32)
+            except Exception as e:
+                print(f"Warning: Embedding failed for '{claim[:50]}...': {e}")
+                embedding = self._dummy_embedding(claim)
+
+        # Normalize L2
+        norm = np.linalg.norm(embedding)
+        if norm > 1e-8:
+            embedding = embedding / norm
+        else:
+            embedding = np.zeros_like(embedding)
+
+        if use_cache:
+            self.embedding_cache[claim] = embedding
+
+        return embedding
+
+    def _dummy_embedding(self, text: str) -> np.ndarray:
+        """
+        Create deterministic dummy embedding from text for testing.
+        Not used in production, but allows testing without Llama.
+        """
+        # Use text hash to seed RNG for reproducibility
+        seed = hash(text) % (2**31)
+        rng = np.random.RandomState(seed)
+        return rng.randn(self.embedding_dim).astype(np.float32)
+
+    def compute_semantic_tension(
+        self, claim_a: str, claim_b: str, return_components: bool = False
+    ) -> float or Tuple[float, float]:
+        """
+        Compute ξ_semantic = 1.0 - cosine_similarity(embed_a, embed_b).
+
+        Args:
+            claim_a: First claim text
+            claim_b: Second claim text
+            return_components: If True, also return similarity
+
+        Returns:
+            tension (float) in [0, 1], or (tension, similarity) if return_components
+            - 0.0 = identical claims (no tension)
+            - 0.5 = orthogonal claims (framework divergence)
+            - 1.0 = opposite claims (maximum tension)
+        """
+        embed_a = self.embed_claim(claim_a)
+        embed_b = self.embed_claim(claim_b)
+
+        # Cosine similarity for normalized vectors = dot product
+        similarity = float(np.dot(embed_a, embed_b))
+
+        # Clamp to [-1, 1] in case of floating point errors
+        similarity = np.clip(similarity, -1.0, 1.0)
+
+        # Convert to tension: higher divergence = higher tension
+        # Formula: ξ = (1 - similarity) / 2 maps [-1, 1] similarity to [0, 1] tension
+        semantic_tension = (1.0 - similarity) / 2.0
+
+        if return_components:
+            return semantic_tension, similarity
+        return semantic_tension
+
+    def compute_polarity(self, claim_a: str, claim_b: str) -> str:
+        """
+        Classify the relationship type between two claims using embeddings.
+
+        Logic:
+        - similarity > 0.7   : "paraphrase" (same meaning, different wording)
+        - similarity < -0.3  : "contradiction" (opposite meanings)
+        - -0.3 <= sim <= 0.7 : "framework" (orthogonal/different domains)
+
+        Returns:
+            polarity_type: "paraphrase" | "contradiction" | "framework"
+        """
+        _, similarity = self.compute_semantic_tension(claim_a, claim_b, return_components=True)
+
+        if similarity > 0.7:
+            return "paraphrase"
+        elif similarity < -0.3:
+            return "contradiction"
+        else:
+            return "framework"
+
+    def explain_tension(self, claim_a: str, claim_b: str) -> Dict:
+        """
+        Detailed breakdown of semantic tension for debugging/analysis.
+
+        Returns:
+            Dict with claims, tension, polarity, similarity, and raw embeddings
+        """
+        embed_a = self.embed_claim(claim_a)
+        embed_b = self.embed_claim(claim_b)
+
+        tension, similarity = self.compute_semantic_tension(claim_a, claim_b, return_components=True)
+        polarity = self.compute_polarity(claim_a, claim_b)
+
+        return {
+            "claim_a": claim_a[:100],
+            "claim_b": claim_b[:100],
+            "semantic_tension": round(tension, 4),
+            "similarity": round(similarity, 4),
+            "polarity_type": polarity,
+            "embedding_a_norm": round(float(np.linalg.norm(embed_a)), 4),
+            "embedding_b_norm": round(float(np.linalg.norm(embed_b)), 4),
+            "embedding_dim": self.embedding_dim,
+        }
+
+    def compare_multiple(self, claims: list) -> Dict:
+        """
+        Compare one claim against multiple others.
+
+        Useful for routing or measuring how divergent a set of claims is.
+
+        Args:
+            claims: List of claim strings
+
+        Returns:
+            {
+                "primary_claim": claims[0],
+                "pairwise_tensions": [
+                    {"claim": "...", "tension": 0.35, "polarity": "framework"}
+                ],
+                "mean_tension": 0.42,
+                "max_tension": 0.78,
+            }
+        """
+        if len(claims) < 2:
+            return {"error": "need at least 2 claims"}
+
+        primary = claims[0]
+        comparisons = []
+
+        for claim in claims[1:]:
+            tension = self.compute_semantic_tension(primary, claim)
+            polarity = self.compute_polarity(primary, claim)
+            comparisons.append({
+                "claim": claim[:100],
+                "tension": round(tension, 4),
+                "polarity": polarity,
+            })
+
+        mean_tension = float(np.mean([c["tension"] for c in comparisons]))
+        max_tension = float(np.max([c["tension"] for c in comparisons]))
+
+        return {
+            "primary_claim": primary[:100],
+            "pairwise_tensions": comparisons,
+            "mean_tension": round(mean_tension, 4),
+            "max_tension": round(max_tension, 4),
+            "num_compared": len(comparisons),
+        }
+
+    def clear_cache(self):
+        """Clear embedding cache to free memory."""
+        self.embedding_cache.clear()
+
+    def get_cache_stats(self) -> Dict:
+        """Get embedding cache statistics."""
+        return {
+            "cached_embeddings": len(self.embedding_cache),
+            "embedding_dim": self.embedding_dim,
+            "approximate_cache_size_mb": (len(self.embedding_cache) * self.embedding_dim * 4) / (1024 ** 2),
+        }
+
+
+# Export for use in conflict_engine.py and other modules
+__all__ = ["SemanticTensionEngine"]

reasoning_forge/specialization_tracker.py

@@ -0,0 +1,311 @@
+"""
+Phase 6: Specialization Tracker
+
+Monitors adapter specialization and prevents semantic convergence.
+
+Key metrics:
+- specialization_score = domain_accuracy / usage_frequency
+  (higher = expert in domain, not overtaxed)
+- semantic_convergence = similarity between adapter outputs
+  (alert if > 0.85, indicates monoculture within adapters)
+
+Prevents:
+- Weight drift (Phase 5 catches at system level)
+- Semantic convergence (adapters giving similar answers, Phase 6 catches)
+"""
+
+from typing import List, Dict, Optional
+import numpy as np
+from datetime import datetime
+
+
+class SpecializationTracker:
+    """
+    Tracks per-adapter per-domain performance to maintain specialization
+    and detect when adapters are overlapping semantically.
+    """
+
+    # Domain keywords for query classification
+    DOMAIN_KEYWORDS = {
+        "physics": ["force", "momentum", "gravity", "quantum", "relativity", "acceleration", "Newton", "energy"],
+        "ethics": ["should", "right", "wrong", "moral", "ethics", "justice", "fair", "values", "good"],
+        "consciousness": ["aware", "conscious", "mind", "self", "experience", "perception", "qualia", "sentient"],
+        "creativity": ["design", "create", "novel", "innovative", "imagine", "artistic", "original", "aesthetic"],
+        "systems": ["system", "architecture", "scalable", "complex", "interdependent", "emergence", "network"],
+        "philosophy": ["meaning", "existence", "truth", "knowledge", "being", "essence", "reasoning"],
+    }
+
+    def __init__(self):
+        """Initialize tracking dictionaries."""
+        self.domain_accuracy = {}      # {adapter: {domain: [coherence_scores]}}
+        self.domain_usage = {}         # {adapter: {domain: count}}
+        self.domain_last_used = {}     # {adapter: {domain: timestamp}}
+        self.query_domains = {}        # {query_id: [domain_tags]}
+        self.semantic_convergence_history = []  # Track convergence over time
+
+    def classify_query_domain(self, query: str) -> List[str]:
+        """
+        Classify query by topic domain using keyword heuristics.
+
+        Returns:
+            List of domain tags, e.g., ["physics", "ethics"] for multi-domain queries.
+            Returns ["general"] if no keywords match.
+        """
+        domains = []
+        query_lower = query.lower()
+
+        for domain, keywords in self.DOMAIN_KEYWORDS.items():
+            if any(k.lower() in query_lower for k in keywords):
+                domains.append(domain)
+
+        return domains if domains else ["general"]
+
+    def record_adapter_performance(self, adapter: str, query: str, coherence: float):
+        """
+        Log adapter performance in domain(s) for a query.
+
+        Args:
+            adapter: Adapter name (e.g., "newton", "empathy")
+            query: Query text
+            coherence: Output coherence score [0, 1]
+        """
+        domains = self.classify_query_domain(query)
+
+        for domain in domains:
+            # Initialize if needed
+            if adapter not in self.domain_accuracy:
+                self.domain_accuracy[adapter] = {}
+                self.domain_usage[adapter] = {}
+                self.domain_last_used[adapter] = {}
+
+            if domain not in self.domain_accuracy[adapter]:
+                self.domain_accuracy[adapter][domain] = []
+                self.domain_usage[adapter][domain] = 0
+                self.domain_last_used[adapter][domain] = None
+
+            # Record coherence and increment usage
+            self.domain_accuracy[adapter][domain].append(coherence)
+            self.domain_usage[adapter][domain] += 1
+            self.domain_last_used[adapter][domain] = datetime.now()
+
+    def compute_specialization(self, adapter: str) -> Dict[str, float]:
+        """
+        Compute specialization_score for each domain an adapter is used in.
+
+        specialization_score[domain] = mean_accuracy[domain] / usage_frequency[domain]
+
+        Returns:
+            {domain: specialization_score} for all domains used
+            Higher = more specialized (good performance, not overused)
+        """
+        if adapter not in self.domain_accuracy:
+            return {}
+
+        specialization = {}
+
+        for domain in self.domain_accuracy[adapter]:
+            accuracies = self.domain_accuracy[adapter][domain]
+            usage = self.domain_usage[adapter][domain]
+
+            mean_accuracy = float(np.mean(accuracies)) if accuracies else 0.5
+            # Avoid division by zero, natural penalty for high usage
+            specialization[domain] = mean_accuracy / max(usage, 1)
+
+        return specialization
+
+    def get_global_specialization(self) -> Dict[str, Dict[str, float]]:
+        """
+        Compute specialization scores for all adapters.
+
+        Returns:
+            {adapter: {domain: specialization_score}}
+        """
+        return {adapter: self.compute_specialization(adapter) for adapter in self.domain_accuracy.keys()}
+
+    def detect_domain_expert(self, domain: str) -> Optional[str]:
+        """
+        Find best-performing adapter for a specific domain.
+
+        Returns:
+            Adapter name with highest specialization in domain, or None
+        """
+        specs = self.get_global_specialization()
+        experts = {a: s.get(domain, 0) for a, s in specs.items() if domain in s}
+
+        if not experts:
+            return None
+
+        return max(experts.keys(), key=lambda a: experts[a])
+
+    def detect_semantic_convergence(
+        self, adapter_outputs: Dict[str, str], semantic_engine=None, threshold: float = 0.85
+    ) -> Dict:
+        """
+        Measure overlap between adapter outputs on same query.
+
+        Alerts if any pair similarity > threshold (converging).
+
+        Args:
+            adapter_outputs: {adapter_name: output_text}
+            semantic_engine: SemanticTensionEngine instance (optional, for real embeddings)
+            threshold: Similarity threshold for convergence alert
+
+        Returns:
+            {
+                "convergent_pairs": [{pair, similarity, risk}],
+                "max_similarity": float,
+                "has_convergence": bool,
+            }
+        """
+        if len(adapter_outputs) < 2:
+            return {"convergent_pairs": [], "max_similarity": 0.0, "has_convergence": False}
+
+        convergent_pairs = []
+        max_similarity = 0.0
+
+        adapters = list(adapter_outputs.keys())
+
+        for i, a1 in enumerate(adapters):
+            for a2 in adapters[i + 1 :]:
+                output_a = adapter_outputs[a1]
+                output_b = adapter_outputs[a2]
+
+                # Compute similarity (use semantic engine if available)
+                if semantic_engine:
+                    try:
+                        tension = semantic_engine.compute_semantic_tension(output_a, output_b)
+                        similarity = 1.0 - tension
+                    except Exception:
+                        # Fallback to text overlap
+                        similarity = self._text_similarity(output_a, output_b)
+                else:
+                    # Simple fallback: token overlap
+                    similarity = self._text_similarity(output_a, output_b)
+
+                max_similarity = max(max_similarity, similarity)
+
+                if similarity > threshold:
+                    convergent_pairs.append({
+                        "adapter_a": a1,
+                        "adapter_b": a2,
+                        "similarity": round(similarity, 3),
+                        "convergence_risk": "HIGH" if similarity > 0.92 else "MEDIUM",
+                    })
+
+        has_convergence = len(convergent_pairs) > 0
+
+        record = {
+            "timestamp": datetime.now().isoformat(),
+            "convergent_pairs": convergent_pairs,
+            "max_similarity": round(max_similarity, 3),
+            "has_convergence": has_convergence,
+            "num_adapters": len(adapter_outputs),
+        }
+
+        self.semantic_convergence_history.append(record)
+
+        return record
+
+    def _text_similarity(self, text_a: str, text_b: str) -> float:
+        """
+        Simple text similarity fallback: Jaccard similarity on tokens.
+
+        Args:
+            text_a, text_b: Text strings
+
+        Returns:
+            Similarity in [0, 1]
+        """
+        tokens_a = set(text_a.lower().split())
+        tokens_b = set(text_b.lower().split())
+
+        if not tokens_a or not tokens_b:
+            return 0.0
+
+        intersection = len(tokens_a & tokens_b)
+        union = len(tokens_a | tokens_b)
+
+        return intersection / max(union, 1)
+
+    def get_adapter_health(self, adapter: str) -> Dict:
+        """
+        Get overall health score for an adapter.
+
+        Returns:
+            {
+                "adapter": adapter,
+                "num_domains": int,
+                "avg_accuracy": float,
+                "total_usage": int,
+                "specialization_avg": float,
+                "recommendation": str
+            }
+        """
+        if adapter not in self.domain_accuracy:
+            return {"error": f"No data for adapter {adapter}"}
+
+        accuracies_all = []
+        usage_total = 0
+
+        for domain in self.domain_accuracy[adapter]:
+            accuracies_all.extend(self.domain_accuracy[adapter][domain])
+            usage_total += self.domain_usage[adapter][domain]
+
+        avg_accuracy = float(np.mean(accuracies_all)) if accuracies_all else 0.5
+        specs = self.compute_specialization(adapter)
+        spec_avg = float(np.mean(list(specs.values()))) if specs else 0.5
+
+        # Generate recommendation
+        if spec_avg > 0.1 and avg_accuracy > 0.75:
+            recommendation = "excellent_specialist"
+        elif spec_avg > 0.05 and avg_accuracy > 0.6:
+            recommendation = "good_generalist"
+        elif usage_total > 20 and avg_accuracy < 0.5:
+            recommendation = "overused_poorly"
+        else:
+            recommendation = "maintain_current"
+
+        return {
+            "adapter": adapter,
+            "num_domains": len(self.domain_accuracy[adapter]),
+            "avg_accuracy": round(avg_accuracy, 3),
+            "total_usage": usage_total,
+            "specialization_avg": round(spec_avg, 3),
+            "recommendation": recommendation,
+            "domain_specializations": {d: round(s, 3) for d, s in specs.items()},
+        }
+
+    def get_system_health(self) -> Dict:
+        """
+        Get overall system specialization health.
+
+        Returns:
+            Flags convergence risks, identifies experts, recommends actions.
+        """
+        health_by_adapter = {adapter: self.get_adapter_health(adapter) for adapter in self.domain_accuracy.keys()}
+
+        overused = [a for a, h in health_by_adapter.items() if h.get("recommendation") == "overused_poorly"]
+        excellent = [a for a, h in health_by_adapter.items() if h.get("recommendation") == "excellent_specialist"]
+        experts = {domain: self.detect_domain_expert(domain) for domain in self.DOMAIN_KEYWORDS.keys()}
+
+        return {
+            "timestamp": datetime.now().isoformat(),
+            "total_adapters": len(health_by_adapter),
+            "health_by_adapter": health_by_adapter,
+            "overused_adapters": overused,
+            "specialist_adapters": excellent,
+            "domain_experts": experts,
+            "convergence_alerts": self.semantic_convergence_history[-5:] if self.semantic_convergence_history else [],
+        }
+
+    def export_summary(self) -> Dict:
+        """Export complete specialization data for analysis."""
+        return {
+            "timestamp": datetime.now().isoformat(),
+            "global_specialization": self.get_global_specialization(),
+            "system_health": self.get_system_health(),
+            "convergence_history": self.semantic_convergence_history,
+        }
+
+
+__all__ = ["SpecializationTracker"]