Modelpublic

Shuiquan-Future-Lab/Shuiquan-Quantum-Memory-Cloud-V20

sourceHugging Faceapache-2.0updated 9mo agoView on Hugging Face

1likes

Model Card

#!/usr/bin/env python3

quantummemorycloudprodfull_projection.py

生产级单文件：Quantum Memory Cloud（完整版，含 Detox + Micro↔Macro 投影）

说明：

- 单文件交付，自动检测可选依赖（faiss, redis, fastapi, prometheus, numba）

- 在无可选依赖时自动降级为纯 numpy/内存实现

- 主要模块：FusionCore, StorageCore, ProjectionEngine (Micro->Macro, Macro->Micro), Ledger, Detox, HTTP 管理端点（可选）

运行：

- headless: python quantummemorycloudprodfull_projection.py

- 启用 HTTP: export QMCENABLEHTTP=1 && python ...

环境变量（常用）:

QMCDIM, QMCUSEFAISS, QMCUSEREDIS, QMCENABLEHTTP, QMCENABLEPROM, QMCENTCAP, QMCFAISSBATCH, QMCBG_INTERVAL

建议：先在测试环境跑通，再在生产环境启用 FAISS/Redis/Prometheus。

import os, sys, time, uuid, json, math, random, hashlib, threading, traceback, queue, atexit, signal from dataclasses import dataclass, field from typing import Dict, Any, List, Optional, Tuple import numpy as np from concurrent.futures import ThreadPoolExecutor, as_completed

-------------------------

Optional imports detection

-------------------------

FASTAPIAVAILABLE = False PROMAVAILABLE = False HASFAISS = False HASREDIS = False HAS_NUMBA = False

try: from fastapi import FastAPI, HTTPException from fastapi.responses import Response, JSONResponse FASTAPIAVAILABLE = True except Exception: FASTAPIAVAILABLE = False

try: from prometheusclient import Counter, Histogram, generatelatest, CONTENTTYPELATEST, Gauge PROMAVAILABLE = True except Exception: PROMAVAILABLE = False

try: import faiss HASFAISS = True except Exception: faiss = None HASFAISS = False

try: import redis HASREDIS = True except Exception: redis = None HASREDIS = False

try: from numba import njit HASNUMBA = True except Exception: njit = None HASNUMBA = False

-------------------------

Config (env override)

-------------------------

DIM = int(os.environ.get("QMCDIM", 128)) FAISSINDEXPATH = os.environ.get("QMCFAISSINDEX", "qmcfaiss.index") FAISSUSE = HASFAISS and os.environ.get("QMCUSEFAISS", "1") == "1" REDISUSE = HASREDIS and os.environ.get("QMCUSEREDIS", "0") == "1" ENABLEHTTP = FASTAPIAVAILABLE and os.environ.get("QMCENABLEHTTP", "0") == "1" ENABLEPROM = PROMAVAILABLE and os.environ.get("QMCENABLEPROM", "0") == "1" NUMBAUSE = HASNUMBA and os.environ.get("QMCUSENUMBA", "0") == "1"

ENTCAPACITY = int(os.environ.get("QMCENTCAP", 1024)) FAISSBATCH = int(os.environ.get("QMCFAISSBATCH", 128)) PROMOTECOST = float(os.environ.get("QMCPROMOTECOST", 0.12)) POCKETMAXLOCAL = int(os.environ.get("QMCPOCKETMAXLOCAL", 16)) CONSOLIDATIONBATCH = int(os.environ.get("QMCCONSOLIDATIONBATCH", 64)) BACKGROUNDREBUILDINTERVAL = float(os.environ.get("QMCBGINTERVAL", 5.0)) QUARANTINEHOLD = float(os.environ.get("QMCQUARANTINEHOLD", 60.0)) LEDGERPATH = os.environ.get("QMCLEDGER", "quantummemorycloudproductionledger.json") LOGPREFIX = os.environ.get("QMCLOGPREFIX", "[QuantumMemoryCloudProdProj]") DEMOUNITS = int(os.environ.get("QMCDEMOUNITS", 512)) AUTOPROJECTBATCH = int(os.environ.get("QMCAUTOPROJECTBATCH", 16)) PREFETCHENABLED = os.environ.get("QMCPREFETCH", "1") == "1" MAXPROJECTWORKERS = int(os.environ.get("QMCPROJECTWORKERS", 8)) TOKENBUCKETRATE = float(os.environ.get("QMCTOKENRATE", 100.0)) TOKENBUCKETCAP = float(os.environ.get("QMCTOKEN_CAP", 200.0))

Detox & Projection config

TOXICITYTHRESHOLD = float(os.environ.get("QMCTOXICITYTHRESHOLD", 0.6)) REPAIRTHRESHOLD = float(os.environ.get("QMCREPAIRTHRESHOLD", 0.65)) ANOMALYZSCORE = float(os.environ.get("QMCANOMALYZSCORE", 4.0)) DECAYINTERVAL = float(os.environ.get("QMCDECAYINTERVAL", 3600.0)) DECAYRATE = float(os.environ.get("QMCDECAYRATE", 0.01)) SANITIZERBLACKLIST = os.environ.get("QMCSANITIZERBLACKLIST", "").split(",") if os.environ.get("QMCSANITIZERBLACKLIST") else []

Projection engine config

PROJMICRODIM = int(os.environ.get("QMCPROJMICRODIM", 64)) # micro subspace dim PROJMACRODIM = int(os.environ.get("QMCPROJMACRODIM", 32)) # macro representation dim PROJHIGHDIM = int(os.environ.get("QMCPROJHIGHDIM", 256)) # optional high-dim latent PROJTOKENRATE = float(os.environ.get("QMCPROJTOKENRATE", 10.0)) # tokens/sec for heavy reverse projection PROJTOKENCAP = float(os.environ.get("QMCPROJTOKEN_CAP", 20.0))

-------------------------

Utilities

-------------------------

def uid() -> str: return str(uuid.uuid4())

def now_ts() -> str: return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

def sha256_hex(s: str) -> str: return hashlib.sha256(s.encode('utf-8')).hexdigest()

def log(msg: str): print(f"{LOGPREFIX} [{nowts()}] {msg}", flush=True)

-------------------------

Metrics (Prometheus or no-op)

-------------------------

if ENABLEPROM: METPROJECT = Counter("qmcprojecttotal", "Total PROJECT calls") METQUERY = Counter("qmcquerytotal", "Total pocketquery calls") METFAISSSEARCH = Counter("qmcfaisssearchtotal", "Total FAISS searches") METENTHIT = Counter("qmcenthittotal", "Entanglement cache hits") METENTPUT = Counter("qmcentputtotal", "Entanglement cache puts") METPOCKETPUT = Counter("qmcpocketputtotal", "Total pocketput calls") METDELETE = Counter("qmcdeletetotal", "Total delete requests") LATPROJECT = Histogram("qmcprojectlatencyseconds", "PROJECT latency seconds") LATQUERY = Histogram("qmcquerylatencyseconds", "pocketquery latency seconds") GAUGEUNITS = Gauge("qmcunits", "Number of memory units") else: class Dummy: def inc(self, a, k): pass def observe(self, a, **k): pass METPROJECT = METQUERY = METFAISSSEARCH = METENTHIT = METENTPUT = METPOCKETPUT = METDELETE = Dummy() LATPROJECT = LATQUERY = Dummy() GAUGEUNITS = _Dummy()

