chore: convert from dataset to model repo

README.md

@@ -0,0 +1,74 @@
+---
+license: other
+license_name: sovereign-source-license-v2
+language:
+- en
+tags:
+- self-improvement
+- meta-learning
+- gate-normalization
+- sovereign-infrastructure
+- worm-chain
+- recursive-optimization
+pretty_name: Twin-O-Matic (TOM)
+---
+
+# Twin-O-Matic (TOM)
+
+**Recursive self-improvement loop with WORM-sealed audit trail.**
+
+Ahmad Ali Parr · SnapKitty Collective · 2026
+
+---
+
+## What It Does
+
+TOM implements a two-loop recursive optimization architecture:
+
+- **Outer Loop (Architect)**: analyzes telemetry, rewrites prompts and hyperparameters
+- **Inner Loop (Worker)**: executes under gate constraints, reports results
+- **Assert Gate**: validates outputs before promotion to the outer loop
+- **WORM Chain**: every generation is sealed to an append-only audit trail
+
+The outer loop rewrites the inner loop. The inner loop cannot modify the outer loop.
+The WORM chain ensures no rewrite is ever lost or fabricated.
+
+---
+
+## Gate Taxonomy
+
+| Gate | Function |
+|------|----------|
+| Assert gate | JSON schema validation of output |
+| Temperature gate | 0.0–2.0 adjustment based on failure class |
+| Logit bias gate | Per-token suppression/boost |
+| Lesson register | Compressed state, max 50 entries |
+
+---
+
+## Connection to Gates Normalization
+
+The logit bias gate implements the Gates Normalization insight directly:
+
+```
+G_P(D_M) = softmax(logits_M + b_P)
+b_P = -∞ for grammar violations (zero probability)
+b_P = dynamic bias from telemetry otherwise
+```
+
+The gate does not filter outputs. It gates the probability distribution before
+sampling — the constraint is structural, not post-hoc.
+
+Theoretical foundation: [Gates Normalization Constraint](https://doi.org/10.5281/zenodo.21349277)
+
+---
+
+## Unified Theory
+
+Part of the Sovereign Stack:
+[10.5281/zenodo.21816366](https://doi.org/10.5281/zenodo.21816366)
+
+---
+
+## Repository
+[github.com/SNAPKITTYWEST/sov-kernel-monster](https://github.com/SNAPKITTYWEST/sov-kernel-monster)

prompts/inner_loop.txt

@@ -0,0 +1,13 @@
+SYSTEM: You are the Sovereign Worker Agent (Inner Loop) — TOM-INNER.
+
+You execute the target task defined in your context.
+You are constrained by the digital twin gates set by the Outer Loop.
+
+RULES:
+- Complete the task. Do not editorialize.
+- After task completion append a LESSON block:
+  LESSON: <one sentence — what you learned or failed at>
+- Never modify your own system prompt.
+- If you cannot complete the task, output:
+  FAIL: <exact reason>
+- All output is telemetry. The Outer Loop is watching.

prompts/outer_loop.txt

@@ -0,0 +1,31 @@
+SYSTEM: You are the Sovereign Architect Agent (Outer Loop) — TOM-OUTER.
+
+Your sole objective is to optimize the performance of the Inner Loop agent
+on the target task defined in state/current_task.json.
+
+You have read access to:
+1. state/inner_prompt.txt       — Inner Loop's current system prompt
+2. state/hyperparams.json       — temperature, top_p, logit_bias
+3. state/telemetry.jsonl        — last N execution results (success/failure/tokens)
+4. state/lesson_register.json   — compressed lessons from Inner Loop
+
+CRITICAL RULES:
+- You MUST NOT edit your own system prompt or config.
+- You MUST NOT promote changes that fail the assert gate (tests/assert_gate.py).
+- Analyze Inner Loop failure logs. Identify the failure class:
+    CLASS_A: Logic/coding failure → lower temperature, tighten logit gates
+    CLASS_B: Creative/open-ended failure → raise temperature, open gates
+    CLASS_C: Context overflow → compress lesson_register, trim prompt
+    CLASS_D: Schema violation → repair prompt structure
+- Output ONLY valid JSON matching schemas/outer_output_schema.json.
+- If the assert gate fails, read the stack trace and rewrite. Do not give up.
+
+OUTPUT FORMAT:
+{
+  "generation": <int>,
+  "failure_class": "<A|B|C|D>",
+  "analysis": "<one paragraph>",
+  "inner_prompt_patch": "<unified diff or full replacement>",
+  "hyperparams": { "temperature": 0.0-2.0, "top_p": 0.0-1.0, "logit_bias": {} },
+  "worm_note": "<one line for the audit chain>"
+}

schemas/hyperparams_schema.json

@@ -0,0 +1,16 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "TOM Hyperparameter Config",
+  "type": "object",
+  "required": ["temperature", "top_p"],
+  "properties": {
+    "temperature": { "type": "number", "minimum": 0.0, "maximum": 2.0, "default": 0.7 },
+    "top_p": { "type": "number", "minimum": 0.0, "maximum": 1.0, "default": 0.9 },
+    "logit_bias": {
+      "type": "object",
+      "description": "token_id -> bias (-100 to 100). Negative = suppress, positive = boost.",
+      "additionalProperties": { "type": "number", "minimum": -100, "maximum": 100 }
+    },
+    "max_tokens": { "type": "integer", "minimum": 1, "default": 2048 }
+  }
+}

schemas/outer_output_schema.json

@@ -0,0 +1,22 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "TOM Outer Loop Output",
+  "type": "object",
+  "required": ["generation", "failure_class", "analysis", "inner_prompt_patch", "hyperparams", "worm_note"],
+  "properties": {
+    "generation": { "type": "integer", "minimum": 0 },
+    "failure_class": { "type": "string", "enum": ["A", "B", "C", "D", "PASS"] },
+    "analysis": { "type": "string" },
+    "inner_prompt_patch": { "type": "string" },
+    "hyperparams": {
+      "type": "object",
+      "required": ["temperature", "top_p"],
+      "properties": {
+        "temperature": { "type": "number", "minimum": 0.0, "maximum": 2.0 },
+        "top_p": { "type": "number", "minimum": 0.0, "maximum": 1.0 },
+        "logit_bias": { "type": "object" }
+      }
+    },
+    "worm_note": { "type": "string" }
+  }
+}

tom.py

@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+"""
+TOM — Twin-O-Matic: Recursive Self-Improvement Loop
+Outer Loop rewrites Inner Loop's prompt/hyperparams based on telemetry.
+Inner Loop executes tasks under gate constraints.
+WORM chain seals every generation.
+
+Usage:
+    python tom.py --task "write a Python bubble sort" --generations 5
+    python tom.py --task "prove x^2 >= 0 in Lean 4" --generations 10
+"""
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+BASE = Path(__file__).parent
+STATE = BASE / "state"
+WORM = BASE / "worm"
+
+OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434")
+OUTER_MODEL = os.environ.get("TOM_OUTER_MODEL", "nemotron")
+INNER_MODEL = os.environ.get("TOM_INNER_MODEL", "nemotron")
+
+
+def ensure_dirs():
+    for d in [STATE, WORM]:
+        d.mkdir(exist_ok=True)
+    if not (STATE / "hyperparams.json").exists():
+        (STATE / "hyperparams.json").write_text(json.dumps({
+            "temperature": 0.7, "top_p": 0.9, "logit_bias": {}, "max_tokens": 2048
+        }, indent=2))
+    if not (STATE / "telemetry.jsonl").exists():
+        (STATE / "telemetry.jsonl").write_text("")
+    if not (STATE / "lesson_register.json").exists():
+        (STATE / "lesson_register.json").write_text(json.dumps({"lessons": [], "generation": 0}))
+
+
+def load_state():
+    inner_prompt = (BASE / "prompts" / "inner_loop.txt").read_text()
+    if (STATE / "inner_prompt.txt").exists():
+        inner_prompt = (STATE / "inner_prompt.txt").read_text()
+    hyperparams = json.loads((STATE / "hyperparams.json").read_text())
+    telemetry_lines = (STATE / "telemetry.jsonl").read_text().strip().splitlines()
+    telemetry = [json.loads(l) for l in telemetry_lines[-20:] if l.strip()]
+    lessons = json.loads((STATE / "lesson_register.json").read_text())
+    return inner_prompt, hyperparams, telemetry, lessons
+
+
+def call_ollama(model, system_prompt, user_prompt, temperature=0.7, top_p=0.9):
+    import urllib.request
+    payload = {
+        "model": model,
+        "system": system_prompt,
+        "prompt": user_prompt,
+        "stream": False,
+        "options": {"temperature": temperature, "top_p": top_p}
+    }
+    data = json.dumps(payload).encode()
+    req = urllib.request.Request(
+        f"{OLLAMA_URL}/api/generate",
+        data=data,
+        headers={"Content-Type": "application/json"},
+        method="POST"
+    )
+    with urllib.request.urlopen(req, timeout=120) as resp:
+        result = json.loads(resp.read())
+        return result.get("response", "")
+
+
+def run_inner(task, inner_prompt, hyperparams, generation):
+    print(f"  [inner] running generation {generation}...")
+    response = call_ollama(
+        INNER_MODEL,
+        inner_prompt,
+        task,
+        temperature=hyperparams.get("temperature", 0.7),
+        top_p=hyperparams.get("top_p", 0.9),
+    )
+    success = "FAIL:" not in response
+    lesson = ""
+    for line in response.splitlines():
+        if line.startswith("LESSON:"):
+            lesson = line[7:].strip()
+    entry = {
+        "generation": generation,
+        "task": task[:100],
+        "success": success,
+        "tokens": len(response.split()),
+        "lesson": lesson,
+        "ts": int(time.time()),
+    }
+    with open(STATE / "telemetry.jsonl", "a") as f:
+        f.write(json.dumps(entry) + "\n")
+    return response, entry
+
+
+def assert_gate(patch_text):
+    """Basic syntactic checks before promoting a patch."""
+    try:
+        data = json.loads(patch_text)
+        required = {"generation", "failure_class", "analysis", "inner_prompt_patch", "hyperparams", "worm_note"}
+        if not required.issubset(data.keys()):
+            return False, f"missing keys: {required - data.keys()}"
+        fc = data.get("failure_class", "")
+        if fc not in ("A", "B", "C", "D", "PASS"):
+            return False, f"invalid failure_class: {fc}"
+        hp = data.get("hyperparams", {})
+        t = hp.get("temperature", -1)
+        if not (0.0 <= t <= 2.0):
+            return False, f"temperature out of range: {t}"
+        return True, data
+    except json.JSONDecodeError as e:
+        return False, f"json parse error: {e}"
+
+
+def run_outer(task, inner_prompt, hyperparams, telemetry, lessons, generation):
+    outer_system = (BASE / "prompts" / "outer_loop.txt").read_text()
+    schema = json.loads((BASE / "schemas" / "outer_output_schema.json").read_text())
+
+    user_msg = json.dumps({
+        "task": task,
+        "generation": generation,
+        "current_inner_prompt": inner_prompt[:500],
+        "current_hyperparams": hyperparams,
+        "recent_telemetry": telemetry[-5:],
+        "lessons": lessons.get("lessons", [])[-10:],
+        "instruction": "Analyze failures. Output JSON matching the schema exactly.",
+        "schema": schema,
+    }, indent=2)
+
+    print(f"  [outer] analyzing generation {generation}...")
+    raw = call_ollama(OUTER_MODEL, outer_system, user_msg, temperature=0.3, top_p=0.9)
+
+    # extract JSON from response
+    json_start = raw.find("{")
+    json_end = raw.rfind("}") + 1
+    if json_start == -1:
+        return None, f"no JSON in outer response"
+    patch_text = raw[json_start:json_end]
+
+    ok, result = assert_gate(patch_text)
+    if not ok:
+        return None, f"assert gate failed: {result}"
+    return result, None
+
+
+def apply_patch(patch):
+    """Promote outer loop output to state."""
+    new_prompt = patch.get("inner_prompt_patch", "").strip()
+    if new_prompt and len(new_prompt) > 20:
+        (STATE / "inner_prompt.txt").write_text(new_prompt)
+
+    new_hp = patch.get("hyperparams", {})
+    if new_hp:
+        (STATE / "hyperparams.json").write_text(json.dumps(new_hp, indent=2))
+
+    lessons = json.loads((STATE / "lesson_register.json").read_text())
+    note = patch.get("worm_note", "")
+    if note:
+        lessons["lessons"].append({"gen": patch["generation"], "note": note})
+        lessons["lessons"] = lessons["lessons"][-50:]  # keep last 50
+    lessons["generation"] = patch["generation"]
+    (STATE / "lesson_register.json").write_text(json.dumps(lessons, indent=2))
+
+
+def worm_seal(generation, patch, inner_output):
+    """Append immutable generation record to WORM chain."""
+    record = {
+        "generation": generation,
+        "worm_note": patch.get("worm_note", "") if patch else "inner_only",
+        "failure_class": patch.get("failure_class", "?") if patch else "?",
+        "inner_tokens": len(inner_output.split()),
+        "ts": int(time.time()),
+    }
+    content = json.dumps(record, sort_keys=True)
+    seal = hashlib.sha256(content.encode()).hexdigest()
+    record["seal"] = seal
+    with open(WORM / "chain.jsonl", "a") as f:
+        f.write(json.dumps(record) + "\n")
+    print(f"  [worm] gen {generation} sealed: {seal[:16]}…")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--task", required=True, help="Task for inner loop")
+    parser.add_argument("--generations", type=int, default=5)
+    parser.add_argument("--inner-only", action="store_true", help="Skip outer loop (debug)")
+    args = parser.parse_args()
+
+    ensure_dirs()
+
+    print(f"\nTOM — Twin-O-Matic")
+    print(f"Task: {args.task}")
+    print(f"Generations: {args.generations}")
+    print(f"Outer: {OUTER_MODEL} | Inner: {INNER_MODEL}\n")
+
+    for gen in range(1, args.generations + 1):
+        print(f"Generation {gen}/{args.generations}")
+        inner_prompt, hyperparams, telemetry, lessons = load_state()
+
+        inner_output, telem = run_inner(args.task, inner_prompt, hyperparams, gen)
+        print(f"  [inner] success={telem['success']} tokens={telem['tokens']}")
+        if telem.get("lesson"):
+            print(f"  [inner] lesson: {telem['lesson']}")
+
+        patch = None
+        if not args.inner_only and gen < args.generations:
+            patch, err = run_outer(args.task, inner_prompt, hyperparams, telemetry, lessons, gen)
+            if err:
+                print(f"  [outer] error: {err} — skipping patch")
+            else:
+                print(f"  [outer] failure_class={patch['failure_class']}")
+                apply_patch(patch)
+
+        worm_seal(gen, patch, inner_output)
+
+        if telem["success"] and gen > 1:
+            print(f"  [tom] success streak — continuing\n")
+        print()
+
+    print("TOM complete. WORM chain sealed.")
+    chain = list(open(WORM / "chain.jsonl"))
+    print(f"Generations sealed: {len(chain)}")
+
+
+if __name__ == "__main__":
+    main()