diff --git a/tel_deploy/convergence_split.py b/tel_deploy/convergence_split.py index 8c46fb2..46f5d07 100644 --- a/tel_deploy/convergence_split.py +++ b/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.""" diff --git a/test_ollama_local.py b/test_ollama_local.py new file mode 100644 index 0000000..ee09b7b --- /dev/null +++ b/test_ollama_local.py @@ -0,0 +1,120 @@ +""" +TEL convergence battery — local Ollama endpoint. +Usage: python3 test_ollama_local.py +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 ") + 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())