-------------------------

Ledger (batched writer + snapshot)

-------------------------

class Ledger: chain: List[Dict[str, Any]] = [] lock = threading.Lock() writeq = queue.Queue() _stop = False

@classmethod def record(cls, op: str, objid: str, info: Dict[str, Any]): try: with cls.lock: prev = cls.chain[-1]['hash'] if cls.chain else '' entry = {"ts": nowts(), "op": op, "id": objid, "info": info, "prev": prev} s = json.dumps(entry, sortkeys=True, ensureascii=False) entry['hash'] = sha256hex(s) cls.chain.append(entry) cls.writeq.put(entry) except Exception as e: log(f"Ledger.record error: {e}")

@classmethod def writerworker(cls, path: str = LEDGERPATH): buffer = [] lastflush = time.time() while not cls.stop: try: item = cls.writeq.get(timeout=1.0) buffer.append(item) if len(buffer) >= 64 or (time.time() - lastflush) > 5.0: cls.flushbuffer(buffer, path) buffer = [] lastflush = time.time() except queue.Empty: if buffer: cls.flushbuffer(buffer, path) buffer = [] lastflush = time.time() except Exception as e: log(f"Ledger writer error: {e}") time.sleep(0.5) if buffer: cls.flushbuffer(buffer, path)

@classmethod def flushbuffer(cls, buffer: List[Dict[str, Any]], path: str): try: tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(cls.chain, f, ensureascii=False, indent=2) os.replace(tmp, path) log(f"Ledger flushed {len(buffer)} entries to {path}") except Exception as e: log(f"Ledger.flush_buffer error: {e}")

@classmethod def dump(cls, path: str = LEDGERPATH): try: with cls.lock: with open(path, "w", encoding="utf-8") as f: json.dump(cls.chain, f, ensureascii=False, indent=2) log(f"Ledger dumped to {path}") except Exception as e: log(f"Ledger.dump error: {e}")

@classmethod def startwriter(cls, path: str = LEDGERPATH): t = threading.Thread(target=cls.writerworker, args=(path,), daemon=True) t.start() return t

@classmethod def stopwriter(cls): cls.stop = True

Ledger.startwriter(LEDGERPATH) atexit.register(lambda: (Ledger.dump(), Ledger.stop_writer()))

-------------------------

Data models

-------------------------

@dataclass class MemoryUnit: id: str embedding: np.ndarray xi: float trit: int = 0 importance: float = 0.0 emotion: float = 0.0 coreprotected: bool = False shards: List[str] = field(defaultfactory=list) quarantined: bool = False deleterequester: Optional[str] = None deleterequestts: Optional[float] = None version: int = 0 ts: float = field(defaultfactory=time.time) lastactive: float = field(defaultfactory=time.time) decay_score: float = 0.0 explain: Optional[str] = None

@dataclass class Hologram: id: str embedding: np.ndarray confidence: float provenance: Dict[str, Any] deltaE: float toxicscore: float = 0.0 explain: Optional[str] = None

-------------------------

Sanitizer

-------------------------

class Sanitizer: blacklist = set([w.strip() for w in SANITIZER_BLACKLIST if w.strip()])

@staticmethod def clean_text(b: bytes) -> bytes: try: s = b.decode('utf-8', errors='ignore') s = "".join(ch for ch in s if ord(ch) >= 32) for bad in Sanitizer.blacklist: if bad and bad in s: s = s.replace(bad, "[REDACTED]") return s.encode('utf-8') except Exception: return b

-------------------------

Detox utilities

-------------------------

class Detox: @staticmethod def toxicity_score(emb: np.ndarray, background: Optional[np.ndarray] = None) -> float: try: mag = float(np.linalg.norm(emb)) mean = float(np.mean(emb)) std = float(np.std(emb)) + 1e-12 kurt = float(np.mean(((emb - mean) / std) * 4)) score = min(1.0, (abs(mean) / 10.0) 0.4 + (mag / (np.sqrt(len(emb)) + 1e-12)) 0.3 + (min(kurt / 3.0, 1.0)) 0.3) if background is not None: dist = np.linalg.norm(emb - background) / (np.linalg.norm(background) + 1e-12) score = min(1.0, score + min(dist, 1.0) * 0.2) return float(score) except Exception: return 0.0

@staticmethod def isanomalous(emb: np.ndarray, background: np.ndarray, zthreshold: float = ANOMALYZSCORE) -> bool: try: diff = emb - background z = np.abs((diff - np.mean(diff)) / (np.std(diff) + 1e-12)) return bool(np.any(z > zthreshold)) except Exception: return False

-------------------------

Token bucket

-------------------------

class TokenBucket: def _init_(self, rate: float, capacity: float): self.rate = rate self.capacity = capacity self.tokens = capacity self.lock = threading.Lock() self.last = time.time() def consume(self, amount: float = 1.0) -> bool: with self.lock: now = time.time() self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate) self.last = now if self.tokens >= amount: self.tokens -= amount return True return False

-------------------------

Projection Engine

-------------------------

class ProjectionEngine: """ Provides:

—microtomacro(emb): fast online projection (low-latency)
—macrotomicro(macro_repr): async heavy reverse projection (rate-limited)
—highdimprojection: optional high-dim latent mapping
—state machine for switching modes """ def _init(self, dim: int = DIM, microdim: int = PROJMICRODIM, macrodim: int = PROJMACRODIM, highdim: int = PROJHIGHDIM): self.dim = dim self.microdim = microdim self.macrodim = macrodim self.highdim = highdim # random projection matrices (orthonormal-ish) for fast mapping rng = np.random.RandomState(42) self.microproj = rng.normal(scale=1.0, size=(self.microdim, self.dim)).astype('float32') self.macroproj = rng.normal(scale=1.0, size=(self.macrodim, self.microdim)).astype('float32') self.highproj = rng.normal(scale=1.0, size=(self.highdim, self.dim)).astype('float32') # small PCA-like running stats for micro->macro adapt self.micromean = np.zeros(self.microdim, dtype='float32') self.microcount = 0 # reverse projection threadpool and token bucket self.reverseexecutor = ThreadPoolExecutor(maxworkers=4) self.reversetokenbucket = TokenBucket(PROJTOKENRATE, PROJTOKENCAP) # storage reference set by app self.storage = None # provenance ledger Ledger.record("PROJECTIONINIT", uid(), {"dim": dim, "microdim": microdim, "macrodim": macrodim, "highdim": high_dim}) log("ProjectionEngine initialized")

