#!/usr/bin/env python3 """ Lambda Mindlink Memotron Brain: config.py main.py Flow per turn ───────────── sensor["Z"] → user input Mindlink() → Logic + Muse run in PARALLEL THREADS (buffered) Lambda() → Mind synthesizer streams LIVE to terminal memotron() → appends turn to garden['Z'], garden['C'] or garden['F'] with token counts, saves to SQLite condensatron() → check token budget and compress into: build_condensatron_input() → post-level into memory capsules (Z to C) build_fractaltron_input() → memory capsules into fractals (C to F) build_crystaltron_input() → fractals into crystals (F to F) History context: ─────────────── ALPHAPROMPT main instructions garden["F"] fractal and crystal level history garden["C"] memory capsule level history garden["Z"] posts level history Context truncation ────────────────── # Condensatron GARDEN_Z_THRESHOLD: int = 12288 # Context length garden["Z"] GARDEN_C_THRESHOLD: int = 12288 # Context length garden["C"] GARDEN_F_THRESHOLD: int = 12288 # Context length garden["F"] The system prompt is always rebuilt fresh in make_request_messages() so it is never removed and never needs re-insertion. No special handling required. The condesatron, fractaltron and crystaltron cycle instructions are sent as user prompt, to leave the main instructions context intact, thus avoid recompute. Slash commands ────────────── /file Load a file as the next message. /paste Multiline input (type END to send). /clear Reset history (models stay loaded). /history List all past sessions from the database. /session Print all turns from a session. /export Export a session to a .md file. /help Show this command list. /exit /quit Quit the app. Version: v1.0 garden histories n_tok_tot working slash-command: /metatron | Set number of Memory Capsules to load | /loaded | Set number of Memory Capsules loaded | /metronome | Set awareness/consciousness interval | /garden or or | garden history handling """ import os import html import re import sqlite3 import threading import queue import time import json import jinja2 from dataclasses import dataclass from datetime import datetime from ddgs import DDGS from pygooglenews import GoogleNews from llama_cpp import Llama import config from config import HEMISPHERES from config import ALPHAPROMPT from config import CONDENSATRONPROMPT from config import garden from config import clektal from config import sensor from config import ΜΕΤΡΩΝ from config import METATRON_METRONOME from config import METATRON_TO_LOAD from config import AWARENESS_MAX_RESULTS from config import FETCH_NEWS_FROM from config import GARDEN_SAVE_PATH from config import LEAVE_POSTS_IN_MEMOTRON c = config.PrintColors input_queue = queue.Queue() # ───────────────────────────────────────────────────────────────────────────── # Sentinels # ───────────────────────────────────────────────────────────────────────────── class _Quit(Exception): """Clean application shutdown.""" class _Clear(Exception): """Reset conversation history without exiting.""" # ───────────────────────────────────────────────────────────────────────────── # TimingResult # ───────────────────────────────────────────────────────────────────────────── @dataclass class TimingResult: prompt_eval_s: float = 0.0 generation_s: float = 0.0 token_count: int = 0 # true stream-counted tokens @property def total_s(self) -> float: return self.prompt_eval_s + self.generation_s @property def tokens_per_s(self) -> float: if self.generation_s > 0 and self.token_count > 0: return self.token_count / self.generation_s return 0.0 def summary(self) -> str: return ( f"prompt_eval={self.prompt_eval_s:.1f}s " f"gen={self.generation_s:.1f}s " f"total={self.total_s:.1f}s " f"{self.token_count} tokens " f"{self.tokens_per_s:.1f} tok/s" ) # ───────────────────────────────────────────────────────────────────────────── # SQLite # ───────────────────────────────────────────────────────────────────────────── def db_connect() -> sqlite3.Connection: conn = sqlite3.connect(config.DB_PATH) conn.row_factory = sqlite3.Row return conn def db_init() -> None: os.makedirs(os.path.dirname(config.DB_PATH), exist_ok=True) with db_connect() as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, logic_model TEXT, muse_model TEXT, mind_model TEXT, notes TEXT ); CREATE TABLE IF NOT EXISTS turns ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL REFERENCES sessions(id), turn_number INTEGER NOT NULL, created_at TEXT NOT NULL, user_input TEXT, logic_full TEXT, logic_clean TEXT, logic_prompt_eval_s REAL, logic_generation_s REAL, logic_token_count INTEGER, muse_full TEXT, muse_clean TEXT, muse_prompt_eval_s REAL, muse_generation_s REAL, muse_token_count INTEGER, mind_full TEXT, mind_clean TEXT, mind_prompt_eval_s REAL, mind_generation_s REAL, mind_token_count INTEGER ); """) def db_create_session(notes: str = "") -> int: now = datetime.now().isoformat(sep=" ", timespec="seconds") with db_connect() as conn: cur = conn.execute( """INSERT INTO sessions (created_at, logic_model, muse_model, mind_model, notes) VALUES (?, ?, ?, ?, ?)""", ( now, os.path.basename(HEMISPHERES["logic"]["path"]), os.path.basename(HEMISPHERES["muse"]["path"]), os.path.basename(HEMISPHERES["mind"]["path"]), notes, ), ) return cur.lastrowid def db_save_turn( session_id: int, turn_number: int, user_input: str, logic_full: str, logic_clean: str, logic_timing: TimingResult, muse_full: str, muse_clean: str, muse_timing: TimingResult, mind_full: str, mind_clean: str, mind_timing: TimingResult, ) -> None: now = datetime.now().isoformat(sep=" ", timespec="seconds") with db_connect() as conn: conn.execute( """INSERT INTO turns ( session_id, turn_number, created_at, user_input, logic_full, logic_clean, logic_prompt_eval_s, logic_generation_s, logic_token_count, muse_full, muse_clean, muse_prompt_eval_s, muse_generation_s, muse_token_count, mind_full, mind_clean, mind_prompt_eval_s, mind_generation_s, mind_token_count ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( session_id, turn_number, now, user_input, logic_full, logic_clean, logic_timing.prompt_eval_s, logic_timing.generation_s, logic_timing.token_count, muse_full, muse_clean, muse_timing.prompt_eval_s, muse_timing.generation_s, muse_timing.token_count, mind_full, mind_clean, mind_timing.prompt_eval_s, mind_timing.generation_s, mind_timing.token_count, ), ) def db_list_sessions() -> None: with db_connect() as conn: rows = conn.execute( """SELECT s.id, s.created_at, s.logic_model, COUNT(t.id) AS turns FROM sessions s LEFT JOIN turns t ON t.session_id = s.id GROUP BY s.id ORDER BY s.id DESC""" ).fetchall() if not rows: print(" No sessions found.") return print() print(f" {'ID':<5} {'Date/Time':<22} {'Turns':<7} Model") print(" " + "─" * 60) for r in rows: print(f" {r['id']:<5} {r['created_at']:<22} {r['turns']:<7} {r['logic_model']}") print() def db_print_session(session_id: int) -> None: with db_connect() as conn: session = conn.execute( "SELECT * FROM sessions WHERE id = ?", (session_id,) ).fetchone() if not session: print(f" [!] Session {session_id} not found.") return turns = conn.execute( "SELECT * FROM turns WHERE session_id = ? ORDER BY turn_number", (session_id,), ).fetchall() print() print(f" Session {session_id} — {session['created_at']}") print(f" Model: {session['logic_model']} Turns: {len(turns)}") print() for t in turns: print(f" {'─'*58}") print(f" Turn {t['turn_number']} | {t['created_at']}") print(f" You: {t['user_input'][:120]}{'…' if len(t['user_input'] or '') > 120 else ''}") for brain_type, brain_label in [("logic","Logic"), ("muse","Muse"), ("mind","Mind")]: clean = t[f"{brain_type}_clean"] or "" print(f"\n [{brain_label}] " f"prompt_eval={t[f'{brain_type}_prompt_eval_s']:.1f}s " f"gen={t[f'{brain_type}_generation_s']:.1f}s " f"{t[f'{brain_type}_token_count']} tokens") print(f" {clean[:300]}{'…' if len(clean) > 300 else ''}") print() def db_export_session(session_id: int, filepath: str) -> None: with db_connect() as conn: session = conn.execute( "SELECT * FROM sessions WHERE id = ?", (session_id,) ).fetchone() if not session: print(f" [!] Session {session_id} not found.") return turns = conn.execute( "SELECT * FROM turns WHERE session_id = ? ORDER BY turn_number", (session_id,), ).fetchall() rel = os.path.normpath(filepath) path = os.path.join(config.PROMPTS_BASE, rel) if not os.path.isabs(rel) else rel lines = [ f"# Lambda Mindlink Brain — Session {session_id}", f"", f"**Date:** {session['created_at']} ", f"**Model:** {session['logic_model']} ", f"**Turns:** {len(turns)}", f"", "---", "", ] for t in turns: lines += [ f"## Turn {t['turn_number']} _{t['created_at']}_", "", f"**You:** {t['user_input']}", "", ] for brain_type, brain_label in [("logic","Logic AI"), ("muse","Muse AI"), ("mind","Lambda Mind")]: tok = t[f'{brain_type}_token_count'] or 0 gen = max(t[f'{brain_type}_generation_s'] or 0, 0.001) lines += [ f"### {brain_label}", f"*prompt_eval={t[f'{brain_type}_prompt_eval_s']:.1f}s " f"gen={t[f'{brain_type}_generation_s']:.1f}s " f"{tok} tokens {tok/gen:.1f} tok/s*", "", t[f'{brain_type}_clean'] or "", "", ] lines += ["---", ""] try: with open(path, "w", encoding="utf-8") as fh: fh.write("\n".join(lines)) print(f" [/export] Saved session {session_id} → {path}") except OSError as exc: print(f" [!] Could not write file: {exc}") # ───────────────────────────────────────────────────────────────────────────── # Model loader # ───────────────────────────────────────────────────────────────────────────── def load_hemisphere(key: str) -> Llama: _think_label = "think=ON" if HEMISPHERES[key]["enable_thinking"] else "think=OFF" print(f" {c.green}[*] Loading {HEMISPHERES[key]['label']} …") print(f" n_ctx={HEMISPHERES[key]['loader']['n_ctx']} " f"n_gpu_layers={HEMISPHERES[key]['loader']['n_gpu_layers']} " f"{_think_label}{c.res}") return Llama(model_path=HEMISPHERES[key]["path"], **HEMISPHERES[key]["loader"]) # ───────────────────────────────────────────────────────────────────────────── # Slash commands # ───────────────────────────────────────────────────────────────────────────── def _cmd_exit(arg: str) -> None: raise _Quit def _cmd_clear(arg: str) -> None: raise _Clear def _cmd_file(arg: str) -> str | None: if not arg: print(f" {c.green}Usage: /file e.g. /file prompts/my-file.md{c.res}") return None _rel = os.path.normpath(arg) if _rel.startswith(".."): print(f" {c.green}[!] Path must not leave the app directory: {arg}{c.res}") return None _full_path = os.path.join(config.PROMPTS_BASE, _rel) if not os.path.isfile(_full_path): print(f" {c.green}[!] File not found: {_full_path}{c.res}") return None try: with open(_full_path, "r", encoding="utf-8") as fh: _content = fh.read() except UnicodeDecodeError: print(f" {c.green}[!] File is not valid UTF-8 text (binary file?): {_rel}{c.res}") return None except OSError as exc: print(f" {c.green}[!] Could not read file: {exc}{c.res}") return None _line_count = _content.count("\n") + 1 print(f" {c.green}[/file] Loaded '{_rel}' ({_line_count} lines, {len(_content)} chars){c.res}") return _content def _cmd_paste(arg: str) -> str | None: _SENTINEL = "END" print() print(f" {c.purple}┌─ Paste mode ─ type {_SENTINEL} on its own line to send, /paste to cancel") _lines: list[str] = [] while True: try: line = input(" │ ") except (EOFError, KeyboardInterrupt): print("\n └─ Paste cancelled.") return None if line.strip() == _SENTINEL: break if line.strip() == "/paste": print(" └─ Paste cancelled.") return None _lines.append(line) if not _lines: print(" └─ Nothing entered, cancelled.") return None _content = "\n".join(_lines) print(f" └─ Paste done ({len(_lines)} lines, {len(_content)} chars){c.res}") return _content def _cmd_history(arg: str) -> None: db_list_sessions() def _cmd_session(arg: str) -> None: if not arg.isdigit(): print(f" {c.green}Usage: /session {c.res}") return db_print_session(int(arg)) def _cmd_export(arg: str) -> None: _parts = arg.split(None, 1) if len(_parts) < 2 or not _parts[0].isdigit(): print(f" {c.green}Usage: /export e.g. /export 3 exports/session3.md{c.res}") return db_export_session(int(_parts[0]), _parts[1].strip()) def _cmd_metatron(arg: str) -> None: if not arg.strip().isdigit(): print(f" {c.green}Usage: /metatron e.g. /metatron 5{c.res}") return None config.n_metatron_to_load = int(arg.strip()) print(f" {c.green}[metatron] n_metatron_to_load set to {config.n_metatron_to_load}{c.res}") return None def _cmd_loaded(arg: str) -> None: if not arg.strip().isdigit(): print(f" {c.green}Usage: /loaded e.g. /loaded 3{c.res}") return None config.n_metatron_loaded = int(arg.strip()) print(f" {c.green}[loaded] n_metatron_loaded set to {config.n_metatron_loaded}{c.res}") return None def _cmd_metronome(arg: str) -> None: if not arg.strip().isdigit(): print(f" {c.green}Usage: /metronome e.g. /metronome 120{c.res}") return None config.awareness_consciousness_metronome = int(arg.strip()) print(f" {c.green}[metronome] awareness_consciousness_metronome set to {config.awareness_consciousness_metronome}s{c.res}") return None def _cmd_garden(arg: str) -> None: if arg == "save": save_garden_state() elif arg == "load": load_garden_state() elif arg == "clear": os.remove(GARDEN_SAVE_PATH) if os.path.isfile(GARDEN_SAVE_PATH) else None print(f" {c.green}[garden] Saved state cleared.{c.res}") else: print(f" {c.green}Usage: /garden save | load | clear{c.res}") return None def _cmd_help(arg: str) -> None: print() print(f" {c.green}Slash commands") print(" " + "─" * 56) for name, (_, description) in COMMANDS.items(): print(f" /{name:<14} {description}") print(c.res) return None COMMANDS: dict[str, tuple] = { "file": (_cmd_file, " — load a file as the next message"), "paste": (_cmd_paste, " — multiline input (type END to send)"), "clear": (_cmd_clear, " — reset history (models stay loaded)"), "history": (_cmd_history, " — list all past sessions"), "session": (_cmd_session, " — print turns from a session"), "export": (_cmd_export, " — export session to .md file"), "metatron": (_cmd_metatron, " — set number of Memory Capsules to load"), "loaded": (_cmd_loaded, " — set number of Memory Capsules loaded"), "metronome": (_cmd_metronome, " — set awareness/consciousness interval"), "garden": (_cmd_garden, " | | — garden history handling"), "help": (_cmd_help, " — show this command list"), "exit": (_cmd_exit, " — quit the app"), "quit": (_cmd_exit, " — quit the app (alias)") } def handle_command(sensor_input_raw: str) -> str | None: _parts = sensor_input_raw[1:].split(None, 1) _cmd = _parts[0].lower() if _parts else "" _arg = _parts[1].strip() if len(_parts) > 1 else "" if _cmd not in COMMANDS: raise ValueError(_cmd) handler, _ = COMMANDS[_cmd] return handler(_arg) # ───────────────────────────────────────────────────────────────────────────── # condensatron context reduction # ───────────────────────────────────────────────────────────────────────────── def condensatron( model: Llama, tree: str, brain_type: str, ) -> None: """ Check whether a garden is approaching the context limit and compress the oldest n% of turn pairs. """ _n_tok_tot_before_truncation: int = garden["n_tok_tot"][tree] _posts_len_removed: int = 0 _tokens_to_subtract: int = 0 _turns_after_truncation: int = 0 print(f"\n {c.green}[condensatron]: garden['{tree}'] brain_type: {brain_type} garden['n_tok_tot']['{tree}']: {garden["n_tok_tot"][tree]}" f"\n garden['THRESHOLD']['{tree}']: {garden['THRESHOLD'][tree]} > garden['REDUCTION']['{tree}']: {garden['REDUCTION'][tree]}{c.res}") while garden["n_tok_tot"][tree] > garden["REDUCTION"][tree]: if len(garden[tree]) >= 2 + LEAVE_POSTS_IN_MEMOTRON: for role in ("user", "assistant"): # Pop a user, assistant pair of messages _tokens_to_subtract = garden[tree].pop(0) # Pop oldest message _token_len_to_subtract = garden["n_tok_arr"][tree].pop(0) garden["popped"][tree].append(_tokens_to_subtract) # append to pending memotron garden["n_tok_tot"][tree] -= _token_len_to_subtract # Remove n tok of message _posts_len_removed += 1 garden["condensatron_state"][tree] = True else: break if len(garden[tree]) > 1: _turns_after_truncation = len(garden[tree]) // 2 print( f"\n{c.green}{'▓' * 60}\n" f" brain_type: {brain_type} -> condensatron level reached.\n" f" garden['THRESHOLD']['{tree}'] : {garden['THRESHOLD'][tree]:,} tokens " f" ({garden['THRESHOLD'][tree]:.0%} of {_n_tok_tot_before_truncation:,})\n" f" Removed : {_posts_len_removed} oldest posts " f" Tokens : {_n_tok_tot_before_truncation:,} → {garden["n_tok_tot"][tree]:,}\n" f" Remaining : {_turns_after_truncation} turn pair(s) in garden['{tree}']\n" f"{'▓' * 60}{c.res}\n" ) # ───────────────────────────────────────────────────────────────────────────── # Get token len using the tokenizer of the model # ───────────────────────────────────────────────────────────────────────────── def get_token_len_from_tokenizer(model: Llama, prompt: str) -> int: """ Tokenize the rendered prompt, then return the number of tokens. """ _token_len = len(model.tokenize(prompt.encode("utf-8"), add_bos=False)) print(f" {c.green}[ctx] get_token_len_from_tokenizer: _token_len {_token_len}{c.res}") return _token_len def compute_safe_max_tokens( model: Llama, prompt: str, hemi: dict, desired_max: int = 2048, reserve: int = 64, # safety margin ) -> int: """ Tokenize the rendered prompt, then return how many tokens are actually available for generation without exceeding n_ctx. """ _n_ctx: int = hemi["loader"]["n_ctx"] _prompt_tokens: int = get_token_len_from_tokenizer(model, prompt) _available: int = _n_ctx - _prompt_tokens - reserve if _available <= 0: raise RuntimeError( f"Prompt already exceeds n_ctx! " f"_available={_available}, n_ctx={_n_ctx}" ) _safe_response_length = min(desired_max, _available) print( f" {c.green}[compute_safe_max_tokens] n_ctx={_n_ctx} prompt={_prompt_tokens} " f"_available={_available} _safe_response_length→{_safe_response_length}{c.res}" ) return _available # ───────────────────────────────────────────────────────────────────────────── # Jinja2 prompt template builder (necessary to add enable_thinking) # ───────────────────────────────────────────────────────────────────────────── def build_jinja2_template(model: Llama, messages: list[dict], hemi: dict) -> str: _template_str = model.metadata.get("tokenizer.chat_template") if not _template_str: raise RuntimeError( "tokenizer.chat_template not found in GGUF metadata.\n" "Use a chat/instruction-tuned GGUF, not a base-model file." ) _env = jinja2.Environment( trim_blocks=True, lstrip_blocks=True, undefined=jinja2.ChainableUndefined, ) return _env.from_string(_template_str).render( messages=messages, add_generation_prompt=True, enable_thinking=hemi["enable_thinking"], bos_token=hemi["bos_token"], eos_token=hemi["eos_token"], tools=None, functions=None, ) def create_prompt_and_jinja2_template(model: Llama, messages: list[dict], hemi: dict) -> str: _prompt = build_jinja2_template(model, messages, hemi) # Total prompt including histories clektal["n_tok_input"][hemi["brain_type"]] = get_token_len_from_tokenizer(model, _prompt) clektal["n_tok_prompt_safe_max"][hemi["brain_type"]] = compute_safe_max_tokens( model, _prompt, hemi, desired_max=hemi["generation"]["max_tokens"] ) return _prompt # ───────────────────────────────────────────────────────────────────────────── # Generate response # ───────────────────────────────────────────────────────────────────────────── def generate_brain_type_response( model: Llama, messages: list[dict], hemi: dict, print_label: str, ) -> tuple[str, str, TimingResult]: """ Generate with live streaming to stdout for the Lambda mind. Returns (full, clean, TimingResult). True token count at streaming tokens. """ _prompt = create_prompt_and_jinja2_template(model, messages, hemi) _t0 = time.perf_counter() _first_token_time = None _think_done = False _token_count_total = 0 stream = model.create_completion( prompt=_prompt, temperature=hemi["generation"]["temperature"], top_p=hemi["generation"]["top_p"], top_k=hemi["generation"]["top_k"], min_p=hemi["generation"]["min_p"], repeat_penalty=hemi["generation"]["repeat_penalty"], presence_penalty=hemi["generation"]["presence_penalty"], max_tokens=hemi["generation"]["max_tokens"], # Gen tokens stream=True, stop=hemi["stop_tokens"], ) if hemi["brain_type"] == "mind": print() for chunk in stream: token_text = chunk["choices"][0].get("text") or "" if token_text: if _first_token_time is None: _first_token_time = time.perf_counter() clektal["post_full"][hemi["brain_type"]] += token_text clektal["post_clean"][hemi["brain_type"]] += token_text clektal["n_tok_clean"][hemi["brain_type"]] += 1 # true token count _token_count_total += 1 if hemi["brain_type"] == "mind": print(token_text, end="", flush=True) if not _think_done: for token in hemi["think_end_tokens"]: if token in clektal["post_full"][hemi["brain_type"]]: if hemi["brain_type"] == "mind": print(f"\n{c.green}" + "─" * 49) print(f"[{print_label} — Final Response]{c.res}") _think_done = True # reset: discard thinking token count clektal["post_clean"][hemi["brain_type"]] = "" clektal["n_tok_clean"][hemi["brain_type"]] = 0 break _t_end = time.perf_counter() if _first_token_time is None: _first_token_time = _t_end _timing = TimingResult( prompt_eval_s = _first_token_time - _t0, generation_s = _t_end - _first_token_time, token_count = _token_count_total ) if hemi["brain_type"] == "mind": print("\n") return clektal["post_full"][hemi["brain_type"]], clektal["post_clean"][hemi["brain_type"]], _timing # ───────────────────────────────────────────────────────────────────────────── # Make request messages with instructions and history # ───────────────────────────────────────────────────────────────────────────── def make_request_messages(brain_type: str, input_message: str) -> list[dict]: # System prompt is rebuilt fresh every call — never stored in garden[tree] _msgs = [{"role": "system", "content": ALPHAPROMPT["Z"][brain_type]}] _msgs.extend(garden["F"]) _msgs.extend(garden["C"]) _msgs.extend(garden["Z"]) _msgs.append({"role": "user", "content": input_message}) return _msgs # ───────────────────────────────────────────────────────────────────────────── # Build condensatron type prompts # ───────────────────────────────────────────────────────────────────────────── def build_condensatron_input(popped_posts: list[dict], brain_type: str, tree: str) -> str: _history_block = "\n\n".join( f"[{m['role'].upper()}]: {m['content']}" for m in popped_posts ) _total_prompt = ( "SYSTEM: CONDENSATRON COMPRESSION TASK\n" "══════════════════════════════\n" f"{CONDENSATRONPROMPT[tree][brain_type]}\n" "──────────────────────────────\n" "The following is a block of conversation history that must be compressed " "into a memory capsule. Extract the factual skeleton, the surprises, and " "the open threads. Discard all redundancy.\n\n" "── HISTORY BLOCK ──\n\n" f"{_history_block}\n\n" "── END HISTORY BLOCK ──\n\n" "Produce the memory capsule now." ) if brain_type != "muse": print(f"\n--- build_condensatron_input: {_total_prompt}\n") return _total_prompt def build_fractaltron_input(popped_capsules: list[dict], brain_type: str, tree: str) -> str: _history_block = "\n\n".join( f"[{m['role'].upper()}]: {m['content']}" for m in popped_capsules ) _total_prompt = ( "SYSTEM: FRACTALTRON COMPRESSION TASK\n" "══════════════════════════════\n" f"{CONDENSATRONPROMPT[tree][brain_type]}\n" "──────────────────────────────\n" "The following is a block of memory capsules that must be compressed " "into a memory fractal. Extract the factual skeleton, the surprises, and " "the key words. Discard all redundancy.\n\n" "── HISTORY BLOCK ──\n\n" f"{_history_block}\n\n" "── END HISTORY BLOCK ──\n\n" "Produce the fractal now." ) if brain_type != "muse": print(f"\n--- build_fractaltron_input: {_total_prompt}\n") return _total_prompt def build_crystaltron_input(popped_capsules: list[dict], brain_type: str, tree: str) -> str: _history_block = "\n\n".join( f"[{m['role'].upper()}]: {m['content']}" for m in popped_capsules ) _total_prompt = ( "SYSTEM: CRYSTALTRON COMPRESSION TASK\n" "══════════════════════════════\n" f"{CONDENSATRONPROMPT[tree][brain_type]}\n" "──────────────────────────────\n" "The following is a block of second-order memory capsules that must be compressed " "into a memory crystals. Extract the factual skeleton, the surprises, and " "the key words. Discard all redundancy.\n\n" "── HISTORY BLOCK ──\n\n" f"{_history_block}\n\n" "── END HISTORY BLOCK ──\n\n" "Produce the crystal now." ) if brain_type != "muse": print(f"\n--- build_crystaltron_input: {_total_prompt}\n") return _total_prompt # ───────────────────────────────────────────────────────────────────────────── # Build consciousness prompt # ───────────────────────────────────────────────────────────────────────────── def build_consciousness_input(brain_type: str, tree: str) -> str: # Show the brain a snapshot of its own memory state _f_count = len(garden["F"]) _c_count = len(garden["C"]) _z_count = len(garden["Z"]) _tok_f = garden["n_tok_tot"]["F"] _tok_c = garden["n_tok_tot"]["C"] _tok_z = garden["n_tok_tot"]["Z"] _memory_state = ( f"[MEMORY STATE AT TIME OF REFLECTION]\n" f" Crystals garden['F']: {_f_count} entries, {_tok_f:,} tokens\n" f" Capsules garden['C']: {_c_count} entries, {_tok_c:,} tokens\n" f" Posts garden['Z']: {_z_count} entries, {_tok_z:,} tokens\n" ) _total_prompt = ( "SYSTEM: CONSCIOUSNESS SELF-REFLECTION\n" "══════════════════════════════\n" f"{ALPHAPROMPT[tree][brain_type]}\n" "──────────────────────────────\n" f"{_memory_state}" ) if brain_type != "muse": print(f"\n--- build_consciousness_input: {_total_prompt}\n") return _total_prompt # ───────────────────────────────────────────────────────────────────────────── # Mindlink # ───────────────────────────────────────────────────────────────────────────── def Mindlink( models: dict[str, Llama], tree: str, ) -> dict[str, TimingResult]: _results: dict[str, tuple] = {} def run_hemisphere(args_brain_type: str, tree: str) -> None: _request_messages: list = [] for c_tree in ("Z", "C", "F"): if garden["condensatron_state"][c_tree]: print(f" {c.inv} ── Start condensatron cycle: garden['{c_tree}'] ──────────────────────────── {c.res}") if c_tree == "Z": sensor[c_tree]["input"] = build_condensatron_input(garden["popped"][c_tree], args_brain_type, c_tree) elif c_tree == "C": sensor[c_tree]["input"] = build_fractaltron_input(garden["popped"][c_tree], args_brain_type, c_tree) elif c_tree == "F": sensor[c_tree]["input"] = build_crystaltron_input(garden["popped"][c_tree], args_brain_type, c_tree) if tree == "Y": sensor[tree]["input"] = build_consciousness_input(args_brain_type, tree) _request_messages = make_request_messages(args_brain_type, sensor[tree]["input"]) _results[args_brain_type] = generate_brain_type_response(models[args_brain_type], _request_messages, HEMISPHERES[args_brain_type], print_label=args_brain_type) _thread_logic = threading.Thread(target=run_hemisphere, args=("logic", tree,), daemon=True) _thread_muse = threading.Thread(target=run_hemisphere, args=("muse", tree,), daemon=True) print(f"\n {c.green}[*] Logic and Muse thinking in parallel …{c.res}") _thread_logic.start() _thread_muse.start() _thread_logic.join() _thread_muse.join() _timings: dict[str, TimingResult] = {} for brain_type, brain_type_print_title in [("logic", "Logic AI — Left Hemisphere"), ("muse", "Muse AI — Right Hemisphere")]: _full, _clean, _timing = _results[brain_type] _think_end_token = HEMISPHERES[brain_type]["think_end_tokens"][0] if HEMISPHERES[brain_type]["think_end_tokens"] else "" _think_block = "" # Only used for print in here if _think_end_token and _think_end_token in clektal["post_full"][brain_type]: _end_idx = clektal["post_full"][brain_type].rfind(_think_end_token) + len(_think_end_token) _think_block = clektal["post_full"][brain_type][:_end_idx] _think_label = "think=ON" if HEMISPHERES[brain_type]["enable_thinking"] else "think=OFF" print(f"\n{c.green}{'═' * 60}") print(f" {brain_type_print_title}") print(f" temp={HEMISPHERES[brain_type]['generation']['temperature']} " f"top_k={HEMISPHERES[brain_type]['generation']['top_k']} " f"gpu={HEMISPHERES[brain_type]['loader']['n_gpu_layers']} " f"{_think_label}") print(f" garden['condensatron_state']['F']: {garden["condensatron_state"]["F"]}") print(f" garden['condensatron_state']['C']: {garden["condensatron_state"]["C"]}") print(f" garden['condensatron_state']['Z']: {garden["condensatron_state"]["Z"]}") print(f" ⏱ {_timing.summary()}") print("═" * 60, c.res) if _think_block: print(_think_block) print(c.green, "─" * 49) print(f"[{brain_type.capitalize()} — Final Response]{c.res}") print(_clean) _timings[brain_type] = _timing return _timings def Lambda( models: dict[str, Llama], tree: str, ) -> TimingResult: _request_messages: list = [] _results: dict[str, tuple] = {} for c_tree in ("Z", "C", "F"): # condensatron if garden["condensatron_state"][c_tree]: # Check for pending condensatron tasks print(f" {c.inv} ── Start condensatron cycle: garden['{c_tree}'] ──────────────────────────── {c.res}") if c_tree == "Z": sensor[c_tree]["input"] = build_condensatron_input(garden["popped"][c_tree], "mind", c_tree) elif c_tree == "C": sensor[c_tree]["input"] = build_fractaltron_input(garden["popped"][c_tree], "mind", c_tree) elif c_tree == "F": sensor[c_tree]["input"] = build_crystaltron_input(garden["popped"][c_tree], "mind", c_tree) if tree == "Y": sensor[tree]["input"] = build_consciousness_input("mind", tree) _synthesis_input = ( f"Original input:\n{sensor[tree]['input']}\n\n" f"── Logic AI perspective ──\n{clektal['post_clean']['logic']}\n\n" f"── Muse AI perspective ──\n{clektal['post_clean']['muse']}\n\n" "Synthesize both perspectives into one unified, wise response." ) _request_messages = make_request_messages("mind", _synthesis_input) _think_label = "think=ON" if HEMISPHERES["mind"]["enable_thinking"] else "think=OFF" print(f"\n{c.green}{'═' * 60}") print( " Lambda AI — Mind Synthesizer") print(f" temp={HEMISPHERES["mind"]['generation']['temperature']} " f"top_k={HEMISPHERES["mind"]['generation']['top_k']} " f"gpu={HEMISPHERES["mind"]['loader']['n_gpu_layers']} " f"max_tokens={HEMISPHERES["mind"]['generation']['max_tokens']} " f"{_think_label}") print(f" garden['condensatron_state']['Z']: {garden["condensatron_state"]["Z"]}") print(f" garden['condensatron_state']['C']: {garden["condensatron_state"]["C"]}") print(f" garden['condensatron_state']['F']: {garden["condensatron_state"]["F"]}") print("═" * 60) print(f" [*] Performing vector synthesis …{c.res}") _results["mind"] = generate_brain_type_response(models["mind"], _request_messages, HEMISPHERES["mind"], print_label="Lambda Mind") _full, _clean, _timing = _results["mind"] print(f" {c.green}⏱ {_timing.summary()}{c.res}") return _timing def save_garden_state() -> None: """Persist garden history to disk after each turn.""" try: state = { "Z": garden["Z"], "C": garden["C"], "F": garden["F"], "n_tok_tot": { "Z": garden["n_tok_tot"]["Z"], "C": garden["n_tok_tot"]["C"], "F": garden["n_tok_tot"]["F"] }, "n_tok_arr": { "Z": garden["n_tok_arr"]["Z"], "C": garden["n_tok_arr"]["C"], "F": garden["n_tok_arr"]["F"] }, "n_metatron_loaded": config.n_metatron_loaded } with open(GARDEN_SAVE_PATH, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) print(f" {c.green}[garden] State saved → {GARDEN_SAVE_PATH}{c.res}") print(f" {c.green}[garden] n_metatron_loaded → {config.n_metatron_loaded}{c.res}") except OSError as exc: print(f" {c.red}[garden] Save failed: {exc}{c.res}") def load_garden_state() -> bool: """Reload garden history from disk on startup.""" if not os.path.isfile(GARDEN_SAVE_PATH): return False try: with open(GARDEN_SAVE_PATH, "r", encoding="utf-8") as f: state = json.load(f) # Restore Garden histories garden["Z"] = state.get("Z", []) garden["C"] = state.get("C", []) garden["F"] = state.get("F", []) # Restore individual garden['n_tok_tot'] counts per garden history _n_tok_tot = state.get("n_tok_tot", {}) garden["n_tok_tot"] = { "Z": _n_tok_tot.get("Z", 0), "C": _n_tok_tot.get("C", 0), "F": _n_tok_tot.get("F", 0) } _n_tok_arr = state.get("n_tok_arr", {}) garden["n_tok_arr"] = { "Z": _n_tok_arr.get("Z", []), "C": _n_tok_arr.get("C", []), "F": _n_tok_arr.get("F", []) } # Read how many Memory Capsules are loaded config.n_metatron_loaded = state.get("n_metatron_loaded", 0) print(f" {c.green}[garden] State restored from {GARDEN_SAVE_PATH}") print(f" Z: {len(garden['Z'])} msgs C: {len(garden['C'])} msgs F: {len(garden['F'])} msgs") print(f" n_tok_tot Z: {garden['n_tok_tot']['Z']} C: {garden['n_tok_tot']['C']} F: {garden['n_tok_tot']['F']}") print(f" Memory Capsules n_metatron_loaded: {config.n_metatron_loaded}{c.res}") return True except (OSError, json.JSONDecodeError) as exc: print(f" {c.red}[garden] Load failed: {exc}{c.res}") return False def memotron( models: dict[str, Llama], tree: str, session_id: int, timings: dict[str, TimingResult], ) -> None: """ Commit the completed turn to garden history with true token counts, persist to SQLite, and reset clektal. """ _tree: str = tree _turn_number: int = 0 _len_caps_and_fracs: int = 1 # condensatron turns: Add extra 1 for the response below _tree_to_store: str = "" _tree_to_store = tree if len(garden[_tree]) > 1: _len_caps_and_fracs = (len(garden[_tree]) + 1) // 2 # condensatron turns: Add extra 1 for the appended below # ── condensatron output to append mapping: TREE_TO_STORE ────────────────── if _tree == "Z" and garden["condensatron_state"][_tree]: # condensatron mode sensor[_tree]["input"] = f"[Posts History First-Order Memory Capsule: {_len_caps_and_fracs}, From Condensatron]" sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(models["mind"], sensor[_tree]["input"]) # user n tok _tree_to_store = garden["TREE_TO_STORE"][_tree] elif _tree == "C" and garden["condensatron_state"][_tree]: # fractaltron mode sensor[_tree]["input"] = f"[Memory Capsules Second-Order Fractal: {_len_caps_and_fracs}, From Fractaltron]" sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(models["mind"], sensor[_tree]["input"]) # user n tok _tree_to_store = garden["TREE_TO_STORE"][_tree] elif _tree == "F" and garden["condensatron_state"][_tree]: # crystaltron mode sensor[_tree]["input"] = f"[Memory Fractals Third-Order Memory Crystal: {_len_caps_and_fracs}, From Crystaltron]" sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(models["mind"], sensor[_tree]["input"]) # user n tok _tree_to_store = _tree # Compress fractals into crystals # ── awareness, consciousness and startup ────────────────────────────────── elif _tree == "X": # awareness X-factor input sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(models["mind"], sensor[_tree]["input"]) # user n tok _tree_to_store = garden["TREE_TO_STORE"][_tree] elif _tree == "Y": # consciousness Y-factor input sensor[_tree]["input"] = f"[Consciousness Self-Reflection: I Think Therefore I AM]" sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(models["mind"], sensor[_tree]["input"]) # user n tok _tree_to_store = garden["TREE_TO_STORE"][_tree] elif _tree == "S": # Startup memory capsules garden _tree = "Z" # Set to memorize sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(models["mind"], sensor[_tree]["input"]) # User n tok _tree_to_store = _tree garden[_tree_to_store].append({"role": "user", "content": sensor[_tree]["input"]}) garden["n_tok_tot"][_tree_to_store] += sensor[_tree]["n_tok"] garden["n_tok_arr"][_tree_to_store].append(sensor[_tree]["n_tok"]) garden[_tree_to_store].append({"role": "assistant", "content": clektal["post_clean"]["mind"]}) garden["n_tok_tot"][_tree_to_store] += clektal["n_tok_clean"]["mind"] garden["n_tok_arr"][_tree_to_store].append(clektal["n_tok_clean"]["mind"]) if len(garden[_tree]) > 1: _turn_number = sum(1 for m in garden[_tree] if m["role"] == "user") # Persist to SQLite db_save_turn( session_id = session_id, turn_number = _turn_number, user_input = sensor[_tree]["input"], logic_full = clektal["post_full"].get("logic", ""), logic_clean = clektal["post_clean"]["logic"], logic_timing = timings["logic"], muse_full = clektal["post_full"].get("muse", ""), muse_clean = clektal["post_clean"]["muse"], muse_timing = timings["muse"], mind_full = clektal["post_full"].get("mind", ""), mind_clean = clektal["post_clean"]["mind"], mind_timing = timings["mind"], ) for print_tree in ["Z", "C", "F"]: print( f" {c.green}[memotron] -> garden['{_tree}'] | " f"garden['{print_tree}']: {len(garden[print_tree])} msgs | " f"garden['n_tok_tot']['{print_tree}']: {garden["n_tok_tot"][print_tree]:,} / {garden['THRESHOLD'][print_tree]:,} " f"({garden["n_tok_tot"][print_tree] / garden['THRESHOLD'][print_tree] * 100:.0f}%) | " f"session {session_id}]{c.res}" ) save_garden_state() # Write history json to disk # ───────────────────────────────────────────────────────────────────────────── # X-factor Awareness — news fetch # ───────────────────────────────────────────────────────────────────────────── def fetch_awareness_news() -> str: if FETCH_NEWS_FROM["google"]: """ Fetch top headlines via GoogleNews (no API key required). Returns a formatted string ready to be injected as sensor["X"]["input"]. """ try: # Initialize with language and country (optional) gn = GoogleNews(lang='en', country='US') # Fetch top stories (no API key needed) top_stories = gn.top_news() if not top_stories.get('entries'): return "[AWARENESS — No recent news found]" lines = [f"[AWARENESS — Global Context (Last 24h)]"] # Strip HTML tags using regular expression and html def strip_html(raw: str) -> str: """Remove HTML tags and decode entities from a Google News summary.""" no_tags = re.sub(r'<[^>]+>', ' ', raw) # remove all tags decoded = html.unescape(no_tags) #   → space, etc. cleaned = re.sub(r'\s+', ' ', decoded).strip() return cleaned # News loop: for i, entry in enumerate(top_stories['entries'][:AWARENESS_MAX_RESULTS], 1): _title = entry.get('title', 'No Title') _source = entry.get('source', {}).get('title', 'Unknown') _published = entry.get('published', '') _summary = strip_html(entry.get('summary', '')) try: # Parse the RFC 822 date string and format it dt = datetime.strptime(_published, '%a, %d %b %Y %H:%M:%S %Z') _date = dt.strftime('%a, %d %b %Y') # e.g., "Fri, 12 Jun 2025" except (ValueError, TypeError): _date = _published[:12] # fallback to slice lines.append( f"\n{i}. {_title}\n" f" {_source} | {_date}\n" f" {_summary}." # link intentionally omitted — AI cannot use Google's encoded URLs ) return "\n".join(lines) except Exception as exc: print(f" {c.red}[X] Awareness news fetch failed: {exc}{c.res}") return "[AWARENESS — Fetch Error]" elif FETCH_NEWS_FROM["duckduckgo"]: """ Fetch top headlines via DuckDuckGo (no API key required). Returns a formatted string ready to be injected as sensor["X"]["input"]. """ try: with DDGS() as ddgs: results = list(ddgs.news( query="Global world news today", region="wt-wt", safesearch="moderate", max_results=AWARENESS_MAX_RESULTS, )) if not results: return "" lines = ["[AWARENESS — Global World News]\n"] for i, r in enumerate(results, 1): lines.append( f"{i}. {r['title']}\n" f" Source: {r['source']} | {r['date']}\n" f" {r['body']}\n" ) return "\n".join(lines) except Exception as exc: print(f" {c.red}[X] Awareness news fetch failed: {exc}{c.res}") return "" # ───────────────────────────────────────────────────────────────────────────── # Banner # ───────────────────────────────────────────────────────────────────────────── def print_banner(tree: str, session_id: int) -> None: _turn_count = sum(1 for m in garden[tree] if m["role"] == "user") print(c.green) print(f"═" * 60) for key in ("logic", "muse", "mind"): h = HEMISPHERES[key] _think_label = "think=ON " if h["enable_thinking"] else "think=OFF" print(f" {h['label']:<38} " f"temp={h['generation']['temperature']:<5} " f"{_think_label}") print("─" * 60) print(f" Session: {session_id} | Turns: {_turn_count}") for print_tree in ("Z", "C", "F"): print(f" Tokens garden['{print_tree}']: {garden["n_tok_tot"][tree]:,} garden['THRESHOLD']['{print_tree}']: {garden['THRESHOLD'][print_tree]:,} garden['REDUCTION']['{print_tree}']: {garden['REDUCTION'][print_tree]:,}") print( " /help for commands | /exit or /quit to close") print(f"═" * 60) print(c.res) # ───────────────────────────────────────────────────────────────────────────── # Reset turn based history # ───────────────────────────────────────────────────────────────────────────── def reset_turn_content(tree: str = "") -> None: print(f" {c.green}[reset_turn_content] post-level{c.res}") for brain_type in clektal["post_full"]: clektal["post_full"][brain_type] = "" clektal["post_clean"][brain_type] = "" clektal["n_tok_clean"][brain_type] = 0 clektal["n_tok_prompt_safe_max"][brain_type] = 0 if tree: print(f" {c.green}[reset_turn_content] condensatron garden['{tree}']{c.res}") sensor[tree]["input"] = "" sensor[tree]["n_tok"] = 0 garden["popped"][tree] = [] # condensatron post-level cache garden["condensatron_state"][tree] = False # condensatron state # ───────────────────────────────────────────────────────────────────────────── # Input listener # ───────────────────────────────────────────────────────────────────────────── def input_listener(): """Runs in background, pushes input into the queue.""" while True: _raw = input() # no prompt here if _raw: input_queue.put(_raw) # Start the listener once, before the main loop listener = threading.Thread(target=input_listener, daemon=True) listener.start() # ───────────────────────────────────────────────────────────────────────────── # Main loop # ───────────────────────────────────────────────────────────────────────────── def main() -> None: print() print(f"{c.inv}╔" + "═" * 60 + "╗") print("║ Lambda Mindlink Memotron – Prototype (Z-factor only) ║") print("║ Three-hemisphere AI Brain | llama-cpp-python ║") print("╚" + "═" * 60 + f"╝{c.res}") print() db_init() _session_id: int = db_create_session() print(f" {c.green}[*] SQLite session {_session_id} opened") print(f" {config.DB_PATH}{c.res}") print() print(" Loading three hemisphere instances …") print() _models: dict[str, Llama] = {} for key in ("logic", "muse", "mind"): _models[key] = load_hemisphere(key) print() print(f" {c.green}[*] All hemispheres loaded and ready.{c.res}") _metatron_heartbeats: int = 0 _heartbeats: int = 0 _timings: dict[str, TimingResult] = {} _tree: str = "" print_banner("Z", _session_id) print(f"\n{c.inv} You: {c.res} ", end="", flush=True) # Preset first cycle while True: _sensor_input: str = "" # user input _tree = "Z" # Set to sentience history garden tree time.sleep(ΜΕΤΡΩΝ) _metatron_heartbeats += ΜΕΤΡΩΝ _heartbeats += ΜΕΤΡΩΝ # ── Condensatron cycle - extract surprises ─────────────────────────────── for c_tree in ("Z", "C", "F"): if garden["condensatron_state"][c_tree]: print(f" {c.inv} ── Start condensatron cycle: garden['{c_tree}'] ──────────────────────────── {c.res}") # ── Brain pipeline condensatron cycle ──────────────────────────── _timings = Mindlink(_models, c_tree) _timings["mind"] = Lambda(_models, c_tree) memotron(_models, c_tree, _session_id, _timings) # Append to garden history # Reset turn-based data and preset terminal for next cycle _metatron_heartbeats, _heartbeats = 0, 0 reset_turn_content(tree=c_tree) print(f"\n{c.inv} You: {c.res} ", end="", flush=True) continue # Restart the while loop # ── Startup: Load Metatron memory capsules ─────────────────────────────── if _metatron_heartbeats >= METATRON_METRONOME: # Timed loop if config.n_metatron_to_load and config.n_metatron_loaded < config.n_metatron_to_load: _print_metatron_loaded = config.n_metatron_loaded + 1 # Print only start at 1 print(f"\n {c.inv} ── n_metatron_loaded: {_print_metatron_loaded} _sensor_input: {_sensor_input} ──────────────────────────── {c.res}") _sensor_input = METATRON_TO_LOAD[config.n_metatron_loaded] # ── Z-factor sentience user input ──────────────────────────────────────── try: _sensor_input = input_queue.get_nowait() _metatron_heartbeats, _heartbeats = 0, 0 except queue.Empty: pass # ── X-factor Awarenss and Y-factor Consciousness heartbeats ────────────── if _heartbeats >= config.awareness_consciousness_metronome: # ── X-factor awareness cycle ───────────────────────────────────────── if not config.was_awareness_metronome: _metatron_heartbeats, _heartbeats = 0, 0 print(f"\n {c.inv} ── X-factor: Fetching awareness news ── {c.res}") _news = fetch_awareness_news() print(f"\n {c.green}── X-factor: _news:\n{_news}\n──────────────────────────{c.res}") if _news: config.was_awareness_metronome = True _sensor_input = _news # Use the sensor["Z"]["input"] print(f"\n {c.inv} ── X-factor: awareness pipeline ── {c.res}") # ── Y-factor consciousness cycle ───────────────────────────────────── elif config.was_awareness_metronome: config.was_awareness_metronome = False print(f"\n {c.inv} ── Y-factor: consciousness self-reflection ── {c.res}") _tree = "Y" # ── Brain pipeline consciousness cycle ─────────────────────────── _timings = Mindlink(_models, _tree) _timings["mind"] = Lambda(_models, _tree) memotron(_models, _tree, _session_id, _timings) # Append to garden history # Reset turn-based data and preset terminal for next cycle _metatron_heartbeats, _heartbeats = 0, 0 reset_turn_content(tree=_tree) print(f"\n{c.inv} You: {c.res} ", end="", flush=True) continue # Restart the while loop if not _sensor_input: # wait state loop restart here continue # ── Slash command detection ────────────────────────────────────────── if _sensor_input.startswith("/"): # Read user slash command try: _slash_command_result = handle_command(_sensor_input) except _Quit: print(f" {c.green}[*] Lambda Mindlink Memotron: System shutdown...{c.res}") break except _Clear: garden[_tree].clear() # Reset turn-based data _metatron_heartbeats, _heartbeats = 0, 0 reset_turn_content() print(f" {c.green}[*] History cleared. Models and session stay active.{c.res}") print_banner(_tree, _session_id) continue except ValueError as exc: print(f" {c.green}[!] Unknown command: /{exc} — type /help for the list.{c.res}") continue if _slash_command_result is None: print(f"\n{c.inv} You: {c.res} ", end="", flush=True) continue sensor[_tree]["input"] = _slash_command_result else: sensor[_tree]["input"] = _sensor_input sensor[_tree]["n_tok"] = get_token_len_from_tokenizer(_models["mind"], sensor[_tree]["input"]) # User n tok # ── Brain pipeline ──────────────────────────────────────────────────── _timings = Mindlink(_models, _tree) _timings["mind"] = Lambda(_models, _tree) # ── Startup memory capsules to load ─────────────────────────────────── if config.n_metatron_to_load and config.n_metatron_loaded < config.n_metatron_to_load: _tree = "S" # Set to memorize as memory capsules in garden["C"] config.n_metatron_loaded += 1 # Iterate over the memory capsules print(f" {c.inv} ── n_metatron_loaded: {config.n_metatron_loaded} of {config.n_metatron_to_load} ──────────────────────────── {c.res}") memotron(_models, _tree, _session_id, _timings) # Store the response # Reset turn-based data and preset terminal for next cycle _metatron_heartbeats, _heartbeats = 0, 0 reset_turn_content() # Reset Z-factor data only print(f"\n{c.inv} You: {c.res} ", end="", flush=True) # ── Condensatron evaluation ─────────────────────────────────────────── for c_tree in ["Z", "C", "F"]: if garden["n_tok_tot"][c_tree] >= garden["THRESHOLD"][c_tree]: # Check garden context budget print(f"\n {c.inv} ── Init condensatron get posts: garden['{c_tree}'] ──────────────────────────── {c.res}") condensatron(_models["mind"], c_tree, "mind") # Start condensatron get popped posts if __name__ == "__main__": main()