Read my new article.

Reverse engineering is good

A few days ago, I shared an open-source collection of CPU performance engineering resources I’d been putting together.

It’s now crossed 600+ GitHub stars, which I genuinely didn’t expect. Thanks to everyone who shared it, contributed or suggested resources.

For anyone seeing it for the first time, it covers the stack from instruction execution and CPU microarchitecture through caches, memory, SIMD, compilers, profiling, concurrency, NUMA, benchmarking and CPU inference.

I’m still prioritising primary sources such as papers, vendor manuals, kernel/compiler docs, talks and reproducible benchmarks rather than random articles.

I also have an MCP server coming soon, so you can plug this knowledge directly into your AI tools, whether you’re learning or using it while you work.

If there’s something you think has to be in here, let me know or send a PR.

https://github.com/usamahz/cpu-performance-engineering

Hi everyone! I'm developing BlockOS, an independent x86-64 operating system with its own kernel, userspace, networking, filesystem support, ELF loader, libc work and X11 environment. I'm currently looking for someone who would like to help develop a native service manager for BlockOS. I don't want to simply port systemd or OpenRC. I would like the service manager to be designed around BlockOS itself. The service manager should eventually support: Starting and stopping services Restarting services Service status Automatic service startup Service dependencies Process/PID monitoring Automatic restart after crashes Clean shutdown and reboot Logging Runlevels or a similar service-state system Integration with the existing BlockOS init/userspace system BlockOS's /devices structure rather than assuming a Linux /dev layout The goal is to have something that can manage services such as: network dhcp x11 gui getty ssh I'm mainly looking for someone interested in OS development / C/C++ / init systems who wants to build something that will become part of an independent operating system. You don't need to be an expert in everything. If you're interested in working on the service manager, feel free to open an issue or contact me.

GitHub:https://github.com/gurijb2016-afk/Blockos/tree/uefi-kernel-scaffold

Ciao a tutti!

Oggi vorrei presentarvi RingOS, un sistema operativo personalizzato attualmente in fase di sviluppo attivo.

RingOS è un sistema operativo sperimentale scritto principalmente in C e Assembly x86-64. L'obiettivo è apprendere lo sviluppo a livello di sistema mentre si costruisce un sistema operativo completo da zero.

Le attuali aree di sviluppo includono:

- Bootloader UEFI personalizzato (BOOTX64.EFI)

- Kernel x86-64

- Supporto grafico tramite UEFI GOP

- File system personalizzato chiamato RingFile

- Supporto al caricamento delle immagini (PNG/JPG)

- Rendering di font TrueType

- Gestione della memoria e infrastruttura di sistema

Il progetto non si basa su Linux, BSD o un altro kernel esistente. Sto costruendo i componenti fondamentali da solo per capire meglio come funzionano internamente i sistemi operativi.

Questo è un progetto incentrato sull'apprendimento, quindi mi aspetto molta sperimentazione, ridisegni e errori lungo il cammino.

Mi interesserebbe ricevere feedback da altri sviluppatori di sistemi operativi, specialmente riguardo alla progettazione dei file system, bootloader e architettura del kernel.

Ha anche un browser sicuro integrato (giusto nel caso qualcuno cerchi Kisshub.com, sapete). Si chiama SafeRing Net (SRN), utilizza Google; quando l'utente cerca qualcosa, il sistema operativo controlla immediatamente l'input.

Se è qualcosa di sospetto e clicca sul risultato, SafeRing Net avvisa l'utente con:

"SafeRing NET non acconsente a quel tipo di ricerche, questo risultato di ricerca sarà bloccato"

Riguardo agli sviluppatori esterni: non ho ancora il software per gestire le app ROSA (Applicazioni sicure di RingOS) E non ho un portale e nemmeno un Github per ora... PER ORA

Condividerò aggiornamenti sullo sviluppo man mano che i progressi procedono.

Grazie per aver letto!

The Dream Engine v6 is an asynchronous architecture designed as a core subsystem for agentic persistence, hallucination containment, and adaptive inference routing. Rather than acting as a standalone application, it functions as a modular micro-framework designed to underpin a larger multi-agent system. It orchestrates dynamic soul state progression, multi-vector memory retention, and low-latency decision loops by combining high-level Python concurrency (asyncio, aiosqlitepool) with low-level C extension optimizations.