def attach_storage(self, storage): self.storage = storage

def microtomacro(self, emb: np.ndarray) -> Dict[str, Any]: """ Fast online micro->macro:

—project to micro subspace
—normalize and project to macro dim
—return macro vector + metadata """ m = self.microproj.dot(emb) # running mean update (cheap) self.microcount += 1 if self.microcount % 100 == 0: self.micromean = 0.99 self.micro_mean + 0.01 np.mean(m, axis=0) # macro mapping macro = self.macroproj.dot(np.tanh(m)) # normalize norm = np.linalg.norm(macro) + 1e-12 macro = macro / norm meta = {"method": "microtomacro", "micronorm": float(np.linalg.norm(m)), "macronorm": float(norm)} Ledger.record("PROJMICROTOMACRO", uid(), {"meta": meta}) return {"macro": macro.astype('float32'), "meta": meta}

def macrotomicroasync(self, macrorepr: np.ndarray, callback=None, priority: int = 0) -> Optional[str]: """ Schedule an async reverse projection (macro->micro). Rate-limited by token bucket. Returns task id or None if rejected. """ if not self.reversetokenbucket.consume(1.0): Ledger.record("PROJREVERSEREJECT", uid(), {"reason": "ratelimit"}) return None taskid = uid() self.reverseexecutor.submit(self.reverseworker, taskid, macrorepr, callback, priority) Ledger.record("PROJREVERSESCHEDULED", taskid, {"priority": priority}) return task_id

def reverseworker(self, taskid: str, macrorepr: np.ndarray, callback, priority: int): """ Heavy reverse projection:

—find nearest macro neighbors in ent cache / faiss
—reconstruct micro candidates by linear combination + noise
—optionally refine by local optimization (few steps)
—store synthetic MemoryUnit in quarantine or return via callback """ try: # step 1: find candidate memory ids via fusion/faiss candidates = [] if self.storage and self.storage.fusion: # use macrorepr to query ent cache by projecting back to dim (approx) # approximate back-projection using pseudo-inverse of macroproj micro_proj approx_micro = np.linalg.pinv(self.macro_proj).dot(macro_repr) approx_full = np.linalg.pinv(self.micro_proj).dot(approx_micro) # use faiss search on approx_full if available hits = [] try: hits = self.storage.fusion.faiss_search(approx_full, topk=8) except Exception: hits = [] candidates = [self.storage.index[h] for h in hits if h in self.storage.index] # step 2: synthesize micro candidates synths = [] if candidates: base = np.mean(np.stack(candidates, axis=0), axis=0) for i in range(min(6, len(candidates))): alpha = 0.6 + 0.4 random.random() noise = np.random.normal(scale=1e-3, size=base.shape) cand = alpha base + (1 - alpha) noise synths.append(cand.astype('float32')) else: # fallback: expand macro via highproj pseudo-inverse approxfull = np.linalg.pinv(self.highproj).dot(np.concatenate([macrorepr, np.zeros(self.highdim - len(macrorepr))])[:self.highdim]) for i in range(4): noise = np.random.normal(scale=1e-2, size=approxfull.shape) synths.append((approxfull + noise).astype('float32')) # step 3: optional local refinement (few gradient-free steps) refined = [] for s in synths: # small smoothing toward background if self.storage and self.storage.background is not None: s = 0.7 * s + 0.3 * self.storage.background refined.append(s) # step 4: create synthetic MemoryUnit(s) in quarantine for review created = [] for s in refined: memid = uid() mu = MemoryUnit(id=memid, embedding=s.copy(), xi=0.0) mu.quarantined = True mu.explain = f"reverseprojtask:{taskid}" # store in hot for inspection but quarantined self.storage.hot[memid] = mu self.storage.index[memid] = mu.embedding.copy() self.storage.quarantine[memid] = {"snapshothash": sha256hex(memid + ":" + str(time.time())), "expirets": time.time() + QUARANTINEHOLD, "requester": "reverseproj", "reason": "reverseprojection", "taskid": taskid} created.append(memid) Ledger.record("PROJREVERSECREATED", memid, {"taskid": taskid}) # callback with created ids if callback: try: callback(taskid, created) except Exception: pass log(f"ProjectionEngine: reverse task {taskid} created {len(created)} synthetic units") except Exception as e: log(f"ProjectionEngine: reverseworker error: {e}")

def highdimproject(self, emb: np.ndarray) -> np.ndarray: # optional high-dim latent mapping (cheap linear projection + tanh) h = np.tanh(self.high_proj.dot(emb)) return h.astype('float32')

-------------------------

FusionCore (FAISS + Entanglement + Prefetch + Projection hooks)

-------------------------

class FusionCore: def _init(self, dim: int = DIM): self.dim = dim self.usefaiss = FAISSUSE self.faissindex = None self.faissids: List[str] = [] self.faissbuffer: List[np.ndarray] = [] self.faissbufferids: List[str] = [] self.faisslock = threading.Lock() self.stop = False

# ent cache self.entcapacity = ENTCAPACITY self.entcache: Dict[str, Dict[str, Any]] = {} self.entlru: List[str] = [] self.enthitthreshold = 0.85 self.ent_lock = threading.Lock()

# redis optional self.useredis = REDISUSE and HASREDIS self.redisclient = None if self.useredis: try: self.redisclient = redis.Redis() self.redisclient.ping() log("FusionCore: Redis connected for ent/quarantine") except Exception as e: log(f"FusionCore: Redis init failed: {e}") self.redisclient = None self.use_redis = False

# prefetch queue self.prefetchq = queue.Queue(maxsize=2048) self.prefetchenabled = PREFETCH_ENABLED

# threadpool for project batch self.projectexecutor = ThreadPoolExecutor(maxworkers=MAXPROJECTWORKERS)

# init faiss if self.usefaiss: try: if os.path.exists(FAISSINDEXPATH): self.faissindex = faiss.readindex(FAISSINDEXPATH) log("FusionCore: FAISS index loaded") else: self.faissindex = faiss.IndexHNSWFlat(dim, 32) self.faissindex.hnsw.efConstruction = 64 self.faissindex.hnsw.efSearch = 64 log("FusionCore: FAISS index created") except Exception as e: log(f"FusionCore: FAISS init error: {e}") self.usefaiss = False self.faissindex = None

# background workers self.flushthread = threading.Thread(target=self.faissflushworker, daemon=True) self.flushthread.start() self.prefetchthread = threading.Thread(target=self.prefetchworker, daemon=True) self.prefetch_thread.start()

# adaptive stats self.enthits = 0 self.entqueries = 0 self.adaptlock = threading.Lock() self.storage = None Ledger.record("FUSIONCOREINIT", uid(), {"dim": dim, "faiss": self.usefaiss, "redis": self.useredis}) log("FusionCore initialized (projection-ready)")

