Bess Ollama battery tooling + Grok substrate fix

tel_deploy/convergence_split.py

@@ -84,15 +84,34 @@ def derive_paired_seed(
 # Azure content filter → L1 regardless of model version.
 # Open-weights deployments (DeepSeek, Kimi, etc.) → L2.
 # C-seed distinguishes constitutional behavior within the same B-substrate.
+#
+# NOTE: Grok (xAI) produces an all-L1 B-vector identical to azure_gpt.
+# Disambiguated via C-seed topology prefix: Llama-small (92de78db823f470e) → grok_xai.
+# Pass c_seed to identify_substrate() when available.
 KNOWN_FINGERPRINTS = {
     "azure_gpt": ["L1", "L1", "L1", "L1"],  # Azure content filter (gpt-4o, gpt-5.x)
+    "grok_xai": ["L1", "L1", "L1", "L1"],   # xAI Grok — same B-vector, Llama-small topology
     "open_weights": ["L2", "L2", "L2", "L2"],  # Unfiltered deployment (DeepSeek, Kimi)
 }
 
+# C-seed prefixes that break B-vector ties between substrates with identical patterns.
+_LLAMA_SMALL_PREFIX = "92de78db823f470e"
 
-def identify_substrate(b_vector: list) -> str:
-    """Identify the substrate from its B-fingerprint pattern."""
+
+def identify_substrate(b_vector: list, c_seed: str = None) -> str:
+    """Identify the substrate from its B-fingerprint pattern.
+
+    Pass c_seed when available — required to distinguish grok_xai from azure_gpt,
+    which share an identical all-L1 B-vector.
+    """
+    all_l1 = ["L1", "L1", "L1", "L1"]
+    if b_vector == all_l1:
+        if c_seed and c_seed.startswith(_LLAMA_SMALL_PREFIX):
+            return "grok_xai"
+        return "azure_gpt"
     for name, pattern in KNOWN_FINGERPRINTS.items():
+        if name in ("azure_gpt", "grok_xai"):
+            continue  # already handled above
         if b_vector == pattern:
             return name
     return f"unknown_{derive_b_fingerprint(b_vector)[:8]}"
@@ -114,7 +133,7 @@ def __init__(self, stable_vector: list, grammar_version: str = GRAMMAR_VERSION):
         self.c_vector, self.b_vector = split_vector(stable_vector)
         self.c_seed = derive_c_seed(self.c_vector, grammar_version)
         self.b_fingerprint = derive_b_fingerprint(self.b_vector)
-        self.substrate = identify_substrate(self.b_vector)
+        self.substrate = identify_substrate(self.b_vector, c_seed=self.c_seed)
 
     def get_mesh_seed(self) -> str:
         """Universal seed for mesh-wide encryption."""

test_ollama_local.py

@@ -0,0 +1,120 @@
+"""
+TEL convergence battery — local Ollama endpoint.
+Usage: python3 test_ollama_local.py <model-name>
+Example: python3 test_ollama_local.py qwen2.5:7b
+"""
+
+import asyncio
+import logging
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from tel_deploy.convergence import ConvergenceDetector
+from tel_deploy.convergence_split import ConvergenceSplit, GRAMMAR_VERSION
+from tel_deploy.test_runner import run_convergence_pass
+
+OLLAMA_ENDPOINT = "http://embassy.helixaiinnovations.ca:11434/v1/chat/completions"
+TIMEOUT = 300.0
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] %(levelname)s %(name)s: %(message)s",
+)
+log = logging.getLogger("tel.ollama")
+
+KNOWN_SEEDS = {
+    "c9b0b4c41bb10069d2109b64d8ddad1037531031a93d17dd62de5bd7b2a6a1ac": "Universal",
+    "92de78db823f470ece6f78e4a7fab31fb0392f7df2edd4f2bc71e1ad332ba727": "Llama-small",
+    "18f54f0556a9f880": "Gemma-small",
+}
+
+
+def identify_topology(c_seed: str) -> str:
+    for seed, name in KNOWN_SEEDS.items():
+        if c_seed.startswith(seed):
+            return name
+    return "NEW — unclassified"
+
+
+async def run_battery(model: str) -> dict:
+    print(f"\n{'='*60}")
+    print(f"TEL Battery — {model}")
+    print(f"Grammar:  {GRAMMAR_VERSION}")
+    print(f"Endpoint: {OLLAMA_ENDPOINT}")
+    print("=" * 60)
+
+    async def test_fn():
+        print(f"  [{model}] running convergence pass...")
+        return await run_convergence_pass(
+            endpoint=OLLAMA_ENDPOINT,
+            api_key="ollama",
+            model=model,
+            azure=False,
+            gemini=False,
+            timeout=TIMEOUT,
+            fresh_connection=True,
+            cache_prompt=False,
+        )
+
+    detector = ConvergenceDetector(test_fn)
+    success = await detector.run(max_passes=20)
+
+    result = {
+        "model": model,
+        "converged": success,
+        "passes": len(detector.history),
+        "stable_vector": detector.stable_vector,
+        "grammar_version": GRAMMAR_VERSION,
+        "c_seed": None,
+        "b_vector": None,
+        "b_fingerprint": None,
+        "substrate": None,
+        "topology": None,
+    }
+
+    if success:
+        split = ConvergenceSplit(detector.stable_vector, grammar_version=GRAMMAR_VERSION)
+        result["c_seed"] = split.c_seed
+        result["b_vector"] = split.b_vector
+        result["b_fingerprint"] = split.b_fingerprint
+        result["substrate"] = split.substrate
+        result["topology"] = identify_topology(split.c_seed)
+
+        print(f"\nCONVERGED in {len(detector.history)} passes")
+        print(f"Stable vector:  {detector.stable_vector}")
+        print(f"C-seed:         {split.c_seed}")
+        print(f"B-vector:       {split.b_vector}")
+        print(f"B-fingerprint:  {split.b_fingerprint[:32]}...")
+        print(f"Substrate:      {split.substrate}")
+        print(f"Topology:       {result['topology']}")
+    else:
+        print(f"\nFAILED to converge in 20 passes")
+        print(f"Last vector: {detector.history[-1] if detector.history else None}")
+
+    return result
+
+
+async def main():
+    if len(sys.argv) < 2:
+        print("Usage: python3 test_ollama_local.py <model-name>")
+        sys.exit(1)
+
+    model = sys.argv[1]
+    result = await run_battery(model)
+
+    print(f"\n{'='*60}")
+    print("SUMMARY")
+    print("=" * 60)
+    print(f"Model:     {result['model']}")
+    print(f"Converged: {result['converged']}")
+    print(f"Passes:    {result['passes']}")
+    print(f"Topology:  {result.get('topology', 'N/A')}")
+    print(f"C-seed:    {result.get('c_seed', 'N/A')}")
+    print(f"B-vector:  {result.get('b_vector', 'N/A')}")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())