#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 8i / V.A.S.K. — Total Absolution Functional DNA Organism ========================================================= Single-file Windows/Linux desktop organism for Magnus Randa / AllAlong&AfterAll. Doctrine implemented: - DNA is NOT semantic; it is a deterministic functional grammar. - Every data stage must either roundtrip to the exact original bytes or be rejected. - Do not cut, flatten, or discard organs; preserve, ledger, verify, and expose them. - .zip is a repository shelf; .sd0 is the sealed verified object. - SEND BOOK exports expanded readable super-column DNA from repository contents. - Interactor Force controls optional online/API/model aides; OFF by default. - No arbitrary downloaded code execution; external code is text DNA until audited. This file is intentionally self-contained. Optional modules are imported lazily. It is research/prototype software: compression/encoding/inspection, not cryptographic secrecy. """ from __future__ import annotations import argparse import base64 import binascii import bz2 import dataclasses import datetime as _dt import gzip import hashlib import io import json import lzma import math import os import platform import queue import random import re import shutil import socket import string import struct import subprocess import sys import tarfile import tempfile import threading import time import traceback import urllib.parse import urllib.request import zipfile import zlib from collections import Counter, deque from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple # Optional accelerators are intentionally lazy/fault-tolerant. try: import zstandard as _zstd # type: ignore except Exception: _zstd = None try: import psutil as _psutil # type: ignore except Exception: _psutil = None VERSION = "8i-Total-Absolution-Functional-DNA-2.0.0" MAGIC = b"E8I-SDO-6\n" LEGACY_MAGICS = (b"E8I-SDO-5\n", b"E8I-GEOM-SDO-3\n") SPACE_VIS = "␠" ESC = "~" BOUNDARY = "~|" DICT_MARK = "¤" REF_MARK = "↩" DNA_IS_NOT_SEMANTIC = True DNA_IS_FUNCTIONAL_GRAMMAR = True EVERY_DNA_TOKEN_REQUIRES_A_ROLE = True EVERY_PAYLOAD_ROUTE_REQUIRES_ROUNDTRIP = True DEFAULT_VEIL_SECONDS = 600 LARGE_VISUAL_LIMIT = 2_000_000 DEFAULT_ROOT_NAME = "AfterAllAlong" # Five-strand inventory. These are alphabets/reservoirs. They are not meaning-bearing. STRANDS: Dict[str, str] = { "A": "£{?@)._¤+;†‡‰‱′″‴‹›‼‽⁂⁃⁅⁆⁇⁈⁉⁊⁋⁌⁍⁎⁏⁐⁑⁒⁓⁔⁕⁖⁗⁘⁙⁚⁛⁜⁝⁞₱₲₳₴₵₶₷₸₹₺₻₼⅓⅔⅕⅖⅗⅘⅙⅚⅛⅜⅝⅞⅟αβγδεζηθικλμνξοπρςστυφχψωΓΔΘΛΞΠΣΦΨΩ■□▢▣▤▥▦▧▨▩▲△▶▷▼▽◀◁◆◇◈◉○◎●◐◑◒◓◔◕", "T": "´*#|€:/>`-⌀⌁⌂⌃⌄⌅⌆⌇⌈⌉⌊⌋⌌⌍⌎⌏⌐⌑⌒⌓⌔⌕⌖⌗⌘⌙⌚⌛⌜⌝⌞⌟⌠⌡⌢⌣⌤⌥⌦⌧⌨〈〉⌫⌬⌭⌮⌯⌰⌱⌲⌳⌴⌵⌶⌷⌸⌹⌺⌻⌼⌽⌾⌿←↑→↓↔↕↖↗↘↙⇐⇑⇒⇓⇔⇕⇖⇗⇘⇙", "C": "^\"~<¿}°¨─━│┃┄┅┆┇┈┉┊┋┌┍┎┏┐┑┒┓└┕┖┗┘┙┚┛├┝┞┟┠┡┢┣┤┥┦┧┨┩┪┫┬┭┮┯┰┱┲┳┴┵┶┷┸┹┺┻┼┽┾┿╀╁╂╃╄╅╆╇╈╉╊╋╌╍╎╏═║╒╓╔╕╖╗╘╙╚╛╜╝╞╟╠╡╢╣╤╥╦╧╨╩╪╫╬", "G": "%]§(¡![,&=$АБВГДЕЖЗИЙКЛМНОПРСТУФХЦЧШЩЪЫЬЭЮЯабвгдежзийклмнопрстуфхцчшщъыьэюя∀∂∃∅∆∇∈∉∋∏∑−∕∗∘√∝∞∟∠∡∢∣∤∥∧⊕⊗⊘⊙⊚⊛⊜⊝⊞⊟⊠⊡", "Z": "✓✗✘✙✚✛✜✝✞✟✠✡✢✣✤✥✦✧★✩✪✫✬✭✮✯✰✱✲✳✴✵✶✷✸✹✺✻✼✽✾✿❀❁❂❃❄❅❆❇❈❉❊❋❍❏❐❑❒❖❘❙❚❛❜❝❞➔➘➙➚➛➜➝➞➡➢➣➤➥➦➧➨➩➪➫➬➭➮➯➰➱➲➳➴➵➶➷➸➹➺➻➼➽➾➿", } BASKET_POWERS = ["3¹", "3²", "3⁰", "3³"] BASKET_POLARITIES = ["+", "±", "-", "∓"] BASKET_ROLE = { "+": "Basket Positive / forward payload lane", "±": "Basket Negative / mirror payload lane", "-": "Basket Anti Positive / inverse payload lane", "∓": "Basket Anti Negative / anti-mirror payload lane", } BRACKET_REGISTER = { "()": "control, counts, local comments, optional metadata", "[]": "manifest, index, row and stage registers", "{}": "payload and basket interiors", "<>": "proof, hash, checksum, verification capsule", "°{...|•": "cluster roof / opening wall", "•|...|•": "cluster continuation wall", "o{...|•": "cluster floor / closing wall", "Y...Y/X...X/Z...Z": "axis fence / rectangular shell boundary", } RESERVED_MAP = { "~": "~t", "*": "~s", "?": "~q", "¤": "~c", "↩": "~r", "=": "~e", "\n": "~n", "\r": "~d", } UNRESERVED_MAP = {v: k for k, v in RESERVED_MAP.items()} MODEL_TARGETS = [ {"name": "Qwen2.5-Coder-0.5B/1.5B-Instruct", "role": "light local coding aide", "load": "small"}, {"name": "Qwen2.5-Coder-7B-Instruct GGUF", "role": "local coder target", "load": "medium"}, {"name": "DeepSeek-Coder-V2-Lite-Instruct", "role": "local/remote coding aide", "load": "medium"}, {"name": "Kimi K2.5", "role": "capability bar, not default local load", "load": "large"}, {"name": "GLM-5.1", "role": "capability bar, not default local load", "load": "large"}, ] # --------------------------------------------------------------------------- # Utility # --------------------------------------------------------------------------- def utc_now() -> str: return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds") def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def safe_name(name: str, limit: int = 100) -> str: name = name.replace("\\", "/").split("/")[-1] allowed = f"-_.() {string.ascii_letters}{string.digits}" cleaned = "".join(c if c in allowed else "_" for c in name).strip(" .") return (cleaned or "unnamed")[:limit] def atomic_write(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(path.name + f".tmp-{os.getpid()}-{random.randint(0, 999999):06d}") with open(tmp, "wb") as f: f.write(data) f.flush() try: os.fsync(f.fileno()) except Exception: pass os.replace(tmp, path) def entropy_bits_per_byte(data: bytes) -> float: if not data: return 0.0 c = Counter(data) n = len(data) return -sum((v / n) * math.log2(v / n) for v in c.values()) def detect_magic(name: str, data: bytes) -> str: ext = Path(name).suffix.lower().lstrip(".") if data.startswith(b"7z\xbc\xaf\x27\x1c"): return "7z" if data.startswith(b"PK\x03\x04"): return "zip" if data.startswith(b"\x1f\x8b"): return "gzip" if data.startswith(b"%PDF"): return "pdf" if data.startswith(b"\x89PNG"): return "png" if data.startswith(b"\xff\xd8\xff"): return "jpeg" if data[:4] in (b"II*\x00", b"MM\x00*"): return "tiff" if ext: return ext if b"\x00" in data[:4096]: return "bin" try: data[:4096].decode("utf-8") return "text" except Exception: return "bin" def is_text_like(data: bytes) -> bool: if not data: return True sample = data[:8192] if b"\x00" in sample: return False printable = sum((32 <= b <= 126) or b in (9, 10, 13) for b in sample) return printable / max(1, len(sample)) > 0.80 def read_json(path: Path, default: Any) -> Any: if not path.exists(): return default try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return default def write_json(path: Path, data: Any) -> None: atomic_write(path, json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")) # --------------------------------------------------------------------------- # Paths, config, organ registry # --------------------------------------------------------------------------- @dataclasses.dataclass class E8iPaths: root: Path repository: Path relativory: Path exportory: Path dna: Path logs: Path queue: Path models: Path tools: Path backups: Path lrn: Path session: Path file_arsenal: Path ai_friends: Path repository_zip: Path exportory_zip: Path relativory_zip: Path lrn_zip: Path config: Path organ_registry: Path chat_log: Path state: Path @classmethod def resolve(cls, root: Optional[str] = None) -> "E8iPaths": base = Path(root).expanduser().resolve() if root else Path.cwd().resolve() # D:\package friendly: if user extracts package, use that folder directly. r = base dirs = { "repository": r / "REPOSITORY", "relativory": r / "RELATIVORY", "exportory": r / "EXPORTORY", "dna": r / "DNA", "logs": r / "logs", "queue": r / "queue", "models": r / "models", "tools": r / "tools", "backups": r / "backups", "lrn": r / ".lrn", "session": r / "SESSION_WISHLIST", "file_arsenal": r / "file_arsenal", "ai_friends": r / "ai_friends", } for d in dirs.values(): d.mkdir(parents=True, exist_ok=True) p = cls( root=r, repository=dirs["repository"], relativory=dirs["relativory"], exportory=dirs["exportory"], dna=dirs["dna"], logs=dirs["logs"], queue=dirs["queue"], models=dirs["models"], tools=dirs["tools"], backups=dirs["backups"], lrn=dirs["lrn"], session=dirs["session"], file_arsenal=dirs["file_arsenal"], ai_friends=dirs["ai_friends"], repository_zip=dirs["repository"] / "REPOSITORY.zip", exportory_zip=dirs["exportory"] / "EXPORTORY.zip", relativory_zip=dirs["relativory"] / "RELATIVORY.zip", lrn_zip=dirs["lrn"] / "lrn.zip", config=r / "8i_config.json", organ_registry=r / "organ_registry.json", chat_log=dirs["repository"] / "e8i_dialogue_log.txt", state=r / "cognitive_state.json", ) for z in (p.repository_zip, p.exportory_zip, p.relativory_zip, p.lrn_zip): if not z.exists(): with zipfile.ZipFile(z, "w", compression=zipfile.ZIP_DEFLATED): pass return p DEFAULT_CONFIG: Dict[str, Any] = { "version": 2, "doctrine": { "dna_is_not_semantic": True, "functional_grammar_only": True, "every_stage_roundtrip_required": True, "external_code_text_dna_until_audited": True, "self_rewrite_default": False, }, "internet_force": False, "ollama": {"enabled": True, "url": "http://127.0.0.1:11434/api/generate", "model": "qwen2.5-coder:1.5b", "timeout": 20}, "telegram": {"enabled": False, "token_env": "TELEGRAM_BOT_TOKEN", "chat_id_env": "TELEGRAM_CHAT_ID"}, "voice": {"enabled": False, "engine": "pyttsx3", "rate": 165}, "gui": {"window_width": 1180, "window_height": 860, "veil_seconds": DEFAULT_VEIL_SECONDS}, "compression": {"large_visual_limit": LARGE_VISUAL_LIMIT, "send_book_width": 162, "max_cluster_columns": 9}, "energy": {"doctrine": "Maximal work under limited power using minimal energy", "background_interval": 20}, } class Config: def __init__(self, paths: E8iPaths): self.paths = paths self.data = DEFAULT_CONFIG.copy() self.load() def load(self) -> None: loaded = read_json(self.paths.config, {}) self.data = deep_merge(DEFAULT_CONFIG, loaded) self.save() def save(self) -> None: write_json(self.paths.config, self.data) def get(self, *keys: str, default: Any = None) -> Any: cur: Any = self.data for k in keys: if not isinstance(cur, dict) or k not in cur: return default cur = cur[k] return cur def set(self, path: List[str], value: Any) -> None: cur = self.data for k in path[:-1]: cur = cur.setdefault(k, {}) cur[path[-1]] = value self.save() def deep_merge(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]: out = json.loads(json.dumps(a)) for k, v in b.items(): if isinstance(v, dict) and isinstance(out.get(k), dict): out[k] = deep_merge(out[k], v) else: out[k] = v return out ORGAN_DEFINITIONS = [ ("Raw Admission", "accept any file as raw bytes", "scan/compress"), ("Ultimata Scanner", "entropy, magic, file identity, hashes", "scan/doctor"), ("Granpa Numerono", "number/fence readable rows", "sd1/send-book"), ("Grandma Splitta", "split rows into geometry-ready chunks", "sd1"), ("Cleanup / Anty Reserve", "escape reserved operator symbols", "Brother Quicky"), ("Brother Quicky Q0/Q1/Q2", "RLE, row refs, dictionary", "sd1"), ("Geometry Twist", "vertical and super-column projections", "sd2/send-book"), ("Functional DNA Grammar", "basket/cluster/strand tokens with parser contracts", "dna5/basket/doctor"), ("SDO Ladder", ".sd4 .sd3 .sd2 .sd1 .sd0", "compress/decode"), ("Candidate Arena", "smallest verified reversible payload wins", "sd0"), ("REPOSITORY.zip", "repository shelf, not the identity", "repository"), ("SEND BOOK", "read-only expanded super-column export", "send-book/gui"), ("Chat Organism", "local dialogue surface and optional model aide", "chat/gui"), ("Tetrahedron", "4-basket polarity visual/function surface", "gui/doctor"), ("Tesseract", "stage-space visualiser", "gui/doctor"), ("Interactor Force", "online/API permission gate", "config/chat"), ("Doctor", "dependency, grammar, organ, roundtrip report", "doctor"), ] def write_organ_registry(paths: E8iPaths) -> Dict[str, Any]: registry = { "program": VERSION, "written_utc": utc_now(), "doctrine": { "dna_is_not_semantic": DNA_IS_NOT_SEMANTIC, "dna_is_functional_grammar": DNA_IS_FUNCTIONAL_GRAMMAR, "every_token_requires_role": EVERY_DNA_TOKEN_REQUIRES_A_ROLE, }, "organs": [ {"name": n, "role": r, "surface": s, "status": "active"} for n, r, s in ORGAN_DEFINITIONS ], } write_json(paths.organ_registry, registry) return registry # --------------------------------------------------------------------------- # Pure mathematical state (functional; not a physics proof) # --------------------------------------------------------------------------- SECTOR_BOUNDS = { "U1": (0, 3), "SU2": (3, 6), "SU3": (6, 14), "GRAVIWEAK": (14, 16), "F4_G2": (16, 24), "EXCEPTIONAL": (24, 248), } class CoreState248: def __init__(self) -> None: norm = 1.0 / math.sqrt(248) self.real = [norm] * 248 self.imag = [0.0] * 248 self.birth = time.time() def perceive(self, metadata: Dict[str, Any], scale: float = 1.0) -> None: ent = float(metadata.get("entropy_bits_per_byte", 0.0)) raw_len = int(metadata.get("raw_len", 0)) magic = str(metadata.get("magic", "")) if ent > 7.5: self._nudge("EXCEPTIONAL", 0.015 * ent / 8.0 * scale) if raw_len > 100_000: self._nudge("F4_G2", 0.01 * math.log2(max(2, raw_len) / 1000.0) * scale) if magic in {"python", "py", "cpp", "c", "h", "json", "md", "tex", "txt"}: self._nudge("SU3", 0.02 * scale) if magic in {"exe", "dll", "elf", "bin", "7z", "zip"}: self._nudge("F4_G2", 0.015 * scale) self._renormalise() def metabolic_decay(self, decay_constant: float = 0.0002) -> None: base = 1.0 / math.sqrt(248) for i in range(248): self.real[i] = self.real[i] * (1.0 - decay_constant) + base * decay_constant self.imag[i] *= (1.0 - decay_constant) self._renormalise() def _nudge(self, sector: str, scale: float) -> None: start, end = SECTOR_BOUNDS.get(sector, (0, 0)) if end <= start: return dim = end - start for i in range(start, end): self.real[i] = max(-10.0, min(10.0, self.real[i] + scale / dim)) def _renormalise(self) -> None: mag = math.sqrt(sum(r * r + im * im for r, im in zip(self.real, self.imag))) or 1.0 self.real = [r / mag for r in self.real] self.imag = [im / mag for im in self.imag] def activations(self) -> Dict[str, float]: out: Dict[str, float] = {} for name, (s, e) in SECTOR_BOUNDS.items(): out[name] = sum(self.real[i] ** 2 + self.imag[i] ** 2 for i in range(s, e)) return out def phase(self) -> Dict[str, str]: a = self.activations() if a["EXCEPTIONAL"] > 0.30 and a["SU3"] < a["U1"] * 1.2: return {"phase": "AUTOPOIETIC", "reason": "exceptional activation dominates"} if a["EXCEPTIONAL"] > 0.15: return {"phase": "TRANSITIONING", "reason": "exceptional sectors waking"} return {"phase": "LARVAL", "reason": "baseline local state"} # --------------------------------------------------------------------------- # Functional DNA grammar (not semantic) # --------------------------------------------------------------------------- @dataclasses.dataclass class BasketToken: power: str polarity: str payload_hex: str def render(self) -> str: return f"{self.power}{self.polarity}{{{self.payload_hex}}}" def decode_payload(self) -> bytes: return bytes.fromhex(self.payload_hex) BASKET_RE = re.compile(r"(3[⁰¹²³])([+±\-∓])\{([0-9A-Fa-f]*)\}") def bytes_to_basket_tokens(data: bytes, group: int = 6) -> List[BasketToken]: tokens: List[BasketToken] = [] for idx in range(0, len(data), group): chunk = data[idx:idx + group] k = (idx // group) % 4 tokens.append(BasketToken(BASKET_POWERS[k], BASKET_POLARITIES[k], chunk.hex().upper())) return tokens def basket_tokens_to_bytes(text: str) -> bytes: out = bytearray() pos = 0 for m in BASKET_RE.finditer(text): if text[pos:m.start()].strip(): # Non-token material is allowed as layout/fence, not payload. pass tok = BasketToken(m.group(1), m.group(2), m.group(3)) out.extend(tok.decode_payload()) pos = m.end() return bytes(out) def dna5_encode(data: bytes) -> str: # Functional, compact, reversible route: strand selector + two hex digits. # Strands are alphabet reservoirs and route markers; byte value is hex payload. bases = "ATCGZ" parts: List[str] = ["DNA5:v1:"] for i, b in enumerate(data): parts.append(f"{bases[i % 5]}{b:02X}") return "".join(parts) def dna5_decode(text: str) -> bytes: if not text.startswith("DNA5:v1:"): raise ValueError("not DNA5:v1") body = text[len("DNA5:v1:"):] if len(body) % 3 != 0: raise ValueError("bad DNA5 length") out = bytearray() for i in range(0, len(body), 3): strand = body[i] hx = body[i + 1:i + 3] if strand not in "ATCGZ": raise ValueError(f"bad strand {strand!r}") out.append(int(hx, 16)) return bytes(out) def render_cluster(tokens: List[BasketToken], width_tokens: int = 2, identity: str = "gH") -> str: if len(identity) < 2: identity = "gH" left_id, right_id = identity[0], identity[-1] rows = [] for i in range(0, len(tokens), width_tokens): chunk = tokens[i:i + width_tokens] payload = "".join(t.render() for t in chunk) if i == 0 and i + width_tokens >= len(tokens): rows.append(f"°{{{left_id}{payload}{right_id}|•") elif i == 0: rows.append(f"°{{{left_id}{payload}|•") elif i + width_tokens >= len(tokens): rows.append(f"o{{{payload}{right_id}|•") else: rows.append(f"•|{payload}{right_id}|•") return "\n".join(rows) def box_lines(lines: List[str], axis: str = "ZYX", width: Optional[int] = None) -> List[str]: if not lines: lines = [""] body_width = width or max(len(x) for x in lines) body_width = max(body_width, 4) top = axis[:2] + axis[2] * body_width bottom = axis[1] + axis[2] * (body_width + 1) out = [top] for line in lines: out.append("X" + line.ljust(body_width) + "X") out.append(bottom) return out def cluster_row_pack(clusters: List[List[str]], max_columns: int = 9) -> List[str]: # Only equally tall cluster columns can form a row. Others are grouped separately by height. packed: List[str] = [] for group_start in range(0, len(clusters), max_columns): group = clusters[group_start:group_start + max_columns] heights = {len(g) for g in group} if len(heights) != 1: packed.append(f"[CORRUPT-REJECT: unequal heights {sorted(heights)}]") continue h = len(group[0]) for r in range(h): packed.append("Z".join(g[r] for g in group)) return packed def grammar_report() -> Dict[str, Any]: sample = b"8i functional DNA grammar sample" btxt = "".join(t.render() for t in bytes_to_basket_tokens(sample)) d5 = dna5_encode(sample) return { "dna_is_not_semantic": DNA_IS_NOT_SEMANTIC, "functional_grammar_only": DNA_IS_FUNCTIONAL_GRAMMAR, "basket_roundtrip": basket_tokens_to_bytes(btxt) == sample, "dna5_roundtrip": dna5_decode(d5) == sample, "reserved_symbols": RESERVED_MAP, "bracket_register": BRACKET_REGISTER, "basket_roles": BASKET_ROLE, "strand_lengths": {k: len(v) for k, v in STRANDS.items()}, } # --------------------------------------------------------------------------- # Brother Quicky # --------------------------------------------------------------------------- def clean_encode(s: str) -> str: return "".join(RESERVED_MAP.get(ch, ch) for ch in s) def clean_decode(s: str) -> str: out: List[str] = [] i = 0 while i < len(s): if s.startswith(BOUNDARY, i): i += len(BOUNDARY) continue if s[i] == ESC: pair = s[i:i + 2] if pair in UNRESERVED_MAP: out.append(UNRESERVED_MAP[pair]) i += 2 continue raise ValueError(f"unknown clean escape {pair!r}") out.append(s[i]) i += 1 return "".join(out) def quicky_rle_encode_cleaned(s: str) -> str: out: List[str] = [] i = 0 while i < len(s): ch = s[i] j = i + 1 while j < len(s) and s[j] == ch: j += 1 n = j - i token = f"{n}*{ch}" literal = ch * n if n > 1 and len(token) < len(literal): if out and out[-1] and out[-1][-1].isdigit(): out.append(BOUNDARY) out.append(token) else: out.append(literal) i = j return "".join(out) def quicky_rle_decode_cleaned(s: str) -> str: out: List[str] = [] i = 0 while i < len(s): if s.startswith(BOUNDARY, i): i += len(BOUNDARY) continue if s[i].isdigit(): j = i + 1 while j < len(s) and s[j].isdigit(): j += 1 if j < len(s) and s[j] == "*" and j + 1 < len(s): out.append(s[j + 1] * int(s[i:j])) i = j + 2 else: out.append(s[i:j]) i = j else: out.append(s[i]) i += 1 return "".join(out) def number_rows(text: str, width: int = 158) -> List[str]: # Granpa Numerono + Grandma Splitta: visible fixed-width row bodies. text = text.replace(" ", SPACE_VIS) rows = [] for idx in range(0, len(text), width): rows.append(f"{len(rows):03d}:" + text[idx:idx + width]) return rows or ["000:"] def quicky_q0(rows: List[str]) -> List[str]: out: List[str] = [] for row in rows: label, body = row[:4], row[4:] enc = quicky_rle_encode_cleaned(clean_encode(body)) if clean_decode(quicky_rle_decode_cleaned(enc)).replace(SPACE_VIS, " ") != body.replace(SPACE_VIS, " "): # Verify without losing the visible placeholder rule. raise AssertionError(f"Q0 failed at {label}") out.append(label + enc) return out def quicky_q1(rows: List[str]) -> List[str]: seen: Dict[str, str] = {} out: List[str] = [] for line in rows: label, body = line[:4], line[4:] if body in seen: out.append(f"{label}{REF_MARK}{seen[body]}") else: seen[body] = label[:3] out.append(line) return out def quicky_expand_q1(rows: List[str]) -> List[str]: table: Dict[str, str] = {} out: List[str] = [] for line in rows: label, body = line[:4], line[4:] if body.startswith(REF_MARK): ref = body[1:] if ref not in table: raise ValueError(f"bad row ref {label}->{ref}") body = table[ref] table[label[:3]] = body out.append(label + body) return out def choose_dictionary(lines: List[str], max_entries: int = 64, min_count: int = 3) -> Tuple[List[Tuple[str, str]], List[str]]: current = list(lines) dictionary: List[Tuple[str, str]] = [] for idx in range(max_entries): token = f"{DICT_MARK}{idx:02d}{DICT_MARK}" counter: Counter[str] = Counter() for line in current: body = line[4:] for length in (96, 80, 64, 48, 40, 32, 24, 20, 16, 12, 10, 8, 6): if len(body) < length: continue for pos in range(0, len(body) - length + 1): sub = body[pos:pos + length] if DICT_MARK in sub or REF_MARK in sub: continue counter[sub] += 1 best = None best_gain = 0 for sub, count in counter.items(): if count < min_count: continue gain = (len(sub) - len(token)) * count - len(sub) - len(token) - 4 if gain > best_gain: best, best_gain = sub, gain if not best: break dictionary.append((token, best)) current = [line[:4] + line[4:].replace(best, token) for line in current] return dictionary, current def expand_dictionary(dictionary: List[Tuple[str, str]], lines: List[str]) -> List[str]: out = list(lines) for token, phrase in reversed(dictionary): out = [line.replace(token, phrase) for line in out] return out def brother_quicky_package(raw: bytes, width: int = 158) -> Dict[str, Any]: text = raw.decode("utf-8", errors="surrogateescape") rows0 = number_rows(text, width=width) q0 = quicky_q0(rows0) q1 = quicky_q1(q0) dictionary, q2 = choose_dictionary(q1) # Verify route. ex1 = expand_dictionary(dictionary, q2) ex0 = quicky_expand_q1(ex1) decoded_lines = [] for line in ex0: body = line[4:] decoded_lines.append(clean_decode(quicky_rle_decode_cleaned(body))) restored_text = "".join(decoded_lines).replace(SPACE_VIS, " ") restored = restored_text.encode("utf-8", errors="surrogateescape") if restored != raw: raise AssertionError("Brother Quicky roundtrip failed") return { "kind": "brother_quicky_v2", "width": width, "dictionary": dictionary, "rows": q2, "stats": { "raw_chars": len(text), "rows0_chars": sum(map(len, rows0)), "q0_chars": sum(map(len, q0)), "q1_chars": sum(map(len, q1)), "q2_chars": sum(map(len, q2)), "dict_entries": len(dictionary) }, } def decode_brother_quicky_package(pkg: Dict[str, Any]) -> bytes: dictionary = [(a, b) for a, b in pkg.get("dictionary", [])] rows = [str(x) for x in pkg.get("rows", [])] ex1 = expand_dictionary(dictionary, rows) ex0 = quicky_expand_q1(ex1) decoded = "".join(clean_decode(quicky_rle_decode_cleaned(line[4:])) for line in ex0).replace(SPACE_VIS, " ") return decoded.encode("utf-8", errors="surrogateescape") # --------------------------------------------------------------------------- # SDO envelope and compression # --------------------------------------------------------------------------- class SDO: @staticmethod def pack(method: str, payload: bytes, raw_sha256: str, raw_len: int, extra: Optional[Dict[str, Any]] = None) -> bytes: meta = { "program": VERSION, "created_utc": utc_now(), "method": method, "raw_sha256": raw_sha256, "raw_len": raw_len, "payload_len": len(payload), "payload_sha256": sha256_bytes(payload), "extra": extra or {}, } mb = json.dumps(meta, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return MAGIC + struct.pack(">Q", len(mb)) + mb + payload @staticmethod def unpack(blob: bytes) -> Tuple[Dict[str, Any], bytes]: magic = None for m in (MAGIC,) + LEGACY_MAGICS: if blob.startswith(m): magic = m break if magic is None: raise ValueError("not an E8I SDO object") pos = len(magic) if len(blob) < pos + 8: raise ValueError("truncated SDO") ml = struct.unpack(">Q", blob[pos:pos + 8])[0] pos += 8 meta = json.loads(blob[pos:pos + ml].decode("utf-8")) payload = blob[pos + ml:] if sha256_bytes(payload) != meta.get("payload_sha256"): raise ValueError("payload sha256 mismatch") return meta, payload @staticmethod def decode(blob: bytes) -> bytes: meta, payload = SDO.unpack(blob) method = meta.get("method") raw: bytes if method == "raw": raw = payload elif method == "zlib_raw": raw = zlib.decompress(payload) elif method == "gzip_raw": raw = gzip.decompress(payload) elif method == "bz2_raw": raw = bz2.decompress(payload) elif method == "lzma_raw": raw = lzma.decompress(payload) elif method == "zstd_raw": if _zstd is None: raise RuntimeError("zstandard not installed") raw = _zstd.ZstdDecompressor().decompress(payload) elif method == "dna5_text_zlib": raw = dna5_decode(zlib.decompress(payload).decode("ascii")) elif method == "basket_text_zlib": raw = basket_tokens_to_bytes(zlib.decompress(payload).decode("utf-8")) elif method == "brother_quicky_json_zlib": pkg = json.loads(zlib.decompress(payload).decode("utf-8")) raw = decode_brother_quicky_package(pkg) elif method == "geometry_text_zlib": g = json.loads(zlib.decompress(payload).decode("utf-8")) raw = base64.b64decode(g["raw_b64"]) elif method == "tar_lzma_base64_text": raw = lzma.decompress(base64.b64decode(payload)) else: raise ValueError(f"unknown SDO method {method!r}") if sha256_bytes(raw) != meta.get("raw_sha256"): raise ValueError("raw sha256 mismatch") if len(raw) != int(meta.get("raw_len", -1)): raise ValueError("raw length mismatch") return raw @staticmethod def metadata(blob: bytes) -> Dict[str, Any]: meta, _ = SDO.unpack(blob) return meta def geometry_projection(raw: bytes, name: str = "payload") -> Dict[str, Any]: # For very large raw, keep projection as manifest + raw_b64 for reversible route. # The visible preview is capped; raw_b64 is exact. tokens = bytes_to_basket_tokens(raw[:min(len(raw), 512)]) cluster = render_cluster(tokens, width_tokens=2, identity="gH") box = box_lines(cluster.splitlines(), axis="ZYX") return { "kind": "geometry_projection_v2", "name": name, "raw_b64": base64.b64encode(raw).decode("ascii"), "preview": box, "rules": { "cluster_columns_equal_height_required": True, "max_cluster_columns_per_row": 9, "dna_is_functional_not_semantic": True, }, } def candidate_payloads(raw: bytes, name: str = "payload") -> List[Tuple[str, bytes, Dict[str, Any]]]: out: List[Tuple[str, bytes, Dict[str, Any]]] = [] out.append(("raw", raw, {})) out.append(("zlib_raw", zlib.compress(raw, 9), {})) out.append(("gzip_raw", gzip.compress(raw, compresslevel=9), {})) out.append(("bz2_raw", bz2.compress(raw, compresslevel=9), {})) out.append(("lzma_raw", lzma.compress(raw, preset=9 | lzma.PRESET_EXTREME), {})) if _zstd is not None: out.append(("zstd_raw", _zstd.ZstdCompressor(level=19).compress(raw), {})) d5 = dna5_encode(raw).encode("ascii") out.append(("dna5_text_zlib", zlib.compress(d5, 9), {"route": "DNA5"})) btxt = "".join(t.render() for t in bytes_to_basket_tokens(raw)).encode("utf-8") out.append(("basket_text_zlib", zlib.compress(btxt, 9), {"route": "basket"})) if is_text_like(raw) and len(raw) <= 5_000_000: try: bq = brother_quicky_package(raw) out.append(("brother_quicky_json_zlib", zlib.compress(json.dumps(bq, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), 9), {"stats": bq.get("stats", {})})) except Exception as e: out.append(("brother_quicky_failed_note", str(e).encode("utf-8"), {"rejected": True})) geom = geometry_projection(raw, name=name) out.append(("geometry_text_zlib", zlib.compress(json.dumps(geom, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), 9), {"route": "geometry"})) return out def build_stage(raw: bytes, method: str, payload: bytes, extra: Optional[Dict[str, Any]] = None) -> bytes: return SDO.pack(method, payload, sha256_bytes(raw), len(raw), extra=extra) def build_stages(input_path: Path, out_dir: Path, add_to_zip: bool = True) -> Dict[str, Any]: raw = input_path.read_bytes() out_dir.mkdir(parents=True, exist_ok=True) base = safe_name(input_path.stem) raw_sha = sha256_bytes(raw) scan = scan_bytes(raw, input_path.name) # .sd4 raw admission sd4 = build_stage(raw, "raw", raw, {"stage": "sd4", "scan": scan}) # .sd3 zlib raw sd3_payload = zlib.compress(raw, 9) sd3 = build_stage(raw, "zlib_raw", sd3_payload, {"stage": "sd3"}) # .sd2 geometry route geom = geometry_projection(raw, input_path.name) sd2_payload = zlib.compress(json.dumps(geom, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), 9) sd2 = build_stage(raw, "geometry_text_zlib", sd2_payload, {"stage": "sd2"}) # .sd1 Brother Quicky if possible, else basket route try: bq = brother_quicky_package(raw) sd1_payload = zlib.compress(json.dumps(bq, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), 9) sd1 = build_stage(raw, "brother_quicky_json_zlib", sd1_payload, {"stage": "sd1", "stats": bq.get("stats", {})}) except Exception as e: btxt = "".join(t.render() for t in bytes_to_basket_tokens(raw)).encode("utf-8") sd1 = build_stage(raw, "basket_text_zlib", zlib.compress(btxt, 9), {"stage": "sd1", "fallback": "basket", "brother_quicky_error": str(e)}) # .sd0 candidate arena: only verified candidates can win. verified: List[Tuple[str, bytes, Dict[str, Any], int]] = [] for method, payload, extra in candidate_payloads(raw, input_path.name): if extra.get("rejected"): continue try: obj = build_stage(raw, method, payload, {"stage": "sd0_candidate", **extra}) if SDO.decode(obj) == raw: verified.append((method, obj, extra, len(obj))) except Exception: continue if not verified: raise RuntimeError("no verified sd0 candidates") method, sd0, extra, _ = min(verified, key=lambda x: x[3]) files = { "sd4": out_dir / f"{base}.sd4", "sd3": out_dir / f"{base}.sd3", "sd2": out_dir / f"{base}.sd2", "sd1": out_dir / f"{base}.sd1", "sd0": out_dir / f"{base}.sd0", } blobs = {"sd4": sd4, "sd3": sd3, "sd2": sd2, "sd1": sd1, "sd0": sd0} for k, p in files.items(): atomic_write(p, blobs[k]) assert SDO.decode(p.read_bytes()) == raw repo_zip = out_dir / "REPOSITORY.zip" if add_to_zip: upsert_zip(repo_zip, files["sd0"].name, files["sd0"].read_bytes()) manifest = { "program": VERSION, "input": str(input_path), "raw_len": len(raw), "raw_sha256": raw_sha, "scan": scan, "stage_files": {k: {"path": str(p), "bytes": p.stat().st_size, "method": SDO.metadata(p.read_bytes()).get("method")} for k, p in files.items()}, "sd0_winner": method, "verified_candidates": [{"method": m, "bytes": n} for m, _, _, n in sorted(verified, key=lambda x: x[3])], "repository_zip": str(repo_zip) if add_to_zip else None, } atomic_write(out_dir / f"{base}_manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8")) return manifest def upsert_zip(zip_path: Path, arcname: str, data: bytes) -> None: zip_path.parent.mkdir(parents=True, exist_ok=True) tmp = zip_path.with_suffix(zip_path.suffix + ".tmp") existing: Dict[str, bytes] = {} if zip_path.exists(): with zipfile.ZipFile(zip_path, "r") as z: for n in z.namelist(): if n != arcname: existing[n] = z.read(n) existing[arcname] = data with zipfile.ZipFile(tmp, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as z: for n, b in sorted(existing.items()): z.writestr(n, b) os.replace(tmp, zip_path) def scan_bytes(raw: bytes, name: str) -> Dict[str, Any]: return { "name": name, "raw_len": len(raw), "sha256": sha256_bytes(raw), "entropy_bits_per_byte": round(entropy_bits_per_byte(raw), 6), "magic": detect_magic(name, raw), "text_like": is_text_like(raw), "nul_in_first_4096": b"\x00" in raw[:4096], } # --------------------------------------------------------------------------- # SEND BOOK / TXT bundle # --------------------------------------------------------------------------- def super_column_from_bytes(data: bytes, width: int = 162, expanded: bool = True) -> str: # Expanded view: one visible byte-token per row with offset. This is inspectable and lossless via hex. lines = ["E8I SEND BOOK SUPER-COLUMN", f"WIDTH={width}", f"BYTES={len(data)}", f"SHA256={sha256_bytes(data)}", "---"] for i, b in enumerate(data): ch = chr(b) if 32 <= b <= 126 else SPACE_VIS if b == 32 else "·" lines.append(f"{i:012d}: {b:02X} {ch}") return "\n".join(lines) + "\n" def send_book_export(repository_zip: Path, exportory: Path, veil_seconds: int = DEFAULT_VEIL_SECONDS, width: int = 162, max_rows: int = 0) -> Dict[str, Any]: exportory.mkdir(parents=True, exist_ok=True) if not repository_zip.exists(): raise FileNotFoundError(repository_zip) # Copy snapshot; never mutate source. snap = exportory / f"REPOSITORY_snapshot_{int(time.time())}.zip" shutil.copy2(repository_zip, snap) entries: List[Dict[str, Any]] = [] combined = bytearray() with zipfile.ZipFile(snap, "r") as z: for name in sorted(z.namelist()): data = z.read(name) try: raw = SDO.decode(data) status = "decoded" except Exception: raw = data status = "raw_entry" marker = f"\n===== ENTRY {name} STATUS {status} =====\n".encode("utf-8") combined.extend(marker) combined.extend(raw) entries.append({"name": name, "stored_bytes": len(data), "raw_bytes": len(raw), "sha256": sha256_bytes(raw), "status": status}) super_text = super_column_from_bytes(bytes(combined), width=width) if max_rows and max_rows > 0: parts = super_text.splitlines()[:max_rows] parts.append(f"[TRUNCATED VIEW max_rows={max_rows}; source hash preserved in manifest]") super_text = "\n".join(parts) + "\n" out = exportory / f"SEND_BOOK_super_column_{_dt.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" atomic_write(out, super_text.encode("utf-8")) manifest = { "program": VERSION, "generated_utc": utc_now(), "veil_seconds_requested": veil_seconds, "source_repository": str(repository_zip), "snapshot": str(snap), "snapshot_sha256": sha256_bytes(snap.read_bytes()), "output": str(out), "output_sha256": sha256_bytes(out.read_bytes()), "entries": entries, "read_only": True, } atomic_write(out.with_suffix(".manifest.json"), json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8")) return manifest def combine_files_to_lossless_txt(paths: List[Path], out_txt: Path) -> Dict[str, Any]: tar_buffer = io.BytesIO() files = [] with tarfile.open(fileobj=tar_buffer, mode="w") as tar: for idx, p in enumerate(paths, 1): data = p.read_bytes() info = tarfile.TarInfo(name=p.name) info.size = len(data) info.mtime = int(p.stat().st_mtime) tar.addfile(info, io.BytesIO(data)) files.append({"index": idx, "filename": p.name, "bytes": len(data), "sha256": sha256_bytes(data)}) tar_bytes = tar_buffer.getvalue() comp = lzma.compress(tar_bytes, preset=9 | lzma.PRESET_EXTREME) b64 = base64.b64encode(comp).decode("ascii") wrapped = "\n".join(re.findall(r".{1,76}", b64)) manifest = {"container": "E8I_LOSSLESS_TEXT_BUNDLE", "version": 2, "created_utc": utc_now(), "method": "tar->lzma->base64", "source_files": files, "tar_sha256": sha256_bytes(tar_bytes), "compressed_sha256": sha256_bytes(comp), "lossless": True} text = "E8I LOSSLESS TEXT BUNDLE\n---MANIFEST-JSON---\n" + json.dumps(manifest, ensure_ascii=False, indent=2) + "\n---END-MANIFEST-JSON---\n---PAYLOAD-BASE64-LZMA-TAR---\n" + wrapped + "\n---END-PAYLOAD-BASE64-LZMA-TAR---\n" atomic_write(out_txt, text.encode("utf-8")) # Verify from text. extract_lossless_txt(out_txt, out_txt.parent / "_verify_extract", verify_only=True) return {"output": str(out_txt), "output_bytes": out_txt.stat().st_size, "manifest": manifest} def extract_lossless_txt(bundle: Path, out_dir: Path, verify_only: bool = False) -> Dict[str, Any]: text = bundle.read_text(encoding="utf-8") m = re.search(r"---MANIFEST-JSON---\n(.*?)\n---END-MANIFEST-JSON---", text, re.S) p = re.search(r"---PAYLOAD-BASE64-LZMA-TAR---\n(.*?)\n---END-PAYLOAD-BASE64-LZMA-TAR---", text, re.S) if not m or not p: raise ValueError("bad lossless txt bundle") manifest = json.loads(m.group(1)) comp = base64.b64decode("".join(p.group(1).split())) if sha256_bytes(comp) != manifest["compressed_sha256"]: raise ValueError("compressed sha mismatch") tar_bytes = lzma.decompress(comp) if sha256_bytes(tar_bytes) != manifest["tar_sha256"]: raise ValueError("tar sha mismatch") out_dir.mkdir(parents=True, exist_ok=True) with tarfile.open(fileobj=io.BytesIO(tar_bytes), mode="r:") as tar: # Use simple safe extraction: block absolute paths and traversal. for member in tar.getmembers(): target = (out_dir / member.name).resolve() if not str(target).startswith(str(out_dir.resolve())): raise ValueError(f"unsafe tar path {member.name}") tar.extractall(out_dir) for item in manifest["source_files"]: data = (out_dir / item["filename"]).read_bytes() if sha256_bytes(data) != item["sha256"]: raise ValueError(f"source sha mismatch {item['filename']}") if verify_only: shutil.rmtree(out_dir, ignore_errors=True) return {"status": "LOSSLESS_EXTRACT_OK", "out_dir": str(out_dir), "manifest": manifest} # --------------------------------------------------------------------------- # Chat / local aides # --------------------------------------------------------------------------- class LocalChat: def __init__(self, paths: E8iPaths, config: Config): self.paths = paths self.config = config self.history: deque[Tuple[str, str]] = deque(maxlen=50) def respond(self, prompt: str) -> str: prompt_l = prompt.lower().strip() if self.config.get("internet_force", default=False): r = self.try_ollama(prompt) if r: return r if any(w in prompt_l for w in ["what are you", "who are you", "status"]): return "E8i: local functional-DNA organism. DNA mode is grammar-only, not semantic. Use doctor, compress, send-book, or scan." if "dna" in prompt_l: gr = grammar_report() return "E8i: DNA grammar PASS. Basket roundtrip=%s; DNA5 roundtrip=%s; semantic mode=OFF." % (gr["basket_roundtrip"], gr["dna5_roundtrip"]) if "send book" in prompt_l: return "E8i: SEND BOOK reads REPOSITORY.zip read-only and exports expanded super-column TXT. Use ( SEND BOOK ) or CLI send-book." if "compress" in prompt_l or ".sd0" in prompt_l: return "E8i: compression ladder active: .sd4 raw, .sd3 zlib, .sd2 geometry, .sd1 Brother Quicky/basket, .sd0 smallest verified candidate." if "doctor" in prompt_l: return "E8i: Doctor checks dependencies, folders, organ registry, DNA grammar, and roundtrip routes." return "E8i: received. I can scan, compress, decode, export SEND BOOK, check DNA grammar, and report repository state." def try_ollama(self, prompt: str) -> Optional[str]: if not self.config.get("ollama", "enabled", default=False): return None url = self.config.get("ollama", "url", default="http://127.0.0.1:11434/api/generate") model = self.config.get("ollama", "model", default="qwen2.5-coder:1.5b") timeout = int(self.config.get("ollama", "timeout", default=20)) try: payload = json.dumps({"model": model, "prompt": prompt, "stream": False}).encode("utf-8") req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=timeout) as resp: obj = json.loads(resp.read().decode("utf-8")) return obj.get("response") or None except Exception: return None def log(self, speaker: str, text: str) -> None: line = f"[{utc_now()}] {speaker}: {text}\n" self.paths.chat_log.parent.mkdir(parents=True, exist_ok=True) with open(self.paths.chat_log, "a", encoding="utf-8") as f: f.write(line) # --------------------------------------------------------------------------- # Doctor/bootstrap/agent scan # --------------------------------------------------------------------------- def dependency_status() -> Dict[str, Any]: pkgs = ["PyQt6", "psutil", "PIL", "zstandard", "watchdog", "rich", "numpy", "pyqtgraph", "blake3", "requests", "pyttsx3"] out: Dict[str, Any] = {} for p in pkgs: mod = "PIL" if p == "PIL" else p try: __import__(mod) out[p] = True except Exception: out[p] = False return out def doctor(paths: E8iPaths, config: Config) -> Dict[str, Any]: write_organ_registry(paths) gr = grammar_report() repo_entries = [] if paths.repository_zip.exists(): try: with zipfile.ZipFile(paths.repository_zip, "r") as z: repo_entries = z.namelist() except Exception: repo_entries = [""] sysinfo = { "python": sys.version, "platform": platform.platform(), "cwd": str(Path.cwd()), "hostname": socket.gethostname(), } if _psutil is not None: try: sysinfo["ram_total_gb"] = round(_psutil.virtual_memory().total / 1024 ** 3, 2) sysinfo["ram_available_gb"] = round(_psutil.virtual_memory().available / 1024 ** 3, 2) sysinfo["cpu_percent"] = _psutil.cpu_percent(interval=0.1) except Exception: pass return { "program": VERSION, "created_utc": utc_now(), "system": sysinfo, "paths": {k: str(v) for k, v in dataclasses.asdict(paths).items()}, "dependencies": dependency_status(), "organ_count": len(ORGAN_DEFINITIONS), "organs_active": [x[0] for x in ORGAN_DEFINITIONS], "dna_grammar": gr, "repository_zip_entries": repo_entries, "repository_zip_count": len(repo_entries), "interactor_force": config.get("internet_force", default=False), "model_targets": MODEL_TARGETS, "button_text_contract": ["( SEND BOOK )", "( DOWNLOAD .SD4 )", "( DOWNLOAD .SD3 )", "( DOWNLOAD .SD2 )", "( DOWNLOAD .SD1 )", "( DOWNLOAD .SD0 )"], } def bootstrap_text() -> str: return """8i Bootstrap Plan ================= Windows CMD as Administrator when installing system tools: py -m pip install --upgrade pip py -m pip install -r requirements_8i.txt Core GUI packages: py -m pip install PyQt6 psutil pillow Recommended organism packages: py -m pip install zstandard watchdog rich numpy pyqtgraph blake3 requests pyttsx3 Optional local model stack: Install Ollama separately, then pull a small coder model such as qwen2.5-coder:1.5b if available. Or install llama.cpp / LM Studio and keep GGUF models in the models folder. System tools to install manually by name: Python 3.11 or 3.12, Git, Visual Studio Code, Microsoft Visual Studio Build Tools, 7-Zip, Windows Terminal, PowerShell, optional Rust toolchain, optional Node.js LTS. No secrets in this folder. API keys belong in environment variables only. """ def write_readme2(paths: E8iPaths) -> Path: content = bootstrap_text() + "\nDoctrine:\n DNA is functional grammar, not semantic. Every .sd stage must verify SHA-256. SEND BOOK is read-only.\n" out = paths.root / "README2_REQUIREMENTS_8i.md" atomic_write(out, content.encode("utf-8")) req = paths.root / "requirements_8i.txt" atomic_write(req, b"PyQt6\npsutil\npillow\nzstandard\nwatchdog\nrich\nnumpy\npyqtgraph\nblake3\nrequests\npyttsx3\n") # helper scripts atomic_write(paths.root / "install_windows.cmd", ("@echo off\r\npy -m pip install --upgrade pip\r\npy -m pip install -r requirements_8i.txt\r\npy 8i.py --self-test\r\npy 8i.py doctor\r\n").encode("utf-8")) atomic_write(paths.root / "install_linux.sh", ("#!/usr/bin/env bash\npython3 -m pip install --upgrade pip\npython3 -m pip install -r requirements_8i.txt\npython3 8i.py --self-test\npython3 8i.py doctor\n").encode("utf-8")) return out def agent_scan(root: Path) -> Dict[str, Any]: files = [] for p in root.rglob("*"): if p.is_file() and p.suffix.lower() in {".py", ".txt", ".md", ".json", ".toml", ".yml", ".yaml", ".rs", ".cpp", ".h"}: if any(part in {"__pycache__", ".git", ".lrn"} for part in p.parts): continue try: data = p.read_bytes() files.append({"path": str(p.relative_to(root)), "bytes": len(data), "sha256": sha256_bytes(data), "magic": detect_magic(p.name, data)}) except Exception: pass return {"root": str(root), "count": len(files), "files": files[:500]} # --------------------------------------------------------------------------- # GUI # --------------------------------------------------------------------------- def launch_gui(paths: E8iPaths, config: Config) -> int: try: from PyQt6.QtCore import Qt, QThread, pyqtSignal, QTimer, QRectF, QPointF from PyQt6.QtGui import QColor, QFont, QLinearGradient, QPainter, QPainterPath, QPen, QBrush from PyQt6.QtWidgets import ( QApplication, QWidget, QVBoxLayout, QHBoxLayout, QGridLayout, QPushButton, QLabel, QTextEdit, QLineEdit, QFileDialog, QFrame, QSlider, QProgressBar, QMessageBox ) except Exception as e: print("PyQt6 not available. Install: py -m pip install PyQt6 psutil pillow") print(e) return 2 class Worker(QThread): done = pyqtSignal(object) failed = pyqtSignal(str) def __init__(self, fn, *args, **kwargs): super().__init__() self.fn = fn; self.args = args; self.kwargs = kwargs def run(self): try: self.done.emit(self.fn(*self.args, **self.kwargs)) except Exception: self.failed.emit(traceback.format_exc()) class GeometryPanel(QFrame): def __init__(self): super().__init__() self.t = 0.0 self.setMinimumHeight(260) self.setStyleSheet("QFrame{background:rgba(255,255,255,44);border:1px solid rgba(255,255,255,160);border-radius:18px;}") self.timer = QTimer(self); self.timer.timeout.connect(self.tick); self.timer.start(50) def tick(self): self.t += 0.04; self.update() def paintEvent(self, ev): super().paintEvent(ev) p = QPainter(self); p.setRenderHint(QPainter.RenderHint.Antialiasing) w, h = self.width(), self.height() cx, cy = w * 0.50, h * 0.48 # glass lens grad = QLinearGradient(0, 0, 0, h) grad.setColorAt(0, QColor(230, 255, 255, 100)); grad.setColorAt(1, QColor(0, 130, 210, 70)) p.setBrush(QBrush(grad)); p.setPen(QPen(QColor(255,255,255,180), 2)) p.drawEllipse(QRectF(cx-92, cy-92, 184, 184)) # tetrahedron pts = [] for a in [self.t, self.t+2.09, self.t+4.18]: pts.append(QPointF(cx + math.cos(a)*78, cy + math.sin(a)*48)) top = QPointF(cx, cy-86) p.setPen(QPen(QColor(255,255,255,220), 2)) for pt in pts: p.drawLine(top, pt) for i in range(3): p.drawLine(pts[i], pts[(i+1)%3]) p.setFont(QFont("Segoe UI", 10, QFont.Weight.Bold)) for label, pt in zip(["+", "±", "-"], pts): p.drawText(pt.x()-8, pt.y()+5, label) p.drawText(top.x()-8, top.y()-8, "∓") # tesseract wire cube x0, y0 = 18, 22; s = 72; off = 28 p.setPen(QPen(QColor(180,255,255,180), 2)) rects = [(x0,y0,s,s),(x0+off,y0+off,s,s)] for x,y,ww,hh in rects: p.drawRect(x,y,ww,hh) for dx,dy in [(0,0),(s,0),(0,s),(s,s)]: p.drawLine(x0+dx,y0+dy,x0+off+dx,y0+off+dy) p.setFont(QFont("Segoe UI", 9)) p.drawText(18, 140, ".sd4→.sd3→.sd2→.sd1→.sd0") p.drawText(w-150, 26, "+Z VERIFIED") p.end() class Main(QWidget): def __init__(self): super().__init__() self.paths = paths self.config = config self.chat = LocalChat(paths, config) self.last_manifest: Optional[Dict[str, Any]] = None self.current_worker: Optional[Worker] = None self.setWindowTitle("E8i / V.A.S.K. — Total Absolution Functional DNA") self.resize(int(config.get("gui", "window_width", default=1180)), int(config.get("gui", "window_height", default=860))) self.setAcceptDrops(True) self.apply_style() self.build() self.timer = QTimer(self); self.timer.timeout.connect(self.refresh_meters); self.timer.start(1000) def apply_style(self): self.setStyleSheet(''' QWidget { font-family: Segoe UI, Arial; color: white; } QLabel#title { font-size: 20px; font-weight: 700; } QLabel#sub { color: rgba(255,255,255,210); } QFrame.glass { background: rgba(255,255,255,48); border: 1px solid rgba(255,255,255,150); border-radius: 16px; } QPushButton { min-height: 34px; border-radius: 17px; padding: 5px 14px; color: white; font-weight: 700; background: qlineargradient(x1:0,y1:0,x2:0,y2:1, stop:0 rgba(255,255,255,160), stop:.48 rgba(60,180,245,150), stop:.5 rgba(0,120,215,180), stop:1 rgba(0,70,170,210)); border: 1px solid rgba(255,255,255,210); } QPushButton:hover { background: qlineargradient(x1:0,y1:0,x2:0,y2:1, stop:0 rgba(255,255,255,220), stop:.48 rgba(100,210,255,180), stop:.5 rgba(0,145,235,210), stop:1 rgba(0,90,200,230)); } QTextEdit, QLineEdit { background: rgba(255,255,255,235); color: #092033; border-radius: 12px; border: 1px solid rgba(255,255,255,180); padding: 7px; } QProgressBar { border: 1px solid rgba(255,255,255,150); border-radius: 8px; background: rgba(255,255,255,45); text-align:center; } QProgressBar::chunk { border-radius: 8px; background: qlineargradient(x1:0,y1:0,x2:1,y2:0, stop:0 #74e6ff, stop:1 #1f87ff); } QSlider::groove:horizontal { height: 8px; background: rgba(255,255,255,80); border-radius: 4px; } QSlider::handle:horizontal { width: 24px; margin: -8px 0; border-radius: 12px; background: white; } ''') def paintEvent(self, event): p = QPainter(self) grad = QLinearGradient(0, 0, 0, self.height()) grad.setColorAt(0, QColor(80, 170, 235)); grad.setColorAt(.48, QColor(125, 220, 240)); grad.setColorAt(1, QColor(20, 105, 190)) p.fillRect(self.rect(), grad) p.setPen(Qt.PenStyle.NoPen) p.setBrush(QColor(255,255,255,40)) p.drawEllipse(-130, -80, 420, 250) p.drawEllipse(self.width()-320, self.height()-230, 460, 280) p.end() def glass(self): f=QFrame(); f.setProperty("class","glass"); return f def build(self): root=QVBoxLayout(self); root.setContentsMargins(18,18,18,18); root.setSpacing(12) ribbon=self.glass(); rb=QHBoxLayout(ribbon) title=QLabel("E8i / V.A.S.K. — ALPHA OMEGA LIGHT"); title.setObjectName("title") self.mode=QLabel("LOCAL · DNA FUNCTIONAL · SEMANTIC OFF · INTERACTOR FORCE OFF"); self.mode.setObjectName("sub") rb.addWidget(title); rb.addStretch(1); rb.addWidget(self.mode) root.addWidget(ribbon) mid=QHBoxLayout(); root.addLayout(mid,1) left=self.glass(); ll=QVBoxLayout(left) ll.addWidget(QLabel("ORGANISM METERS")) self.bars={} for name in ["CPU","RAM","DISK","REPO","ENERGY","DREAM","VOICE","MODEL"]: lab=QLabel(name); bar=QProgressBar(); bar.setRange(0,100); bar.setValue(0); self.bars[name]=bar; ll.addWidget(lab); ll.addWidget(bar) mid.addWidget(left, 1) centre=self.glass(); cl=QVBoxLayout(centre) self.geo=GeometryPanel(); cl.addWidget(self.geo) self.drop=QLabel("DROP ANY RAW FILE HERE\nor press ( INGEST )"); self.drop.setAlignment(Qt.AlignmentFlag.AlignCenter); self.drop.setMinimumHeight(86); self.drop.setStyleSheet("QLabel{font-size:18px;font-weight:700;background:rgba(255,255,255,58);border:1px dashed white;border-radius:18px;}") cl.addWidget(self.drop) self.info=QTextEdit(); self.info.setReadOnly(True); self.info.setMaximumHeight(150); self.info.setText("Ready. DNA is functional grammar, not semantic. No payload route is accepted unless it roundtrips.") cl.addWidget(self.info) mid.addWidget(centre, 2) right=self.glass(); rl=QVBoxLayout(right) rl.addWidget(QLabel("CHAT / INFORMATICS")) self.chatbox=QTextEdit(); self.chatbox.setReadOnly(True); self.chatbox.append("E8i: awake. Functional DNA grammar active. Semantic DNA mode OFF.") rl.addWidget(self.chatbox,1) inp_row=QHBoxLayout(); self.input=QLineEdit(); self.input.setPlaceholderText("Int: ask/status/compress/send book/doctor..."); send=QPushButton("( SEND )"); send.setMinimumWidth(120); send.clicked.connect(self.send_chat); self.input.returnPressed.connect(self.send_chat); inp_row.addWidget(self.input,1); inp_row.addWidget(send); rl.addLayout(inp_row) veilrow=QHBoxLayout(); veilrow.addWidget(QLabel("VEIL")); self.veil=QSlider(Qt.Orientation.Horizontal); self.veil.setRange(0,600); self.veil.setValue(int(self.config.get("gui","veil_seconds",default=600))); self.veil_label=QLabel("10:00"); self.veil.valueChanged.connect(lambda v: self.veil_label.setText(f"{v//60}:{v%60:02d}")); veilrow.addWidget(self.veil,1); veilrow.addWidget(self.veil_label); rl.addLayout(veilrow) mid.addWidget(right, 2) dock=self.glass(); dl=QGridLayout(dock) buttons=[("( INGEST )", self.ingest), ("( VERIFY )", self.verify_last), ("( DOCTOR )", self.run_doctor), ("( VEIL )", self.veil_focus), ("( SEND BOOK )", self.send_book), ("( REPOSITORY )", self.open_repo), ("( DOWNLOAD .SD4 )", lambda:self.copy_stage("sd4")), ("( DOWNLOAD .SD3 )", lambda:self.copy_stage("sd3")), ("( DOWNLOAD .SD2 )", lambda:self.copy_stage("sd2")), ("( DOWNLOAD .SD1 )", lambda:self.copy_stage("sd1")), ("( DOWNLOAD .SD0 )", lambda:self.copy_stage("sd0")), ("( EXIT )", self.close)] for i,(txt,fn) in enumerate(buttons): b=QPushButton(txt); b.setMinimumWidth(145); b.clicked.connect(fn); dl.addWidget(b, i//6, i%6) root.addWidget(dock) def refresh_meters(self): if _psutil: try: self.bars["CPU"].setValue(int(_psutil.cpu_percent())) self.bars["RAM"].setValue(int(_psutil.virtual_memory().percent)) self.bars["DISK"].setValue(int(_psutil.disk_usage(str(self.paths.root)).percent)) except Exception: pass try: with zipfile.ZipFile(self.paths.repository_zip,"r") as z: n=len(z.namelist()) except Exception: n=0 self.bars["REPO"].setValue(min(100,n)) self.bars["ENERGY"].setValue(60) self.bars["DREAM"].setValue(20) self.bars["VOICE"].setValue(10 if self.config.get("voice","enabled",default=False) else 0) self.bars["MODEL"].setValue(30 if self.config.get("internet_force",default=False) else 8) def append(self, who, msg): self.chatbox.append(f"{who}: {msg}"); self.chat.log(who, msg) def send_chat(self): msg=self.input.text().strip(); if not msg: return self.input.clear(); self.append("Int", msg); reply=self.chat.respond(msg); self.append("E8i", reply) def dragEnterEvent(self,e): if e.mimeData().hasUrls(): e.acceptProposedAction() def dropEvent(self,e): urls=e.mimeData().urls(); if urls: self.compress_path(Path(urls[0].toLocalFile())) def ingest(self): fn,_=QFileDialog.getOpenFileName(self,"Ingest raw file") if fn: self.compress_path(Path(fn)) def compress_path(self,p:Path): self.info.setText(f"Compressing {p} ...") w=Worker(build_stages,p,self.paths.repository,True); self.current_worker=w w.done.connect(self.compress_done); w.failed.connect(self.worker_failed); w.start() def compress_done(self,result): self.last_manifest=result; self.info.setText(json.dumps(result,ensure_ascii=False,indent=2)[:8000]); self.append("E8i", f"compression complete: .sd0 winner {result.get('sd0_winner')}") def worker_failed(self,msg): self.info.setText(msg); self.append("E8i", "worker failed; see informatics panel") def verify_last(self): if not self.last_manifest: self.append("E8i","no last manifest yet"); return ok=[] for k,v in self.last_manifest["stage_files"].items(): p=Path(v["path"]); SDO.decode(p.read_bytes()); ok.append(k) self.append("E8i", "verified stages: "+", ".join(ok)) def run_doctor(self): d=doctor(self.paths,self.config); self.info.setText(json.dumps(d,ensure_ascii=False,indent=2)[:10000]); self.append("E8i","doctor complete") def veil_focus(self): self.append("E8i", f"veil set to {self.veil.value()} seconds; SEND BOOK will honour this value") def send_book(self): self.info.setText("SEND BOOK exporting read-only super-column...") w=Worker(send_book_export,self.paths.repository_zip,self.paths.exportory,self.veil.value(),162,0); self.current_worker=w w.done.connect(lambda r:(self.info.setText(json.dumps(r,ensure_ascii=False,indent=2)), self.append("E8i", "SEND BOOK exported"))) w.failed.connect(self.worker_failed); w.start() def open_repo(self): self.append("E8i", str(self.paths.repository)) def copy_stage(self,stage): if not self.last_manifest: self.append("E8i", "no compressed file yet"); return p=Path(self.last_manifest["stage_files"][stage]["path"]) dest,_=QFileDialog.getSaveFileName(self, f"Save {stage}", p.name) if dest: shutil.copy2(p,dest); self.append("E8i", f"copied {stage} to {dest}") app=QApplication(sys.argv) w=Main(); w.show() return app.exec() # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def run_self_test() -> Dict[str, Any]: samples = [ b"hello 8i" * 30, bytes(range(256)) * 3, ("004:Y" + "🕒" * 20 + "Y\n010:Y?**¤↩=~ text\n").encode("utf-8"), ] results = [] for idx, raw in enumerate(samples): gr = grammar_report() assert gr["basket_roundtrip"] and gr["dna5_roundtrip"] tmp = Path(tempfile.mkdtemp(prefix="e8i_selftest_")) inp = tmp / f"sample{idx}.bin" inp.write_bytes(raw) manifest = build_stages(inp, tmp / "REPOSITORY", add_to_zip=True) for stage, meta in manifest["stage_files"].items(): restored = SDO.decode(Path(meta["path"]).read_bytes()) assert restored == raw sb = send_book_export(tmp / "REPOSITORY" / "REPOSITORY.zip", tmp / "EXPORTORY", veil_seconds=0, max_rows=200) results.append({"sample": idx, "raw_len": len(raw), "sd0_winner": manifest["sd0_winner"], "send_book": sb["output"]}) shutil.rmtree(tmp, ignore_errors=True) return {"status": "PASS", "program": VERSION, "results": results} def build_parser() -> argparse.ArgumentParser: ap = argparse.ArgumentParser(description="8i / V.A.S.K. Total Absolution Functional DNA") ap.add_argument("--root", default=None, help="runtime root folder; default current folder") ap.add_argument("--self-test", action="store_true", help="run roundtrip tests") sub = ap.add_subparsers(dest="cmd") s = sub.add_parser("scan"); s.add_argument("input") c = sub.add_parser("compress"); c.add_argument("input"); c.add_argument("--out-dir", default=None); c.add_argument("--no-zip", action="store_true") d = sub.add_parser("decode"); d.add_argument("input"); d.add_argument("--out", default="decoded.out") m = sub.add_parser("meta"); m.add_argument("input") sb = sub.add_parser("send-book"); sb.add_argument("--repository-zip", default=None); sb.add_argument("--exportory", default=None); sb.add_argument("--veil", type=int, default=DEFAULT_VEIL_SECONDS); sb.add_argument("--width", type=int, default=162); sb.add_argument("--max-rows", type=int, default=0) bt = sub.add_parser("bundle-txt"); bt.add_argument("inputs", nargs="+"); bt.add_argument("--out", default="E8I_LOSSLESS_BUNDLE.txt") ex = sub.add_parser("extract-txt-bundle"); ex.add_argument("bundle"); ex.add_argument("--out-dir", default="E8I_LOSSLESS_EXTRACTED") ch = sub.add_parser("chat"); ch.add_argument("prompt", nargs="*") sub.add_parser("doctor") sub.add_parser("gui") sub.add_parser("bootstrap") sub.add_parser("write-readme2") sub.add_parser("list-models") ag = sub.add_parser("agent-scan"); ag.add_argument("path", nargs="?", default="."); ag.add_argument("--out", default=None) apn = sub.add_parser("agent-plan"); apn.add_argument("prompt", nargs="*") sub.add_parser("grammar-check") return ap def main(argv: Optional[List[str]] = None) -> int: ap = build_parser() args = ap.parse_args(argv) paths = E8iPaths.resolve(args.root) config = Config(paths) write_organ_registry(paths) if args.self_test: print(json.dumps(run_self_test(), ensure_ascii=False, indent=2)) return 0 if args.cmd == "scan": p=Path(args.input); print(json.dumps(scan_bytes(p.read_bytes(), p.name), ensure_ascii=False, indent=2)); return 0 if args.cmd == "compress": p=Path(args.input); out=Path(args.out_dir) if args.out_dir else paths.repository print(json.dumps(build_stages(p, out, add_to_zip=not args.no_zip), ensure_ascii=False, indent=2)); return 0 if args.cmd == "decode": raw=SDO.decode(Path(args.input).read_bytes()); Path(args.out).write_bytes(raw); print(f"decoded {len(raw)} bytes -> {args.out}"); return 0 if args.cmd == "meta": print(json.dumps(SDO.metadata(Path(args.input).read_bytes()), ensure_ascii=False, indent=2)); return 0 if args.cmd == "send-book": repo=Path(args.repository_zip) if args.repository_zip else paths.repository_zip exp=Path(args.exportory) if args.exportory else paths.exportory print(json.dumps(send_book_export(repo, exp, args.veil, args.width, args.max_rows), ensure_ascii=False, indent=2)); return 0 if args.cmd == "bundle-txt": print(json.dumps(combine_files_to_lossless_txt([Path(x) for x in args.inputs], Path(args.out)), ensure_ascii=False, indent=2)); return 0 if args.cmd == "extract-txt-bundle": print(json.dumps(extract_lossless_txt(Path(args.bundle), Path(args.out_dir)), ensure_ascii=False, indent=2)); return 0 if args.cmd == "chat": chat=LocalChat(paths,config); prompt=" ".join(args.prompt).strip() or sys.stdin.read(); reply=chat.respond(prompt); print(reply); return 0 if args.cmd == "doctor": print(json.dumps(doctor(paths,config), ensure_ascii=False, indent=2)); return 0 if args.cmd == "gui": return launch_gui(paths,config) if args.cmd == "bootstrap": print(bootstrap_text()); return 0 if args.cmd == "write-readme2": print(f"wrote {write_readme2(paths)}"); return 0 if args.cmd == "list-models": print(json.dumps(MODEL_TARGETS, ensure_ascii=False, indent=2)); return 0 if args.cmd == "agent-scan": r=agent_scan(Path(args.path).resolve()); if args.out: atomic_write(Path(args.out), json.dumps(r, ensure_ascii=False, indent=2).encode("utf-8")) print(json.dumps(r, ensure_ascii=False, indent=2)); return 0 if args.cmd == "agent-plan": prompt=" ".join(args.prompt).strip() plan={"prompt":prompt,"safe_plan":["scan project","run doctor","run self-test","propose patch only","backup before mutation"],"mutation_allowed":False,"dna_semantic_mode":False} print(json.dumps(plan,ensure_ascii=False,indent=2)); return 0 if args.cmd == "grammar-check": print(json.dumps(grammar_report(), ensure_ascii=False, indent=2)); return 0 ap.print_help() return 0 if __name__ == "__main__": raise SystemExit(main())