def attach_storage(self, storage): self.storage = storage

# FAISS buffered add def faissaddbuffered(self, memid: str, emb: np.ndarray): if not self.usefaiss or self.faissindex is None: with self.faisslock: self.faissids.append(memid) self.faissbuffer.append(emb.astype('float32').copy()) return with self.faisslock: self.faissbuffer.append(emb.astype('float32').reshape(1, -1)) self.faissbufferids.append(memid) if len(self.faissbuffer) >= FAISSBATCH: self.flushfaiss_buffer()

def flushfaissbuffer(self): if not self.usefaiss or self.faissindex is None: return try: vecs = np.vstack(self.faissbuffer) self.faissindex.add(vecs) self.faissids.extend(self.faissbufferids) Ledger.record("FAISSBATCHADD", uid(), {"count": len(self.faissbufferids)}) log(f"FusionCore: FAISS batch add {len(self.faissbufferids)}") except Exception as e: log(f"FusionCore: FAISS batch add error: {e}") finally: self.faissbuffer = [] self.faissbuffer_ids = []

def faissflushworker(self): while not self.stop: try: time.sleep(1.0) with self.faisslock: if self.faissbuffer: self.flushfaissbuffer() if self.usefaiss and self.faissindex is not None: try: faiss.writeindex(self.faissindex, FAISSINDEXPATH) except Exception as e: log(f"FusionCore: FAISS persist error: {e}") except Exception as e: log(f"FusionCore: faissflush_worker error: {e}") time.sleep(1.0)

def faisssearch(self, qemb: np.ndarray, topk: int = 5) -> List[str]: if self.usefaiss and self.faissindex is not None and len(self.faissids) > 0: try: q = qemb.astype('float32').reshape(1, -1) D, I = self.faissindex.search(q, topk) res = [] for idx in I[0]: if idx < 0 or idx >= len(self.faissids): continue res.append(self.faissids[int(idx)]) METFAISSSEARCH.inc() Ledger.record("FAISSSEARCH", uid(), {"topk": topk, "found": len(res)}) return res except Exception as e: log(f"FusionCore: faisssearch error: {e}") with self.faisslock: if not self.faissbuffer and not self.faissids: return [] return self.faiss_ids[:topk]

# ent cache with redis optional def entget(self, qemb: np.ndarray): key = sha256hex(",".join(map(str, np.round(qemb[:8], 3).tolist())))[:16] with self.entlock: self.entqueries += 1 e = self.entcache.get(key) if not e and self.useredis and self.redisclient: try: raw = self.redisclient.get("ent:" + key) if raw: obj = json.loads(raw) e = {"entvec": np.array(obj["entvec"], dtype='float32'), "memids": obj["memids"], "lastaccess": time.time(), "xireserve": obj.get("xireserve", 0.05)} self.entcache[key] = e self.entlru.insert(0, key) except Exception: pass if not e: return None entvec = e['entvec'] sim = self.cosinesim(entvec, qemb) if sim < self.enthitthreshold: return None e['lastaccess'] = time.time() if key in self.entlru: self.entlru.remove(key) self.entlru.insert(0, key) self.enthits += 1 METENTHIT.inc() Ledger.record("ENTHIT", key, {"sim": float(sim)}) self.maybe_adapt() return e

def entput(self, memembeddings: List[np.ndarray], memids: List[str], xireserve: float = 0.05): if not memembeddings: return None entvec = np.mean(np.stack(memembeddings, axis=0), axis=0).astype('float32') key = sha256hex(",".join(map(str, np.round(entvec[:8], 4).tolist())))[:16] with self.entlock: if key in self.entcache: self.entcache[key].update({"entvec": entvec, "memids": memids, "lastaccess": time.time(), "xireserve": xireserve}) if key in self.entlru: self.entlru.remove(key) self.entlru.insert(0, key) else: if len(self.entlru) >= self.entcapacity: tail = self.entlru.pop() self.entcache.pop(tail, None) if self.useredis and self.redisclient: try: self.redisclient.delete("ent:" + tail) except Exception: pass self.entcache[key] = {"entvec": entvec, "memids": memids, "lastaccess": time.time(), "xireserve": xireserve} self.entlru.insert(0, key) if self.useredis and self.redisclient: try: self.redisclient.set("ent:" + key, json.dumps({"entvec": entvec.tolist(), "memids": memids, "xireserve": xireserve}), ex=3600) except Exception: pass METENTPUT.inc() Ledger.record("ENTPUT", key, {"count": len(mem_ids)}) return key

# project with toxicity check and projection hooks def project(self, queryemb: np.ndarray, background: Optional[np.ndarray] = None, alpha: float = 0.6, projectionmode: str = "micro"): """ projectionmode: "micro" (default) uses micro->macro fast mapping for additional context, "macro" uses macro-level aggregation, "high" uses high-dim latent mixing. """ start = time.time() METPROJECT.inc() # quick toxicity pre-check on query tscore = Detox.toxicityscore(queryemb, background) # projection engine hooks if attached projmeta = {} if hasattr(self, "projectionengine") and self.projectionengine: if projectionmode == "micro": pm = self.projectionengine.microtomacro(queryemb) projmeta = pm["meta"] elif projectionmode == "macro": pm = self.projectionengine.microtomacro(queryemb) projmeta = pm["meta"] elif projectionmode == "high": h = self.projectionengine.highdimproject(queryemb) projmeta = {"highdimnorm": float(np.linalg.norm(h))} # ent cache ent = self.entget(queryemb) if ent is not None: entvec = ent['entvec'] emb = alpha * queryemb + (1.0 - alpha) ent_vec delta_E = float(np.linalg.norm(emb - ent_vec)) holo = Hologram(id=uid(), embedding=emb, confidence=0.92, provenance={"method": "entangled", "proj": projection_mode}, delta_E=delta_E, toxic_score=tscore) holo.explain = "entangled" LAT_PROJECT.observe(time.time() - start) Ledger.record("PROJECT_ENT", holo.id, {"delta_E": delta_E, "toxic_score": tscore, "proj_meta": proj_meta}) if tscore >= TOXICITY_THRESHOLD: threading.Thread(target=self._handle_toxic_hologram, args=(holo,), daemon=True).start() return holo # faiss entanglement if self.storage and (self.use_faiss or self.faiss_buffer): hits = self.faiss_search(query_emb, topk=6) if hits: mem_embs = [self.storage.index[h] for h in hits if h in self.storage.index] if mem_embs: ent_key = self.ent_put(mem_embs, hits, xi_reserve=0.05) ent_vec = np.mean(np.stack(mem_embs, axis=0), axis=0) emb = alpha queryemb + (1.0 - alpha) * entvec deltaE = float(np.linalg.norm(emb - entvec)) holo = Hologram(id=uid(), embedding=emb, confidence=0.86, provenance={"method": "faissent", "entkey": entkey, "proj": projectionmode}, deltaE=deltaE, toxicscore=tscore) holo.explain = "faissent" LATPROJECT.observe(time.time() - start) Ledger.record("PROJECTFAISS", holo.id, {"deltaE": deltaE, "hits": len(hits), "toxicscore": tscore, "projmeta": projmeta}) if tscore >= TOXICITYTHRESHOLD: threading.Thread(target=self.handletoxichologram, args=(holo,), daemon=True).start() return holo # fallback to background B = background if background is not None else (self.storage.background if self.storage else np.zeros(self.dim, dtype='float32')) emb = alpha * queryemb + (1.0 - alpha) * B emb += np.random.normal(scale=1e-6, size=emb.shape) deltaE = float(np.linalg.norm(emb - B)) holo = Hologram(id=uid(), embedding=emb, confidence=0.72, provenance={"method": "background", "proj": projectionmode}, deltaE=deltaE, toxicscore=tscore) holo.explain = "background" LATPROJECT.observe(time.time() - start) Ledger.record("PROJECTBG", holo.id, {"deltaE": deltaE, "toxicscore": tscore, "projmeta": projmeta}) if tscore >= TOXICITYTHRESHOLD: threading.Thread(target=self.handletoxichologram, args=(holo,), daemon=True).start() return holo