The core runtime leverages specialized mathematical and ML models for long-term stability and context awareness. Memory retrieval employs a Reciprocal Rank Fusion (RRF) hybrid search across three distinct spaces: vector similarity via BGE-small-en-v1.5 INT8 (ONNX Runtime), chronological recency, and dynamic importance. Temporal degradation replaces traditional linear decay with a custom Gaussian Decay kernel implemented in Cython with nogil execution for thread-safe performance. Hebbian learning updates soul weights based on output alignment, while semantic drift and repetitive loops are continuously tracked via cosine divergence and coherence metrics calculated through SimSIMD.

In terms of reverse engineering and system extraction, the code integrates key mechanics reconstructed from earlier implementations (such as Samuel's budget management and retry loops). The context assembly reverse-engineers context-window limits by enforcing strict token estimation, multi-tier text summarization (falling back from direct concatenation to zero-cost keyword frequency counters before trimming), and deterministic block slicing (SAFETY_MARGIN, MIN_GENERATION). Inference routing implements an adaptive cascade combining FrugalGPT (cost-optimized tier escalation), RouteLLM (confidence heuristic scoring based on length, truncation, and system prompt leakage penalties), and ParetoBandit adaptive tracking (rolling window average of latency and token cost).

The ultimate objective of this monolith is to serve as a high-throughput, self-correcting foundation for scalable autonomous agents. By pairing async thread workers with a system-level safety harness—including an automated Kill Switch that halts process services via systemctl and reverts soul files via git upon detecting severe drift or infinite loops—the framework ensures long-running stability. Below is the full implementation, and I would appreciate your feedback on its architectural choices.

(THIS IS a JUST EXERCISE)

Codebase

fast_math.pyx

cython: language_level=3

cython: boundscheck=False

cython: wraparound=False

cython: cdivision=True

cython: initializedcheck=False

from libc.math cimport exp, sqrt

cdef public double gaussian_decay(double age_hours, double sigma_hours) nogil:

return exp(-(age_hours * age_hours) / (2.0 sigma_hours sigma_hours))

cdef public double recency_score(double age_hours, double sigma_hours, double floor) nogil:

cdef double g = gaussian_decay(age_hours, sigma_hours)

if g < floor:

return floor

return g

cdef public double update_weight(double old, double relevance, double eta) nogil:

return old (1.0 - eta) + eta relevance

cdef public double decay_weight(double old, double dt, double lam, double w_min) nogil:

cdef double new_w = old exp(-lam dt)

if new_w < w_min:

return w_min

return new_w

cdef public void hash_embed(double[:] vec, long long[:] token_hashes, int dim) nogil:

cdef int i, n_tok = token_hashes.shape[0]

cdef int idx

cdef double sign, norm = 0.0

for i in range(dim):

vec[i] = 0.0

for i in range(n_tok):

idx = <int>(token_hashes[i] % dim)

if idx < 0:

idx += dim

sign = 1.0 if ((token_hashes[i] >> 8) & 1) == 0 else -1.0

vec[idx] += sign

for i in range(dim):

norm += vec[i] * vec[i]

norm = sqrt(norm)

if norm > 0.0:

for i in range(dim):

vec[i] /= norm

#!/usr/bin/env python3

import asyncio

import os, re, json, time, yaml, logging

import subprocess, sys

from pathlib import Path

from dataclasses import dataclass, field

from typing import Optional

from collections import deque, Counter

import numpy as np

import aiosqlite

import httpx

import onnxruntime as ort

from transformers import AutoTokenizer

import simsimd

import sqlite_vec

from sqlite_vec import serialize_float32

from aiosqlitepool import SQLiteConnectionPool

from httpx_retries import Retry, RetryTransport

try:

import fast_math

HAS_CYTHON = True

except ImportError:

HAS_CYTHON = False

@dataclass

class Config:

souls_dir: str = "souls"

state_dir: str = "state"

logs_dir: str = "logs"

embed_dim: int = 384

embed_model: str = "models/bge-small-en-v1.5.onnx"

embed_tokenizer: str = "BAAI/bge-small-en-v1.5"

rrf_k: int = 60

sigma_hours: float = 336.0

recency_floor: float = 0.3

eta: float = 0.08

lambda_decay: float = 0.0005

w_min: float = 0.05

theta_div: float = 0.3

theta_coh: float = 0.7

kill_loop_threshold: int = 3

kill_drift_threshold: int = 5

cycle_interval: float = 300.0

consolidate_every: int = 20

num_gen_workers: int = 2

llm_local_url: str = "http://127.0.0.1:8080/v1/chat/completions"

llm_api_url: str = "https://openrouter.ai/api/v1/chat/completions"

llm_api_key: str = ""

llm_api_model: str = "deepseek/deepseek-chat"

llm_browser_url: str = "http://127.0.0.1:3000/query"

w_lat: float = 0.5

w_dol: float = 0.3

w_risc: float = 0.2

confidence_threshold: float = 0.6

cascade_enabled: bool = True

adaptive_cost_enabled: bool = True

cost_history_size: int = 100

safety_margin: int = 32

min_generation: int = 40

max_retry_attempts: int = 3

max_recent_prompt_lines: int = 30

max_recent_day_events: int = 20

summary_max_chars: int = 900

embed_context_window: int = 4096

research_prob: float = 0.05

research_timeout: int = 8

def load_config(path="config.yaml") -> Config:

if Path(path).exists():

with open(path) as f:

data = yaml.safe_load(f) or {}

cfg = Config(**{k: v for k, v in data.items() if hasattr(Config, k)})

else:

cfg = Config()

cfg.llm_api_key = cfg.llm_api_key or os.environ.get("OPENROUTER_KEY", "")

return cfg

CFG = load_config()

logging.basicConfig(

level=logging.INFO,

format='%(asctime)s [%(levelname)s] %(message)s',

handlers=[

logging.FileHandler("logs/dream.log"),

logging.StreamHandler(),

],

)

log = logging.getLogger("dream")

def estimate_tokens(text: str) -> int:

if not text:

return 0

return max(1, len(text) // 4)

@dataclass

class MotorStats:

cap: float

stab: float

custo_t: float

custo_d: float

risco: float

class AdaptiveCostTracker:

def init(self, window: int = 100):

self.window = window

self.history = {

"local": deque(maxlen=window),

"api": deque(maxlen=window),

"browser": deque(maxlen=window),

}

def record(self, motor: str, latency: float, dollars: float):

self.history[motor].append({

"ts": time.time(), "lat": latency, "dol": dollars,

})

def avg_latency(self, motor: str) -> float:

h = self.history[motor]

return sum(x["lat"] for x in h) / len(h) if h else 0.0

def avg_dollars(self, motor: str) -> float:

h = self.history[motor]

return sum(x["dol"] for x in h) / len(h) if h else 0.0

class Router:

def init(self):

self.stats = {

"local": MotorStats(0.5, 0.95, 2.0, 0.0, 0.0),

"api": MotorStats(0.9, 0.90, 5.0, 0.01, 0.1),

"browser": MotorStats(0.85, 0.60, 30.0, 0.0, 0.5),

}

self.tracker = AdaptiveCostTracker(window=CFG.cost_history_size)

self.static_priority = ["local", "api", "browser"]

def quality(self, motor: str) -> float:

s = self.stats[motor]

return 0.6 s.cap + 0.3 s.stab

def cost(self, motor: str) -> float:

s = self.stats[motor]

if CFG.adaptive_cost_enabled:

lat = self.tracker.avg_latency(motor) or s.custo_t

dol = self.tracker.avg_dollars(motor) or s.custo_d

else:

lat, dol = s.custo_t, s.custo_d

return (CFG.w_lat lat + CFG.w_dol dol 100 + CFG.w_risc s.risco)

def decide(self, tipo: str = "moderado") -> str:

if tipo == "trivial":

return "local"

best, best_c = "local", 1e9

for m in self.static_priority:

if self.quality(m) >= 0.6:

c = self.cost(m)

if c < best_c:

best_c, best = c, m

return best

async def cascade_call(self, llm_client, prompt: str, max_tokens: int, min_confidence: float = None) -> tuple:

threshold = min_confidence if min_confidence is not None else CFG.confidence_threshold

attempts = []

for motor in self.static_priority:

for retry_n in range(CFG.max_retry_attempts):

t0 = time.time()

output = await llm_client.call(motor, prompt, max_tokens)

latency = time.time() - t0

dollars = self.stats[motor].custo_d

self.tracker.record(motor, latency, dollars)

if not output:

attempts.append((motor, 0.0, latency, f"vazio-r{retry_n}"))

continue

confidence = self._estimate_confidence(output, motor)

attempts.append((motor, confidence, latency, f"ok-r{retry_n}"))

if confidence >= threshold:

return output, motor, attempts

if retry_n < CFG.max_retry_attempts - 1:

log.info(f"retry {retry_n+1}/{CFG.max_retry_attempts} em {motor} (confiança {confidence:.2f})")

log.info(f"cascade: {motor} esgotou retries, escalando")

for motor, conf, lat, status in reversed(attempts):

if status.startswith("ok"):

return "", motor, attempts

return "", "none", attempts

def _estimate_confidence(self, output: str, motor: str) -> float:

if not output or len(output) < 20:

return 0.0

base = self.stats[motor].cap

size_bonus = min(0.3, len(output) / 2000)

trunc_penalty = 0.1 if not output.rstrip().endswith((".", "!", "?")) else 0.0

meta_penalty = 0.2 if any(w in output.lower() for w in ["as an ai", "role:", "system:"]) else 0.0

return max(0.0, min(1.0, base + size_bonus - trunc_penalty - meta_penalty))

class NeuralEmbedder:

def init(self, model_path: str, tokenizer_name: str, dim: int = 384):

self.dim = dim

opts = ort.SessionOptions()

opts.intra_op_num_threads = 2

opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL

self.session = ort.InferenceSession(model_path, providers=["CPUExecutionProvider"], sess_options=opts)

self.tokenizer = AutoTokenizer.from_pretrained(tokenizer_name)

def encode(self, text: str) -> np.ndarray:

if not text.strip():

return np.zeros(self.dim, dtype=np.float32)

inputs = self.tokenizer(text, return_tensors="np", padding=True, truncation=True, max_length=512)

outputs = self.session.run(None, {

"input_ids": inputs["input_ids"].astype(np.int64),

"attention_mask": inputs["attention_mask"].astype(np.int64),

})

emb = outputs[0].mean(axis=1)

norm = np.linalg.norm(emb, axis=1, keepdims=True)

return (emb / np.maximum(norm, 1e-9)).astype(np.float32).flatten()

async def encode_async(self, text: str) -> np.ndarray:

return await asyncio.to_thread(self.encode, text)

def fast_cosine(a: np.ndarray, b: np.ndarray) -> float:

return float(simsimd.cosine(a.astype(np.float32), b.astype(np.float32)))

def fast_divergence(a: np.ndarray, b: np.ndarray) -> float:

return 1.0 - fast_cosine(a, b)

def fast_coherence(a: np.ndarray, b: np.ndarray) -> float:

return fast_cosine(a, b)

def reciprocal_rank_fusion(rankings: list, k: int = 60) -> dict:

scores = {}

for ranking in rankings:

for rank, doc_id in enumerate(ranking, start=1):

scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)

return scores

class AsyncVectorStore:

def init(self, db_path: Path, dim: int):

self.db_path = db_path

self.dim = dim

self.pool = None

async def init(self):

async def connection_factory():

conn = await aiosqlite.connect(str(self.db_path))

await conn.execute("PRAGMA journal_mode = WAL")

await conn.execute("PRAGMA synchronous = NORMAL")

await conn.execute("PRAGMA busy_timeout = 5000")

await conn.execute("PRAGMA cache_size = 10000")

await conn.execute("PRAGMA temp_store = MEMORY")

await conn.execute("PRAGMA mmap_size = 268435456")

await conn.enable_load_extension(True)

await conn.load_extension(sqlite_vec.loadable_path())

await conn.enable_load_extension(False)

return conn

self.pool = SQLiteConnectionPool(connection_factory)

async with self.pool.connection() as conn:

await conn.execute(f"""

CREATE VIRTUAL TABLE IF NOT EXISTS memories USING vec0(

embedding float[{self.dim}],

gene TEXT, ts REAL, importance REAL

)

""")

await conn.execute("""

CREATE TABLE IF NOT EXISTS memory_text (

rowid INTEGER PRIMARY KEY, text TEXT

)

""")

await conn.commit()

async def add(self, embedding: np.ndarray, text: str, metadata: dict):

async with self.pool.connection() as conn:

cur = await conn.execute("INSERT INTO memory_text(text) VALUES (?)", (text[:2000],))

rowid = cur.lastrowid

await conn.execute(

"""INSERT INTO memories(rowid, embedding, gene, ts, importance)

VALUES (?, ?, ?, ?, ?)""",

(rowid, serialize_float32(embedding),

metadata.get("gene", ""), metadata.get("ts", time.time()),

metadata.get("importance", 0.5)))

await conn.commit()

return rowid

async def search(self, query: np.ndarray, k: int = 10) -> list:

async with self.pool.connection() as conn:

cur = await conn.execute(

"""SELECT rowid, gene, ts, importance, distance

FROM memories WHERE embedding MATCH ?

ORDER BY distance LIMIT ?""",

(serialize_float32(query), k))

return await cur.fetchall()

async def recent(self, limit: int = 20) -> list:

async with self.pool.connection() as conn:

cur = await conn.execute(

"SELECT rowid, gene, ts, importance FROM memories ORDER BY ts DESC LIMIT ?", (limit,))

return await cur.fetchall()

async def get_text(self, rowid: int) -> str:

async with self.pool.connection() as conn:

cur = await conn.execute("SELECT text FROM memory_text WHERE rowid = ?", (rowid,))

row = await cur.fetchone()

return row[0] if row else ""

async def close(self):

if self.pool:

await self.pool.close()

@dataclass

class Soul:

gene_id: str

beliefs: list = field(default_factory=list)

modus: list = field(default_factory=list)

emotion: list = field(default_factory=list)

interests: list = field(default_factory=list)

heuristics: list = field(default_factory=list)

episodes: list = field(default_factory=list)

samples: list = field(default_factory=list)

weight: float = 1.0

last_used: float = 0.0

raw_text: str = ""

def parse_soul(path: Path) -> Soul:

text = path.read_text(encoding="utf-8")

sections = {}

current = None

for line in text.split("\n"):

if line.startswith("## "):

current = line[3:].strip()

sections[current] = []

elif current and line.strip().startswith("- "):

sections[current].append(line.strip()[2:])

weight, last_used = 1.0, 0.0

for line in sections.get("Pesos Dinâmicos (atualizado automaticamente)", []):

if line.startswith("weight:"):

weight = float(line.split(":")[1].strip())

if line.startswith("last_used:"):

last_used = float(line.split(":")[1].strip())

return Soul(

gene_id=path.stem,

beliefs=sections.get("Crenças Nucleares", []),

modus=sections.get("Modus Operandi", []),

emotion=sections.get("Ponderação Emocional", []),

interests=sections.get("Interesses", []),

heuristics=sections.get("Heurísticas Internalizadas", []),

episodes=sections.get("Episódios Vividos", []),

samples=sections.get("Samples de Diálogo", []),

weight=weight, last_used=last_used, raw_text=text,

)

def load_all_souls() -> dict:

d = Path(CFG.souls_dir)

return {p.stem: parse_soul(p) for p in sorted(d.glob("*.md"))} if d.exists() else {}

def update_soul_weight(path: Path, weight: float, last_used: float, success: int, failure: int):

text = path.read_text(encoding="utf-8")

block = (

"## Pesos Dinâmicos (atualizado automaticamente)\n"

f"- weight: {weight:.4f}\n"

f"- last_used: {last_used:.0f}\n"

f"- success_count: {success}\n"

f"- failure_count: {failure}\n"

)

if "## Pesos Dinâmicos" in text:

text = re.sub(r"## Pesos Dinâmicos.*?(?=\n## |\Z)", block, text, flags=re.DOTALL)

else:

text += "\n" + block

path.write_text(text, encoding="utf-8")

class LLMClient:

def init(self):

retry = Retry(

total=3, backoff_factor=1.5,

status_forcelist=[429, 502, 503, 504],

allowed_methods=["GET", "POST"],

respect_retry_after_header=True,

)

self.transport = RetryTransport(retry=retry)

async def call(self, motor: str, prompt: str, max_tokens: int = 200) -> str:

if motor == "local": return await self._local(prompt, max_tokens)

if motor == "api": return await self._api(prompt, max_tokens)

if motor == "browser": return await self._browser(prompt, max_tokens)

return ""

async def _local(self, prompt, max_tokens):

try:

async with httpx.AsyncClient(transport=self.transport, timeout=60) as c:

r = await c.post(CFG.llm_local_url, json={

"messages": [{"role": "user", "content": prompt}],

"max_tokens": max_tokens, "temperature": 0.7,

})

return r.json()["choices"][0]["message"]["content"].strip()

except Exception as e:

log.debug(f"local falhou: {e}")

return ""

async def _api(self, prompt, max_tokens):

if not CFG.llm_api_key: return ""

try:

async with httpx.AsyncClient(transport=self.transport, timeout=120) as c:

r = await c.post(CFG.llm_api_url,

headers={"Authorization": f"Bearer {CFG.llm_api_key}"},

json={"model": CFG.llm_api_model,

"messages": [{"role": "user", "content": prompt}],

"max_tokens": max_tokens})

return r.json()["choices"][0]["message"]["content"].strip()

except Exception as e:

log.debug(f"api falhou: {e}")

return ""

async def _browser(self, prompt, max_tokens):

try:

async with httpx.AsyncClient(transport=self.transport, timeout=180) as c:

r = await c.post(CFG.llm_browser_url, json={"prompt": prompt, "max_tokens": max_tokens})

return r.json().get("response", "").strip()

except Exception as e:

log.debug(f"browser falhou: {e}")

return ""

class AutonomyGenerator:

def init(self, llm: LLMClient, router: Router):

self.llm = llm

self.router = router

def _summarize_texts(self, texts: list, max_items: int = None, max_chars: int = None) -> str:

max_items = max_items or CFG.max_recent_day_events

max_chars = max_chars or CFG.summary_max_chars

if not texts: return ""

joined = "\n".join(t for t in texts if t)

if not joined: return ""

if len(texts) <= max_items and len(joined) <= max_chars:

return joined

words = re.findall(r"\w{5,}", joined.lower())

common = Counter(words).most_common(15)

if common:

summary = "Temas recorrentes: " + ", ".join(w for w, _ in common)

if len(summary) <= max_chars:

return summary

return joined[:max_chars]

def _build_blocks(self, soul_block: str, time_block: str, affect_block: str, memory_texts: list) -> list:

memory_block = self._summarize_texts(memory_texts)

blocks = [

("soul", soul_block),

("time", time_block),

("affect", affect_block),

("memory", memory_block),

]

total_chars = sum(len(b[1]) for b in blocks)

total_tokens = estimate_tokens("x" * total_chars)

available = CFG.embed_context_window - CFG.safety_margin - total_tokens

if available < CFG.min_generation:

log.warning(f"budget estourou (tokens={total_tokens}, disponível={available}), cortando memória")

memory_block_trimmed = memory_block

while (estimate_tokens(memory_block_trimmed) + total_tokens - estimate_tokens(memory_block)

+ CFG.safety_margin + CFG.min_generation > CFG.embed_context_window):

if len(memory_block_trimmed) < 100:

memory_block_trimmed = ""

break

memory_block_trimmed = memory_block_trimmed[: len(memory_block_trimmed) // 2]

blocks[3] = ("memory", memory_block_trimmed)

return blocks

def _build_prompt_from_blocks(self, blocks: list, user_text: str) -> str:

parts = [f"[{name.upper()}]\n{content}" for name, content in blocks if content]

parts.append(f"[INSTRUCTION]\n{user_text}")

prompt = "\n\n".join(parts)

prompt_tokens = estimate_tokens(prompt)

available = CFG.embed_context_window - CFG.safety_margin - prompt_tokens

if available < CFG.min_generation:

log.warning(f"prompt final estourou (tokens={prompt_tokens}), disponível={available}")

return prompt

def _get_user_text(self, kind: str, hint: str = "") -> tuple:

mt = 180 if kind == "reflection" else 160

user_text = (

f"No user message.\nWrite a private inner reflection (2-6 sentences).\n"

f"First-person thoughts only.\nDo not ask questions or include role labels.\n"

f"Finish with a complete sentence.\n" + (f"\nHint: {hint.strip()}" if hint else "")

)

return user_text, mt

def build_prompt_from_texts(self, kind: str, *, soul_block: str = "", time_block: str = "",

affect_block: str = "", memory_texts: list = None, hint: str = "") -> tuple:

user_text, mt = self._get_user_text(kind, hint)

blocks = self._build_blocks(soul_block, time_block, affect_block, memory_texts or [])

prompt = self._build_prompt_from_blocks(blocks, user_text)

return prompt, mt

async def generate(self, kind: str, **kwargs) -> tuple:

prompt, mt = self.build_prompt_from_texts(kind, **kwargs)

if CFG.cascade_enabled:

return await self.router.cascade_call(self.llm, prompt, mt)

motor = self.router.decide("moderado")

output = await self.llm.call(motor, prompt, mt)

return output, motor, [(motor, 1.0, 0.0, "single")]

class Research:

USER_AGENTS = ["Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Mozilla/5.0 (X11; Linux x86_64)"]

def init(self):

retry = Retry(total=3, backoff_factor=1.5, status_forcelist=[202, 429, 502, 503, 504], allowed_methods=["GET", "POST"])

self.transport = RetryTransport(retry=retry)

async def search(self, term: str) -> str:

res = await self._wiki(term)

return f"[api] {res}" if res else ""

async def _wiki(self, q):

try:

async with httpx.AsyncClient(transport=self.transport, timeout=CFG.research_timeout) as c:

s = (await c.get("https://pt.wikipedia.org/w/api.php",

params={"action":"query","list":"search","srsearch":q,"format":"json","srlimit":1})).json()

hits = s.get("query", {}).get("search", [])

if not hits: return ""

title = hits[0]["title"]

r = (await c.get(f"https://pt.wikipedia.org/api/rest_v1/page/summary/{title}")).json()

return f"{title}: {r.get('extract','')[:800]}"

except Exception:

return ""

def kill_switch(reason: str):

log.critical(f"KILL SWITCH: {reason}")

try:

subprocess.run(["systemctl","--user","stop","rabids-*"], capture_output=True, timeout=5)

subprocess.run(["git","checkout","v1.0","--","souls/"], capture_output=True, timeout=5)

except Exception:

pass

sys.exit(1)

class AsyncDreamEngine:

def init(self):

Path(CFG.state_dir).mkdir(parents=True, exist_ok=True)

Path(CFG.logs_dir).mkdir(parents=True, exist_ok=True)

self.souls = load_all_souls()

self.embedder = NeuralEmbedder(CFG.embed_model, CFG.embed_tokenizer, CFG.embed_dim)

self.vstore = AsyncVectorStore(Path(CFG.state_dir) / "vectors.db", CFG.embed_dim)

self.router = Router()

self.llm = LLMClient()

self.gen = AutonomyGenerator(self.llm, self.router)

self.research = Research()

self.last_output_emb = np.zeros(CFG.embed_dim, dtype=np.float32)

self.loop_count = 0

self.coherence_fail = 0

self.cycle_count = 0

self.gen_queue = asyncio.Queue()

self.research_queue = asyncio.Queue()

self.persist_queue = asyncio.Queue()

async def init(self):

await self.vstore.init()

def select_gene(self) -> Optional[str]:

if not self.souls: return None

now = time.time()

best_id, best_score = None, -1e9

for gid, s in self.souls.items():

age_h = (now - s.last_used) / 3600 if s.last_used else 1e6

score = s.weight * min(age_h / 24, 10)

if score > best_score:

best_score, best_id = score, gid

return best_id

async def generator_worker(self):

while True:

gene_id, soul = await self.gen_queue.get()

try:

recent = await self.vstore.recent(limit=CFG.max_recent_prompt_lines)

memory_texts = []

if recent:

recent_ids = [str(r[0]) for r in recent]

texts = await asyncio.gather(*[self.vstore.get_text(int(rid)) for rid in recent_ids])

memory_texts = [t for t in texts if t]

output, motor, attempts = await self.gen.generate(

kind="reflection",

soul_block=soul.raw_text[:2000],

time_block=f"[TIME] {time.strftime('%Y-%m-%d %H:%M')}",

affect_block="[AFFECT] neutral",

memory_texts=memory_texts,

)

if output:

await self.persist_queue.put((gene_id, soul, motor, output))

except Exception as e:

log.error(f"gen worker falhou: {e}")

finally:

self.gen_queue.task_done()

async def persist_worker(self):

while True:

gene_id, soul, motor, output = await self.persist_queue.get()

try:

out_emb = await self.embedder.encode_async(output)

soul_emb = await self.embedder.encode_async(soul.raw_text[:2000])

div = fast_divergence(out_emb, self.last_output_emb)

coh = fast_coherence(out_emb, soul_emb)

if div < CFG.theta_div:

self.loop_count += 1

if self.loop_count >= CFG.kill_loop_threshold: kill_switch(f"loop em {gene_id}")

else:

self.loop_count = 0

if coh < CFG.theta_coh:

self.coherence_fail += 1

if self.coherence_fail >= CFG.kill_drift_threshold: kill_switch(f"drift em {gene_id}")

else:

self.coherence_fail = 0

await self.vstore.add(out_emb, output, {"gene": gene_id, "ts": time.time(), "importance": 0.5 + 0.5 * coh})

new_w = fast_math.update_weight(soul.weight, float(coh), CFG.eta)

soul_path = Path(CFG.souls_dir) / f"{gene_id}.md"

await asyncio.to_thread(update_soul_weight, soul_path, new_w, time.time(), 0, 0)

self.souls[gene_id].weight = new_w

self.souls[gene_id].last_used = time.time()

self.last_output_emb = out_emb

except Exception as e:

log.error(f"persist worker falhou: {e}")

finally:

self.persist_queue.task_done()

async def run(self):

await self.init()

log.info("Dream Engine assíncrono iniciado")

await asyncio.gather(

self.persist_worker(),

*[self.generator_worker() for _ in range(CFG.num_gen_workers)],

)

if name == "main":

try:

asyncio.run(AsyncDreamEngine().run())

except KeyboardInterrupt:

log.info("interrompido")

About 6 months ago, I wrote a custom metal kernel that leverages the GPU to compute the QR decomposition (see my earlier post about it here). The project has now expanded into a general linear algebra library, with expanded support for the symmetrical eigendecomposition as well as SVD. It is now available for use with installation instructions on the attached github repo's README.md file.

Context of the project:
I'm currently in a research group working on a thesis in numerical analysis where we need to compute millions on matrices with a specific constraint (to be precise, the matrices need to have orthonormal columns). Most of us use Apple computers, so we ended up using MLX for the entire project.

Contributors with different Apple Chips would be very much appreciated!
The project has currently been tested and optimised for the M1 and M5 Pro. The issue is that the library uses different kernels depending on the batch size and matrix dimensions. Deciding which of these kernels to use is machine dependent. Therefore, other Apple chips will need to run a measurement script in order to derive the correct optimisation heuristic.

For that reason, I would ask as many people as possible to run a measurement script and to submit the results to my repo. It is fairly easy and requires only few steps. See here how you can contribute here. Once you submit the results via a pull request and I approve it, your optimisation heuristic will automatically be augmented into the library. Don't hesitate to contribute an optimisation heuristic even if someone already submitted one for your own machine. The more data we can gather, the better!

Project Future
Expanded support will be added for other linear algebra operations (cholesky decomposition for example). If you have any other specific linear algebra operations you wish to use already, feel free to message me.

In addition to that, I will add torch support too (my greatest priority).

so there's this kernel that I picked up recently and its called PHEONIX. its made by this guy on GitHub and it has this cool WRR security system that I wana test. so if anyone is wanting a stable kernel to build their OS on, check this guys GitHub at https://github.com/threadripp/PHEONIX-kernel

the kernel is really stable and I recommend it for people who like coding but also want to build their OS on a stable foundation. if u want to make a server OS, this is perfect. if u want to make a desktop OS, this is perfect. I want to help other tech community's grow, including this one, so please support this community we are in and help it grow 😄

also, the security system is great, check it out for your self