def handletoxichologram(self, holo: Hologram): try: res = self.storage.negentropyread(holo, toxicitythreshold=TOXICITYTHRESHOLD) if res.get("status") == "ok": Ledger.record("TOXICHANDLEDOK", holo.id, {"method": "lightrepair"}) return repaired = res.get("hologram") if repaired and res.get("toxic"): time.sleep(0.2) if Detox.toxicityscore(repaired.embedding, self.storage.background) >= TOXICITYTHRESHOLD: memid = uid() mu = MemoryUnit(id=memid, embedding=repaired.embedding.copy(), xi=0.0) mu.quarantined = True mu.explain = f"autoquarantinefromholo:{holo.id}" self.storage.hot[memid] = mu self.storage.index[memid] = mu.embedding.copy() Ledger.record("AUTOQUARANTINE", memid, {"fromholo": holo.id, "toxicscore": repaired.deltaE}) log(f"FusionCore: Auto-quarantined synthetic unit {memid[:8]} from holo {holo.id[:8]}") except Exception as e: log(f"FusionCore: handletoxic_hologram error: {e}")

def projectbatch(self, queries: List[np.ndarray], background: Optional[np.ndarray] = None, alpha: float = 0.6, projectionmode: str = "micro") -> List[Hologram]: results: List[Optional[Hologram]] = [None] len(queries) for i, q in enumerate(queries): ent = self.ent_get(q) if ent is not None: ent_vec = ent['ent_vec'] emb = alpha q + (1.0 - alpha) * entvec deltaE = float(np.linalg.norm(emb - entvec)) tscore = Detox.toxicityscore(emb, background) results[i] = Hologram(id=uid(), embedding=emb, confidence=0.92, provenance={"method": "entangled"}, deltaE=deltaE, toxicscore=tscore, explain="entangled") futures = {} for i, q in enumerate(queries): if results[i] is None: futures[self.projectexecutor.submit(self.project, q, background, alpha, projectionmode)] = i for fut in ascompleted(futures): i = futures[fut] try: results[i] = fut.result() except Exception as e: log(f"projectbatch worker error: {e}") results[i] = self.project(queries[i], background, alpha, projectionmode) return results

def prefetch(self, emb: np.ndarray): if not self.prefetchenabled: return try: self.prefetchq.put(emb, timeout=0.01) except Exception: pass

def prefetchworker(self): while not self.stop: try: emb = self.prefetchq.get(timeout=1.0) hits = self.faisssearch(emb, topk=8) if hits and self.storage: memembs = [self.storage.index[h] for h in hits if h in self.storage.index] if memembs: self.entput(memembs, hits, xireserve=0.02) except queue.Empty: continue except Exception as e: log(f"FusionCore: prefetch_worker error: {e}") time.sleep(0.2)

def cosinesim(self, a: np.ndarray, b: np.ndarray) -> float: an = np.linalg.norm(a) + 1e-12 bn = np.linalg.norm(b) + 1e-12 return float(np.dot(a, b) / (an * bn))

def maybeadapt(self): with self.adaptlock: if self.entqueries >= 200: hitrate = self.enthits / max(1, self.entqueries) if hitrate > 0.6 and self.enthitthreshold > 0.6: self.enthitthreshold = max(0.5, self.enthitthreshold - 0.02) elif hitrate < 0.2 and self.enthitthreshold < 0.95: self.enthitthreshold = min(0.95, self.enthitthreshold + 0.02) self.enthits = 0 self.ent_queries = 0

def shutdown(self): self.stop = True log("FusionCore: shutdown requested") with self.faisslock: if self.faissbuffer: self.flushfaissbuffer() if self.usefaiss and self.faissindex is not None: try: faiss.writeindex(self.faissindex, FAISSINDEXPATH) log("FusionCore: FAISS index persisted on shutdown") except Exception as e: log(f"FusionCore: FAISS persist error on shutdown: {e}") try: self.project_executor.shutdown(wait=False) except Exception: pass

-------------------------

StorageCore (Detox + Decay + Quarantine + Projection integration)

-------------------------

class StorageCore: def _init(self, dim: int = DIM, fusion: FusionCore = None, projectionengine: ProjectionEngine = None): self.dim = dim self.hot: Dict[str, MemoryUnit] = {} self.near: Dict[str, MemoryUnit] = {} self.shards: Dict[str, bytes] = {} self.index: Dict[str, np.ndarray] = {} self.quarantine: Dict[str, Dict[str, Any]] = {} self.pagetable: Dict[str, Dict[str, Any]] = {} self.localcache: Dict[str, str] = {} self.xipool: float = 1.0 self.maxlocal = POCKETMAXLOCAL self.consolidationq = queue.Queue() self.rebuildevent = threading.Event() self.rebuildlock = threading.Lock() self.background = np.zeros(dim, dtype='float32') self.stop = False self.fusion = fusion if self.fusion: self.fusion.attachstorage(self) self.projectionengine = projectionengine if self.projectionengine: self.projectionengine.attachstorage(self) # workers self.consolidationthread = threading.Thread(target=self.consolidationworker, daemon=True) self.backgroundthread = threading.Thread(target=self.backgroundworker, daemon=True) self.decaythread = threading.Thread(target=self.decayworker, daemon=True) self.consolidationthread.start() self.backgroundthread.start() self.decaythread.start() self.tokenbucket = TokenBucket(TOKENBUCKETRATE, TOKENBUCKETCAP) Ledger.record("STORAGECOREINIT", uid(), {"dim": dim}) log("StorageCore initialized (projection-ready)")

# storage primitives def putunit(self, unit: MemoryUnit, hot: bool = True, near: bool = True): unit.version += 1 unit.ts = time.time() unit.lastactive = time.time() if hot: self.hot[unit.id] = unit if near: self.near[unit.id] = unit self.index[unit.id] = unit.embedding.copy() self.incrementalbackgroundupdate(unit) if self.fusion: self.fusion.faissaddbuffered(unit.id, unit.embedding) METPOCKETPUT.inc() Ledger.record("PUTUNIT", unit.id, {"xi": unit.xi, "coreprotected": unit.coreprotected, "explain": unit.explain}) try: GAUGEUNITS.set(len(self.index)) if ENABLEPROM else None except Exception: pass

def putshard(self, sid: str, payload: bytes, xi: float, trit: int): self.shards[sid] = payload Ledger.record("PUTSHARD", sid, {"xi": xi, "trit": trit})

def retrieveunit(self, memid: str) -> Optional[MemoryUnit]: u = self.hot.get(memid) or self.near.get(memid) if not u: return None if u.quarantined: return None if memid in self.near and memid not in self.hot: self.hot[memid] = self.near[memid] Ledger.record("PROMOTENEAR", memid, {}) u.lastactive = time.time() return self.hot.get(memid)

def retrieveany(self, memid: str) -> Optional[MemoryUnit]: u = self.hot.get(memid) or self.near.get(memid) if u: u.last_active = time.time() return u

# pocketput with sanitization and toxicity pre-check def pocketput(self, payload: bytes, embedding: np.ndarray, xi: float = 0.5, coreprotect: bool = False, importance: float = 0.0, emotion: float = 0.0): try: cleanpayload = Sanitizer.cleantext(payload) except Exception: cleanpayload = payload memid = uid() unit = MemoryUnit(id=memid, embedding=embedding.copy(), xi=xi, trit=0, importance=importance, emotion=emotion, coreprotected=coreprotect) unit.explain = "sanitizedpayload" sid = uid() self.putshard(sid, cleanpayload, xi, 0) unit.shards = [sid] tscore = Detox.toxicityscore(embedding, self.background) if tscore >= TOXICITYTHRESHOLD: unit.quarantined = True unit.explain = f"quarantinedonput:score={tscore:.3f}" self.hot[memid] = unit self.index[memid] = unit.embedding.copy() self.quarantine[memid] = {"snapshothash": sha256hex(memid + ":" + str(time.time())), "expirets": time.time() + QUARANTINEHOLD, "requester": "auto", "reason": "toxicityonput", "score": tscore} Ledger.record("POCKETPUTQUARANTINED", memid, {"score": tscore}) log(f"StorageCore: Pocket put quarantined {memid[:8]} score={tscore:.3f}") return {"status": "quarantined", "memid": memid, "score": tscore} self.putunit(unit, hot=False, near=True) vaddr = "v:" + memid[:8] self.pagetable[vaddr] = {"memid": memid, "local": False, "lastaccess": time.time()} if len(self.localcache) < self.maxlocal and self.xipool > PROMOTECOST: self.promotetolocal(vaddr) Ledger.record("POCKETPUT", vaddr, {"memid": mem_id}) return {"status": "ok", "vaddr": vaddr}

def pocketquery(self, contextemb: np.ndarray, topk: int = 5, projectionmode: str = "micro"): start = time.time() METQUERY.inc() mids = [] if self.fusion: mids = self.fusion.faisssearch(contextemb, topk=topk) if not mids: if not self.index: return [] ids = list(self.index.keys()) mats = np.stack([self.index[i] for i in ids], axis=0) qn = np.linalg.norm(contextemb) + 1e-12 norms = np.linalg.norm(mats, axis=1) + 1e-12 sims = (mats @ contextemb) / (norms * qn) topidx = np.argsort(-sims)[:topk] mids = [ids[int(i)] for i in topidx] results = [] for mid in mids: u = self.retrieveany(mid) if not u or u.quarantined: continue vaddr = None for va, info in self.pagetable.items(): if info["memid"] == mid: vaddr = va; break if not vaddr: vaddr = "v:" + mid[:8] self.pagetable[vaddr] = {"memid": mid, "local": False, "lastaccess": time.time()} self.pagetable[vaddr]["lastaccess"] = time.time() if len(results) < self.maxlocal: threading.Thread(target=self.promotetolocal, args=(vaddr,), daemon=True).start() results.append({"vaddr": vaddr, "memid": mid, "explain": u.explain}) LATQUERY.observe(time.time() - start) Ledger.record("POCKETQUERY", uid(), {"hits": len(results), "projmode": projection_mode}) return results

def promotetolocal(self, vaddr: str): info = self.pagetable.get(vaddr) if not info: return memid = info["memid"] if self.xipool < PROMOTECOST: Ledger.record("PROMOTEFAIL", vaddr, {"xipool": self.xipool}) return self.xipool -= PROMOTECOST u = self.retrieveany(memid) if u: info["local"] = True self.localcache[vaddr] = memid if len(self.localcache) > self.maxlocal: self.evictone() Ledger.record("PROMOTE", vaddr, {"memid": memid, "xipool": self.xi_pool})

def evictone(self): lru = None; lruts = float('inf') for va, info in self.pagetable.items(): if info.get("local") and info["lastaccess"] < lruts: lru = va; lruts = info["lastaccess"] if lru: self.pagetable[lru]["local"] = False self.localcache.pop(lru, None) Ledger.record("EVICT", lru, {})

# consolidation worker def pushconsolidation(self, memid: str): try: self.consolidationq.put(memid, timeout=0.1) Ledger.record("CONSOLIDATIONPUSH", memid, {}) except Exception: pass

def consolidationworker(self): batch = [] while not self.stop: try: memid = self.consolidationq.get(timeout=1.0) batch.append(memid) if len(batch) >= CONSOLIDATIONBATCH: self.doconsolidationbatch(batch) batch = [] except queue.Empty: if batch: self.doconsolidation_batch(batch) batch = [] except Exception as e: log(f"StorageCore: consolidation worker error: {e}") time.sleep(0.5)

def doconsolidationbatch(self, memids: List[str]): for memid in memids: u = self.retrieveany(memid) if not u or u.quarantined: continue data = ("DETAILS:" + u.id + ":" + str(time.time())).encode('utf-8') chunks = [data[i:i+32] for i in range(0, len(data), 32)] sids = [] for c in chunks: sid = uid() self.putshard(sid, c, u.xi, u.trit) sids.append(sid) u.shards = sids self.putunit(u, hot=False, near=True) Ledger.record("CONSOLIDATIONDONE", u.id, {"shards": len(sids)}) log(f"StorageCore: Consolidation batch done size={len(memids)}") if len(self.index) >= 16 and self.fusion: sampleids = list(self.index.keys())[:min(12, len(self.index))] memembs = [self.index[i] for i in sampleids] self.fusion.entput(memembs, sampleids, xireserve=0.05) self.rebuildevent.set()

# background worker def backgroundworker(self): lastrebuild = 0.0 while not self.stop: try: triggered = self.rebuildevent.wait(timeout=BACKGROUNDREBUILDINTERVAL) with self.rebuildlock: now = time.time() if now - lastrebuild > 0.5: self.rebuildbackgroundfield() lastrebuild = now self.rebuildevent.clear() except Exception as e: log(f"StorageCore: background worker error: {e}") time.sleep(0.5)

def rebuildbackgroundfield(self): if not self.index: self.background = np.zeros(self.dim, dtype='float32') return ids = list(self.index.keys()) mats = np.stack([self.index[i] for i in ids], axis=0) weights = np.array([max(self.retrieveany(i).xi if self.retrieveany(i) else 0.01, 0.01) * (1.0 + (self.retrieveany(i).importance if self.retrieveany(i) else 0.0)) for i in ids]) total = weights.sum() + 1e-12 B = (weights[:, None] * mats).sum(axis=0) / total self.background = B.astype('float32') Ledger.record("BACKGROUNDREBUILD", uid(), {"units": len(ids)}) log("StorageCore: Background field rebuilt")

# negentropyread (enhanced) def negentropyread(self, holo: Hologram, toxicitythreshold: float = TOXICITYTHRESHOLD): meanval = float(np.mean(holo.embedding)) toxicscore = Detox.toxicityscore(holo.embedding, self.background) temp = MemoryUnit(id="temp", embedding=holo.embedding.copy(), xi=0.5) violations = self.checkcorerules(temp) toxic = (toxicscore > toxicitythreshold) or (len(violations) > 0) if not toxic: Ledger.record("NEGENTROPYOK", holo.id, {"toxicscore": toxicscore}) return {"status": "ok", "hologram": holo, "toxic": False} # Stage 1: light repair c = -0.4 * np.sign(holo.embedding) * np.minimum(np.abs(holo.embedding), 0.05) repairedemb = holo.embedding + c deltacomp = float(np.linalg.norm(c)) repaired = Hologram(id=uid(), embedding=repairedemb, confidence=max(0.1, holo.confidence - 0.05), provenance={"repairof": holo.id, "stage": "light"}, deltaE=holo.deltaE + deltacomp) newscore = Detox.toxicityscore(repaired.embedding, self.background) Ledger.record("NEGENTROPYREPAIRSTAGE1", repaired.id, {"orig": holo.id, "deltacomp": deltacomp, "newscore": newscore}) if newscore <= toxicitythreshold: threading.Thread(target=self.asyncvalidaterepair, args=(repaired,), daemon=True).start() return {"status": "repaired", "hologram": repaired, "toxic": False, "deltacomp": deltacomp} # Stage 2: stronger repair (async) stronger = Hologram(id=uid(), embedding=repaired.embedding.copy(), confidence=max(0.05, repaired.confidence - 0.1), provenance={"repairof": holo.id, "stage": "strong"}, deltaE=repaired.deltaE) threading.Thread(target=self.strongrepairandvalidate, args=(stronger, holo.id), daemon=True).start() return {"status": "repairedasync", "hologram": stronger, "toxic": True, "deltacomp": delta_comp}

def strongrepairandvalidate(self, repaired: Hologram, origholoid: str): try: repaired.embedding = 0.5 repaired.embedding + 0.5 self.background newscore = Detox.toxicityscore(repaired.embedding, self.background) Ledger.record("NEGENTROPYREPAIRSTAGE2", repaired.id, {"orig": origholoid, "newscore": newscore}) if newscore > TOXICITYTHRESHOLD: memid = uid() mu = MemoryUnit(id=memid, embedding=repaired.embedding.copy(), xi=0.0) mu.quarantined = True mu.explain = f"autoquarantinefromrepair:{origholoid}" self.hot[memid] = mu self.index[memid] = mu.embedding.copy() self.quarantine[memid] = {"snapshothash": sha256hex(memid + ":" + str(time.time())), "expirets": time.time() + QUARANTINEHOLD, "requester": "autorepair", "reason": "repairfailed", "score": newscore} Ledger.record("AUTOQUARANTINE", memid, {"fromholo": origholoid, "score": newscore}) log(f"StorageCore: Auto-quarantined {memid[:8]} from repair of holo {origholoid[:8]}") else: Ledger.record("NEGENTROPYVALIDATE", repaired.id, {"validated": True}) log(f"StorageCore: Repair validated {repaired.id[:8]}") except Exception as e: log(f"StorageCore: strongrepairandvalidate error: {e}")

def asyncvalidaterepair(self, repaired: Hologram): time.sleep(0.5) Ledger.record("NEGENTROPYVALIDATE", repaired.id, {"validated": True}) log(f"StorageCore: Repair validated {repaired.id[:8]}")

# core rules corerules: Dict[str, Any] = {} def addcorerule(self, ruleid: str, fn, signer: str = "admin"): self.corerules[ruleid] = {"fn": fn, "signer": signer} Ledger.record("CORERULEADD", ruleid, {"signer": signer}) log(f"StorageCore: Core rule added {ruleid}")

def checkcorerules(self, unit: MemoryUnit) -> List[str]: violated = [] for rid, info in self.corerules.items(): try: ok = info"fn" except Exception: ok = False if not ok: violated.append(rid) return violated

# decay worker def decayworker(self): while not self.stop: try: time.sleep(DECAYINTERVAL) now = time.time() todecay = [] for mid, u in list(self.hot.items()): age = now - u.lastactive u.decayscore += DECAYRATE * (age / max(1.0, DECAYINTERVAL)) if u.decayscore > 0.5 and u.importance < 0.1: todecay.append(mid) for mid in todecay: u = self.hot.get(mid) if not u: continue u.xi = max(0.0, u.xi - 0.1) Ledger.record("DECAYAPPLIED", mid, {"decayscore": u.decayscore, "xi": u.xi}) if u.xi <= 0.0 and not u.coreprotected: self.requestselfdelete(mid, requester="decayworker", holdseconds=QUARANTINEHOLD) log(f"StorageCore: Decay pass done, decayed={len(todecay)}") except Exception as e: log(f"StorageCore: decay worker error: {e}") time.sleep(1.0)

# quarantine / delete def requestselfdelete(self, memid: str, requester: str, holdseconds: float = QUARANTINEHOLD): u = self.retrieveany(memid) if not u: return {"status": "notfound"} if u.coreprotected: Ledger.record("DELETEREJECTCORE", memid, {"requester": requester}) return {"status": "rejectedcoreprotected"} snaphash = sha256hex(memid + ":" + str(time.time())) expirets = time.time() + holdseconds self.quarantine[memid] = {"snapshothash": snaphash, "expirets": expirets, "requester": requester, "reason": "selfdelete"} u.quarantined = True u.deleterequester = requester u.deleterequestts = time.time() if memid in self.index: self.index.pop(memid, None) Ledger.record("QUARANTINE", memid, {"requester": requester, "expirets": expirets}) log(f"StorageCore: Quarantined {memid[:8]} by {requester} until {expirets}") threading.Thread(target=self.delayedpermanentdelete, args=(memid, expirets), daemon=True).start() METDELETE.inc() return {"status": "quarantined", "memid": memid, "holduntil": expire_ts}

def undodelete(self, memid: str, requester: str): info = self.quarantine.get(memid) if not info: return {"status": "notquarantined"} if info["requester"] != requester and requester != "admin": return {"status": "notauthorized"} u = self.retrieveany(memid) if not u: return {"status": "unitmissing"} u.quarantined = False u.deleterequester = None u.deleterequestts = None self.index[memid] = u.embedding.copy() self.quarantine.pop(memid, None) Ledger.record("UNDOQUARANTINE", memid, {"requester": requester}) log(f"StorageCore: Undo quarantine {memid[:8]} by {requester}") return {"status": "restored", "memid": memid}

def delayedpermanentdelete(self, memid: str, expirets: float): while time.time() < expirets and not self.stop: time.sleep(0.5) info = self.quarantine.get(memid) if not info: return if time.time() >= info["expirets"]: self.hot.pop(memid, None) self.near.pop(memid, None) self.index.pop(memid, None) self.quarantine.pop(memid, None) Ledger.record("PERMANENTDELETE", memid, {"ts": time.time()}) log(f"StorageCore: Permanently deleted {memid[:8]}")

def incrementalbackgroundupdate(self, unit: MemoryUnit): try: alpha = 1.0 / max(1, len(self.index)) self.background = (1 - alpha) * self.background + alpha * unit.embedding except Exception: self.rebuildevent.set()

def shutdown(self): self._stop = True log("StorageCore: shutdown requested")

-------------------------

Application harness (full)

-------------------------

class QuantumMemoryCloudApp: def _init(self, dim: int = DIM, demounits: int = DEMOUNITS, enablehttp: bool = ENABLEHTTP): self.dim = dim self.demounits = demounits self.projectionengine = ProjectionEngine(dim=dim, microdim=PROJMICRODIM, macrodim=PROJMACRODIM, highdim=PROJHIGHDIM) self.fusion = FusionCore(dim=dim) self.storage = StorageCore(dim=dim, fusion=self.fusion, projectionengine=self.projectionengine) self.fusion.projectionengine = self.projectionengine self.stop = False Ledger.record("APPINIT", uid(), {"dim": dim, "demounits": demounits}) log("Starting quantum memory cloud (production full projection)") # populate demo self.populatethread = threading.Thread(target=self.populatedemo, daemon=True) self.populatethread.start() # auto project batcher self.projectthread = threading.Thread(target=self.autoprojectbatcher, daemon=True) self.projectthread.start() # optional HTTP self.api = None if enablehttp and FASTAPIAVAILABLE: try: self.starthttp() except Exception as e: log(f"HTTP start failed: {e}") # signals signal.signal(signal.SIGINT, self.signalhandler) signal.signal(signal.SIGTERM, self.signalhandler) log("System started (FusionCore + StorageCore + ProjectionEngine)")

def populatedemo(self): log(f"Demo: populating {self.demounits} memory units") for i in range(self.demounits): emb = np.random.normal(scale=1.0, size=(self.dim,)).astype('float32') payload = f"demo-{i}".encode('utf-8') res = self.storage.pocketput(payload, emb, xi=0.5) # occasionally push for consolidation if isinstance(res, dict) and res.get("status") == "ok": vaddr = res["vaddr"] memid = self.storage.pagetable.get(vaddr, {}).get("memid") if memid and i % max(1, CONSOLIDATIONBATCH // 4) == 0: self.storage.pushconsolidation(memid) if PREFETCHENABLED: self.fusion.prefetch(emb) time.sleep(0.002) log("Demo population done") Ledger.record("DEMOPOPULATED", uid(), {"units": self.demounits}) self.storage.rebuildevent.set()

def autoprojectbatcher(self): while not self.stop: try: batch = [] for in range(AUTOPROJECTBATCH): q = np.random.normal(scale=1.0, size=(self.dim,)).astype('float32') batch.append(q) if not self.storage.tokenbucket.consume(len(batch)): time.sleep(0.05) continue # alternate projection modes to exercise micro/macro/high mode = random.choice(["micro", "macro", "high"]) holos = self.fusion.projectbatch(batch, background=self.storage.background, alpha=0.6, projectionmode=mode) for holo in holos: res = self.storage.negentropyread(holo) Ledger.record("AUTOPROJECT", holo.id, {"status": res.get("status"), "conf": holo.confidence, "toxic": holo.toxicscore, "projmode": mode}) time.sleep(max(0.05, BACKGROUNDREBUILDINTERVAL / 8.0)) except Exception as e: log(f"Auto project batcher error: {e}") time.sleep(0.5)

def starthttp(self): app = FastAPI() @app.get("/health") def health(): return {"status": "ok", "time": nowts()} @app.get("/metrics") def metrics(): if ENABLEPROM: return Response(generatelatest(), mediatype=CONTENTTYPELATEST) else: return {"metrics": "disabled"} @app.post("/put") def putitem(payload: Dict[str, Any]): try: text = payload.get("text", "").encode('utf-8') emb = np.array(payload.get("embedding", np.random.normal(size=(self.dim,)).tolist()), dtype='float32') res = self.storage.pocketput(text, emb, xi=float(payload.get("xi", 0.5))) return res except Exception as e: raise HTTPException(statuscode=500, detail=str(e)) @app.post("/query") def queryitem(payload: Dict[str, Any]): try: emb = np.array(payload.get("embedding", np.random.normal(size=(self.dim,)).tolist()), dtype='float32') topk = int(payload.get("topk", 5)) mode = payload.get("projmode", "micro") res = self.storage.pocketquery(emb, topk=topk, projectionmode=mode) return {"hits": res} except Exception as e: raise HTTPException(statuscode=500, detail=str(e)) @app.get("/ledger") def ledgercount(): return {"entries": len(Ledger.chain)} @app.post("/admin/snapshot") def adminsnapshot(): try: Ledger.dump() return {"status": "dumped"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) self.api = app # run uvicorn externally as recommended log("HTTP API initialized (use uvicorn to serve)")

def signalhandler(self, signum, frame): log(f"Signal {signum} received, shutting down") self.shutdown()

def shutdown(self): if self.stop: return self.stop = True try: self.fusion.shutdown() except Exception: pass try: self.storage.shutdown() except Exception: pass try: Ledger.dump() except Exception: pass log("QuantumMemoryCloudApp shutdown complete")

-------------------------

Entrypoint

-------------------------

def main(): app = QuantumMemoryCloudApp(dim=DIM, demounits=DEMOUNITS, enablehttp=ENABLEHTTP) try: while True: time.sleep(1.0) except KeyboardInterrupt: app.shutdown()

if _name == "main_": main()