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

@classmethod
def _writer_worker(cls, path: str = LEDGER_PATH):
    buffer = []
    last_flush = time.time()
    while not cls._stop:
        try:
            item = cls._write_q.get(timeout=1.0)
            buffer.append(item)
            if len(buffer) >= 64 or (time.time() - last_flush) > 5.0:
                cls._flush_buffer(buffer, path)
                buffer = []
                last_flush = time.time()
        except queue.Empty:
            if buffer:
                cls._flush_buffer(buffer, path)
                buffer = []
                last_flush = time.time()
        except Exception as e:
            log(f"Ledger writer error: {e}")
            time.sleep(0.5)
    if buffer:
        cls._flush_buffer(buffer, path)

@classmethod
def _flush_buffer(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, ensure_ascii=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 = LEDGER_PATH):
    try:
        with cls.lock:
            with open(path, "w", encoding="utf-8") as f:
                json.dump(cls.chain, f, ensure_ascii=False, indent=2)
        log(f"Ledger dumped to {path}")
    except Exception as e:
        log(f"Ledger.dump error: {e}")

@classmethod
def start_writer(cls, path: str = LEDGER_PATH):
    t = threading.Thread(target=cls._writer_worker, args=(path,), daemon=True)
    t.start()
    return t

@classmethod
def stop_writer(cls):
    cls._stop = True

@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

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

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

def micro_to_macro(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.micro_proj.dot(emb)
    # running mean update (cheap)
    self.micro_count += 1
    if self.micro_count % 100 == 0:
        self.micro_mean = 0.99 * self.micro_mean + 0.01 * np.mean(m, axis=0)
    # macro mapping
    macro = self.macro_proj.dot(np.tanh(m))
    # normalize
    norm = np.linalg.norm(macro) + 1e-12
    macro = macro / norm
    meta = {"method": "micro_to_macro", "micro_norm": float(np.linalg.norm(m)), "macro_norm": float(norm)}
    Ledger.record("PROJ_MICRO_TO_MACRO", uid(), {"meta": meta})
    return {"macro": macro.astype('float32'), "meta": meta}

def macro_to_micro_async(self, macro_repr: 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.reverse_token_bucket.consume(1.0):
        Ledger.record("PROJ_REVERSE_REJECT", uid(), {"reason": "rate_limit"})
        return None
    task_id = uid()
    self.reverse_executor.submit(self._reverse_worker, task_id, macro_repr, callback, priority)
    Ledger.record("PROJ_REVERSE_SCHEDULED", task_id, {"priority": priority})
    return task_id

def _reverse_worker(self, task_id: str, macro_repr: 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 macro_repr to query ent cache by projecting back to dim (approx)
            # approximate back-projection using pseudo-inverse of macro_proj * 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 high_proj pseudo-inverse
            approx_full = np.linalg.pinv(self.high_proj).dot(np.concatenate([macro_repr, np.zeros(self.high_dim - len(macro_repr))])[:self.high_dim])
            for i in range(4):
                noise = np.random.normal(scale=1e-2, size=approx_full.shape)
                synths.append((approx_full + 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:
            mem_id = uid()
            mu = MemoryUnit(id=mem_id, embedding=s.copy(), xi=0.0)
            mu.quarantined = True
            mu.explain = f"reverse_proj_task:{task_id}"
            # store in hot for inspection but quarantined
            self.storage.hot[mem_id] = mu
            self.storage.index[mem_id] = mu.embedding.copy()
            self.storage.quarantine[mem_id] = {"snapshot_hash": sha256_hex(mem_id + ":" + str(time.time())), "expire_ts": time.time() + QUARANTINE_HOLD, "requester": "reverse_proj", "reason": "reverse_projection", "task_id": task_id}
            created.append(mem_id)
            Ledger.record("PROJ_REVERSE_CREATED", mem_id, {"task_id": task_id})
        # callback with created ids
        if callback:
            try:
                callback(task_id, created)
            except Exception:
                pass
        log(f"ProjectionEngine: reverse task {task_id} created {len(created)} synthetic units")
    except Exception as e:
        log(f"ProjectionEngine: _reverse_worker error: {e}")

def high_dim_project(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')

    # ent cache
    self.ent_capacity = ENT_CAPACITY
    self.ent_cache: Dict[str, Dict[str, Any]] = {}
    self.ent_lru: List[str] = []
    self.ent_hit_threshold = 0.85
    self.ent_lock = threading.Lock()

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

    # prefetch queue
    self.prefetch_q = queue.Queue(maxsize=2048)
    self.prefetch_enabled = PREFETCH_ENABLED

    # threadpool for project batch
    self.project_executor = ThreadPoolExecutor(max_workers=MAX_PROJECT_WORKERS)

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

    # background workers
    self._flush_thread = threading.Thread(target=self._faiss_flush_worker, daemon=True)
    self._flush_thread.start()
    self._prefetch_thread = threading.Thread(target=self._prefetch_worker, daemon=True)
    self._prefetch_thread.start()

    # adaptive stats
    self.ent_hits = 0
    self.ent_queries = 0
    self.adapt_lock = threading.Lock()
    self.storage = None
    Ledger.record("FUSIONCORE_INIT", uid(), {"dim": dim, "faiss": self.use_faiss, "redis": self.use_redis})
    log("FusionCore initialized (projection-ready)")

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

# FAISS buffered add
def faiss_add_buffered(self, mem_id: str, emb: np.ndarray):
    if not self.use_faiss or self.faiss_index is None:
        with self.faiss_lock:
            self.faiss_ids.append(mem_id)
            self.faiss_buffer.append(emb.astype('float32').copy())
        return
    with self.faiss_lock:
        self.faiss_buffer.append(emb.astype('float32').reshape(1, -1))
        self.faiss_buffer_ids.append(mem_id)
        if len(self.faiss_buffer) >= FAISS_BATCH:
            self._flush_faiss_buffer()

def _flush_faiss_buffer(self):
    if not self.use_faiss or self.faiss_index is None:
        return
    try:
        vecs = np.vstack(self.faiss_buffer)
        self.faiss_index.add(vecs)
        self.faiss_ids.extend(self.faiss_buffer_ids)
        Ledger.record("FAISS_BATCH_ADD", uid(), {"count": len(self.faiss_buffer_ids)})
        log(f"FusionCore: FAISS batch add {len(self.faiss_buffer_ids)}")
    except Exception as e:
        log(f"FusionCore: FAISS batch add error: {e}")
    finally:
        self.faiss_buffer = []
        self.faiss_buffer_ids = []

def _faiss_flush_worker(self):
    while not self._stop:
        try:
            time.sleep(1.0)
            with self.faiss_lock:
                if self.faiss_buffer:
                    self._flush_faiss_buffer()
                if self.use_faiss and self.faiss_index is not None:
                    try:
                        faiss.write_index(self.faiss_index, FAISS_INDEX_PATH)
                    except Exception as e:
                        log(f"FusionCore: FAISS persist error: {e}")
        except Exception as e:
            log(f"FusionCore: faiss_flush_worker error: {e}")
            time.sleep(1.0)

def faiss_search(self, q_emb: np.ndarray, topk: int = 5) -> List[str]:
    if self.use_faiss and self.faiss_index is not None and len(self.faiss_ids) > 0:
        try:
            q = q_emb.astype('float32').reshape(1, -1)
            D, I = self.faiss_index.search(q, topk)
            res = []
            for idx in I[0]:
                if idx < 0 or idx >= len(self.faiss_ids):
                    continue
                res.append(self.faiss_ids[int(idx)])
            MET_FAISS_SEARCH.inc()
            Ledger.record("FAISS_SEARCH", uid(), {"topk": topk, "found": len(res)})
            return res
        except Exception as e:
            log(f"FusionCore: faiss_search error: {e}")
    with self.faiss_lock:
        if not self.faiss_buffer and not self.faiss_ids:
            return []
        return self.faiss_ids[:topk]

# ent cache with redis optional
def ent_get(self, q_emb: np.ndarray):
    key = sha256_hex(",".join(map(str, np.round(q_emb[:8], 3).tolist())))[:16]
    with self.ent_lock:
        self.ent_queries += 1
        e = self.ent_cache.get(key)
        if not e and self.use_redis and self.redis_client:
            try:
                raw = self.redis_client.get("ent:" + key)
                if raw:
                    obj = json.loads(raw)
                    e = {"ent_vec": np.array(obj["ent_vec"], dtype='float32'), "mem_ids": obj["mem_ids"], "last_access": time.time(), "xi_reserve": obj.get("xi_reserve", 0.05)}
                    self.ent_cache[key] = e
                    self.ent_lru.insert(0, key)
            except Exception:
                pass
        if not e:
            return None
        ent_vec = e['ent_vec']
        sim = self._cosine_sim(ent_vec, q_emb)
        if sim < self.ent_hit_threshold:
            return None
        e['last_access'] = time.time()
        if key in self.ent_lru:
            self.ent_lru.remove(key)
        self.ent_lru.insert(0, key)
        self.ent_hits += 1
        MET_ENT_HIT.inc()
        Ledger.record("ENT_HIT", key, {"sim": float(sim)})
        self._maybe_adapt()
        return e

def ent_put(self, mem_embeddings: List[np.ndarray], mem_ids: List[str], xi_reserve: float = 0.05):
    if not mem_embeddings:
        return None
    ent_vec = np.mean(np.stack(mem_embeddings, axis=0), axis=0).astype('float32')
    key = sha256_hex(",".join(map(str, np.round(ent_vec[:8], 4).tolist())))[:16]
    with self.ent_lock:
        if key in self.ent_cache:
            self.ent_cache[key].update({"ent_vec": ent_vec, "mem_ids": mem_ids, "last_access": time.time(), "xi_reserve": xi_reserve})
            if key in self.ent_lru:
                self.ent_lru.remove(key)
            self.ent_lru.insert(0, key)
        else:
            if len(self.ent_lru) >= self.ent_capacity:
                tail = self.ent_lru.pop()
                self.ent_cache.pop(tail, None)
                if self.use_redis and self.redis_client:
                    try:
                        self.redis_client.delete("ent:" + tail)
                    except Exception:
                        pass
            self.ent_cache[key] = {"ent_vec": ent_vec, "mem_ids": mem_ids, "last_access": time.time(), "xi_reserve": xi_reserve}
            self.ent_lru.insert(0, key)
        if self.use_redis and self.redis_client:
            try:
                self.redis_client.set("ent:" + key, json.dumps({"ent_vec": ent_vec.tolist(), "mem_ids": mem_ids, "xi_reserve": xi_reserve}), ex=3600)
            except Exception:
                pass
        MET_ENT_PUT.inc()
        Ledger.record("ENT_PUT", key, {"count": len(mem_ids)})
        return key

# project with toxicity check and projection hooks
def project(self, query_emb: np.ndarray, background: Optional[np.ndarray] = None, alpha: float = 0.6, projection_mode: str = "micro"):
    """
    projection_mode: "micro" (default) uses micro->macro fast mapping for additional context,
                     "macro" uses macro-level aggregation,
                     "high" uses high-dim latent mixing.
    """
    start = time.time()
    MET_PROJECT.inc()
    # quick toxicity pre-check on query
    tscore = Detox.toxicity_score(query_emb, background)
    # projection engine hooks if attached
    proj_meta = {}
    if hasattr(self, "projection_engine") and self.projection_engine:
        if projection_mode == "micro":
            pm = self.projection_engine.micro_to_macro(query_emb)
            proj_meta = pm["meta"]
        elif projection_mode == "macro":
            pm = self.projection_engine.micro_to_macro(query_emb)
            proj_meta = pm["meta"]
        elif projection_mode == "high":
            h = self.projection_engine.high_dim_project(query_emb)
            proj_meta = {"high_dim_norm": float(np.linalg.norm(h))}
    # ent cache
    ent = self.ent_get(query_emb)
    if ent is not None:
        ent_vec = ent['ent_vec']
        emb = alpha * query_emb + (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 * query_emb + (1.0 - alpha) * ent_vec
                delta_E = float(np.linalg.norm(emb - ent_vec))
                holo = Hologram(id=uid(), embedding=emb, confidence=0.86, provenance={"method": "faiss_ent", "ent_key": ent_key, "proj": projection_mode}, delta_E=delta_E, toxic_score=tscore)
                holo.explain = "faiss_ent"
                LAT_PROJECT.observe(time.time() - start)
                Ledger.record("PROJECT_FAISS", holo.id, {"delta_E": delta_E, "hits": len(hits), "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
    # 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 * query_emb + (1.0 - alpha) * B
    emb += np.random.normal(scale=1e-6, size=emb.shape)
    delta_E = float(np.linalg.norm(emb - B))
    holo = Hologram(id=uid(), embedding=emb, confidence=0.72, provenance={"method": "background", "proj": projection_mode}, delta_E=delta_E, toxic_score=tscore)
    holo.explain = "background"
    LAT_PROJECT.observe(time.time() - start)
    Ledger.record("PROJECT_BG", 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

def _handle_toxic_hologram(self, holo: Hologram):
    try:
        res = self.storage.negentropy_read(holo, toxicity_threshold=TOXICITY_THRESHOLD)
        if res.get("status") == "ok":
            Ledger.record("TOXIC_HANDLED_OK", holo.id, {"method": "light_repair"})
            return
        repaired = res.get("hologram")
        if repaired and res.get("toxic"):
            time.sleep(0.2)
            if Detox.toxicity_score(repaired.embedding, self.storage.background) >= TOXICITY_THRESHOLD:
                mem_id = uid()
                mu = MemoryUnit(id=mem_id, embedding=repaired.embedding.copy(), xi=0.0)
                mu.quarantined = True
                mu.explain = f"auto_quarantine_from_holo:{holo.id}"
                self.storage.hot[mem_id] = mu
                self.storage.index[mem_id] = mu.embedding.copy()
                Ledger.record("AUTO_QUARANTINE", mem_id, {"from_holo": holo.id, "toxic_score": repaired.delta_E})
                log(f"FusionCore: Auto-quarantined synthetic unit {mem_id[:8]} from holo {holo.id[:8]}")
    except Exception as e:
        log(f"FusionCore: _handle_toxic_hologram error: {e}")

def project_batch(self, queries: List[np.ndarray], background: Optional[np.ndarray] = None, alpha: float = 0.6, projection_mode: 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) * ent_vec
            delta_E = float(np.linalg.norm(emb - ent_vec))
            tscore = Detox.toxicity_score(emb, background)
            results[i] = Hologram(id=uid(), embedding=emb, confidence=0.92, provenance={"method": "entangled"}, delta_E=delta_E, toxic_score=tscore, explain="entangled")
    futures = {}
    for i, q in enumerate(queries):
        if results[i] is None:
            futures[self.project_executor.submit(self.project, q, background, alpha, projection_mode)] = i
    for fut in as_completed(futures):
        i = futures[fut]
        try:
            results[i] = fut.result()
        except Exception as e:
            log(f"project_batch worker error: {e}")
            results[i] = self.project(queries[i], background, alpha, projection_mode)
    return results

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

def _prefetch_worker(self):
    while not self._stop:
        try:
            emb = self.prefetch_q.get(timeout=1.0)
            hits = self.faiss_search(emb, topk=8)
            if hits and self.storage:
                mem_embs = [self.storage.index[h] for h in hits if h in self.storage.index]
                if mem_embs:
                    self.ent_put(mem_embs, hits, xi_reserve=0.02)
        except queue.Empty:
            continue
        except Exception as e:
            log(f"FusionCore: prefetch_worker error: {e}")
            time.sleep(0.2)

def _cosine_sim(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 _maybe_adapt(self):
    with self.adapt_lock:
        if self.ent_queries >= 200:
            hit_rate = self.ent_hits / max(1, self.ent_queries)
            if hit_rate > 0.6 and self.ent_hit_threshold > 0.6:
                self.ent_hit_threshold = max(0.5, self.ent_hit_threshold - 0.02)
            elif hit_rate < 0.2 and self.ent_hit_threshold < 0.95:
                self.ent_hit_threshold = min(0.95, self.ent_hit_threshold + 0.02)
            self.ent_hits = 0
            self.ent_queries = 0

def shutdown(self):
    self._stop = True
    log("FusionCore: shutdown requested")
    with self.faiss_lock:
        if self.faiss_buffer:
            self._flush_faiss_buffer()
        if self.use_faiss and self.faiss_index is not None:
            try:
                faiss.write_index(self.faiss_index, FAISS_INDEX_PATH)
                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

# storage primitives
def put_unit(self, unit: MemoryUnit, hot: bool = True, near: bool = True):
    unit.version += 1
    unit.ts = time.time()
    unit.last_active = time.time()
    if hot:
        self.hot[unit.id] = unit
    if near:
        self.near[unit.id] = unit
    self.index[unit.id] = unit.embedding.copy()
    self._incremental_background_update(unit)
    if self.fusion:
        self.fusion.faiss_add_buffered(unit.id, unit.embedding)
    MET_POCKET_PUT.inc()
    Ledger.record("PUT_UNIT", unit.id, {"xi": unit.xi, "core_protected": unit.core_protected, "explain": unit.explain})
    try:
        GAUGE_UNITS.set(len(self.index)) if ENABLE_PROM else None
    except Exception:
        pass

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

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

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

# pocket_put with sanitization and toxicity pre-check
def pocket_put(self, payload: bytes, embedding: np.ndarray, xi: float = 0.5, core_protect: bool = False, importance: float = 0.0, emotion: float = 0.0):
    try:
        clean_payload = Sanitizer.clean_text(payload)
    except Exception:
        clean_payload = payload
    mem_id = uid()
    unit = MemoryUnit(id=mem_id, embedding=embedding.copy(), xi=xi, trit=0, importance=importance, emotion=emotion, core_protected=core_protect)
    unit.explain = "sanitized_payload"
    sid = uid()
    self.put_shard(sid, clean_payload, xi, 0)
    unit.shards = [sid]
    tscore = Detox.toxicity_score(embedding, self.background)
    if tscore >= TOXICITY_THRESHOLD:
        unit.quarantined = True
        unit.explain = f"quarantined_on_put:score={tscore:.3f}"
        self.hot[mem_id] = unit
        self.index[mem_id] = unit.embedding.copy()
        self.quarantine[mem_id] = {"snapshot_hash": sha256_hex(mem_id + ":" + str(time.time())), "expire_ts": time.time() + QUARANTINE_HOLD, "requester": "auto", "reason": "toxicity_on_put", "score": tscore}
        Ledger.record("POCKET_PUT_QUARANTINED", mem_id, {"score": tscore})
        log(f"StorageCore: Pocket put quarantined {mem_id[:8]} score={tscore:.3f}")
        return {"status": "quarantined", "mem_id": mem_id, "score": tscore}
    self.put_unit(unit, hot=False, near=True)
    vaddr = "v:" + mem_id[:8]
    self.page_table[vaddr] = {"mem_id": mem_id, "local": False, "last_access": time.time()}
    if len(self.local_cache) < self.max_local and self.xi_pool > PROMOTE_COST:
        self._promote_to_local(vaddr)
    Ledger.record("POCKET_PUT", vaddr, {"mem_id": mem_id})
    return {"status": "ok", "vaddr": vaddr}

def pocket_query(self, context_emb: np.ndarray, topk: int = 5, projection_mode: str = "micro"):
    start = time.time()
    MET_QUERY.inc()
    mids = []
    if self.fusion:
        mids = self.fusion.faiss_search(context_emb, 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(context_emb) + 1e-12
        norms = np.linalg.norm(mats, axis=1) + 1e-12
        sims = (mats @ context_emb) / (norms * qn)
        top_idx = np.argsort(-sims)[:topk]
        mids = [ids[int(i)] for i in top_idx]
    results = []
    for mid in mids:
        u = self.retrieve_any(mid)
        if not u or u.quarantined:
            continue
        vaddr = None
        for va, info in self.page_table.items():
            if info["mem_id"] == mid:
                vaddr = va; break
        if not vaddr:
            vaddr = "v:" + mid[:8]
            self.page_table[vaddr] = {"mem_id": mid, "local": False, "last_access": time.time()}
        self.page_table[vaddr]["last_access"] = time.time()
        if len(results) < self.max_local:
            threading.Thread(target=self._promote_to_local, args=(vaddr,), daemon=True).start()
        results.append({"vaddr": vaddr, "mem_id": mid, "explain": u.explain})
    LAT_QUERY.observe(time.time() - start)
    Ledger.record("POCKET_QUERY", uid(), {"hits": len(results), "proj_mode": projection_mode})
    return results

def _promote_to_local(self, vaddr: str):
    info = self.page_table.get(vaddr)
    if not info:
        return
    mem_id = info["mem_id"]
    if self.xi_pool < PROMOTE_COST:
        Ledger.record("PROMOTE_FAIL", vaddr, {"xi_pool": self.xi_pool})
        return
    self.xi_pool -= PROMOTE_COST
    u = self.retrieve_any(mem_id)
    if u:
        info["local"] = True
        self.local_cache[vaddr] = mem_id
        if len(self.local_cache) > self.max_local:
            self._evict_one()
        Ledger.record("PROMOTE", vaddr, {"mem_id": mem_id, "xi_pool": self.xi_pool})

def _evict_one(self):
    lru = None; lru_ts = float('inf')
    for va, info in self.page_table.items():
        if info.get("local") and info["last_access"] < lru_ts:
            lru = va; lru_ts = info["last_access"]
    if lru:
        self.page_table[lru]["local"] = False
        self.local_cache.pop(lru, None)
        Ledger.record("EVICT", lru, {})

# consolidation worker
def push_consolidation(self, mem_id: str):
    try:
        self.consolidation_q.put(mem_id, timeout=0.1)
        Ledger.record("CONSOLIDATION_PUSH", mem_id, {})
    except Exception:
        pass

def _consolidation_worker(self):
    batch = []
    while not self._stop:
        try:
            mem_id = self.consolidation_q.get(timeout=1.0)
            batch.append(mem_id)
            if len(batch) >= CONSOLIDATION_BATCH:
                self._do_consolidation_batch(batch)
                batch = []
        except queue.Empty:
            if batch:
                self._do_consolidation_batch(batch)
                batch = []
        except Exception as e:
            log(f"StorageCore: consolidation worker error: {e}")
            time.sleep(0.5)

def _do_consolidation_batch(self, mem_ids: List[str]):
    for mem_id in mem_ids:
        u = self.retrieve_any(mem_id)
        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.put_shard(sid, c, u.xi, u.trit)
            sids.append(sid)
        u.shards = sids
        self.put_unit(u, hot=False, near=True)
        Ledger.record("CONSOLIDATION_DONE", u.id, {"shards": len(sids)})
    log(f"StorageCore: Consolidation batch done size={len(mem_ids)}")
    if len(self.index) >= 16 and self.fusion:
        sample_ids = list(self.index.keys())[:min(12, len(self.index))]
        mem_embs = [self.index[i] for i in sample_ids]
        self.fusion.ent_put(mem_embs, sample_ids, xi_reserve=0.05)
        self.rebuild_event.set()

# background worker
def _background_worker(self):
    last_rebuild = 0.0
    while not self._stop:
        try:
            triggered = self.rebuild_event.wait(timeout=BACKGROUND_REBUILD_INTERVAL)
            with self.rebuild_lock:
                now = time.time()
                if now - last_rebuild > 0.5:
                    self._rebuild_background_field()
                    last_rebuild = now
                self.rebuild_event.clear()
        except Exception as e:
            log(f"StorageCore: background worker error: {e}")
            time.sleep(0.5)

def _rebuild_background_field(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.retrieve_any(i).xi if self.retrieve_any(i) else 0.01, 0.01) * (1.0 + (self.retrieve_any(i).importance if self.retrieve_any(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("BACKGROUND_REBUILD", uid(), {"units": len(ids)})
    log("StorageCore: Background field rebuilt")

# negentropy_read (enhanced)
def negentropy_read(self, holo: Hologram, toxicity_threshold: float = TOXICITY_THRESHOLD):
    mean_val = float(np.mean(holo.embedding))
    toxic_score = Detox.toxicity_score(holo.embedding, self.background)
    temp = MemoryUnit(id="temp", embedding=holo.embedding.copy(), xi=0.5)
    violations = self._check_core_rules(temp)
    toxic = (toxic_score > toxicity_threshold) or (len(violations) > 0)
    if not toxic:
        Ledger.record("NEGENTROPY_OK", holo.id, {"toxic_score": toxic_score})
        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)
    repaired_emb = holo.embedding + c
    delta_comp = float(np.linalg.norm(c))
    repaired = Hologram(id=uid(), embedding=repaired_emb, confidence=max(0.1, holo.confidence - 0.05), provenance={"repair_of": holo.id, "stage": "light"}, delta_E=holo.delta_E + delta_comp)
    new_score = Detox.toxicity_score(repaired.embedding, self.background)
    Ledger.record("NEGENTROPY_REPAIR_STAGE1", repaired.id, {"orig": holo.id, "delta_comp": delta_comp, "new_score": new_score})
    if new_score <= toxicity_threshold:
        threading.Thread(target=self._async_validate_repair, args=(repaired,), daemon=True).start()
        return {"status": "repaired", "hologram": repaired, "toxic": False, "delta_comp": delta_comp}
    # Stage 2: stronger repair (async)
    stronger = Hologram(id=uid(), embedding=repaired.embedding.copy(), confidence=max(0.05, repaired.confidence - 0.1), provenance={"repair_of": holo.id, "stage": "strong"}, delta_E=repaired.delta_E)
    threading.Thread(target=self._strong_repair_and_validate, args=(stronger, holo.id), daemon=True).start()
    return {"status": "repaired_async", "hologram": stronger, "toxic": True, "delta_comp": delta_comp}

def _strong_repair_and_validate(self, repaired: Hologram, orig_holo_id: str):
    try:
        repaired.embedding = 0.5 * repaired.embedding + 0.5 * self.background
        new_score = Detox.toxicity_score(repaired.embedding, self.background)
        Ledger.record("NEGENTROPY_REPAIR_STAGE2", repaired.id, {"orig": orig_holo_id, "new_score": new_score})
        if new_score > TOXICITY_THRESHOLD:
            mem_id = uid()
            mu = MemoryUnit(id=mem_id, embedding=repaired.embedding.copy(), xi=0.0)
            mu.quarantined = True
            mu.explain = f"auto_quarantine_from_repair:{orig_holo_id}"
            self.hot[mem_id] = mu
            self.index[mem_id] = mu.embedding.copy()
            self.quarantine[mem_id] = {"snapshot_hash": sha256_hex(mem_id + ":" + str(time.time())), "expire_ts": time.time() + QUARANTINE_HOLD, "requester": "auto_repair", "reason": "repair_failed", "score": new_score}
            Ledger.record("AUTO_QUARANTINE", mem_id, {"from_holo": orig_holo_id, "score": new_score})
            log(f"StorageCore: Auto-quarantined {mem_id[:8]} from repair of holo {orig_holo_id[:8]}")
        else:
            Ledger.record("NEGENTROPY_VALIDATE", repaired.id, {"validated": True})
            log(f"StorageCore: Repair validated {repaired.id[:8]}")
    except Exception as e:
        log(f"StorageCore: _strong_repair_and_validate error: {e}")

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

# core rules
core_rules: Dict[str, Any] = {}
def add_core_rule(self, rule_id: str, fn, signer: str = "admin"):
    self.core_rules[rule_id] = {"fn": fn, "signer": signer}
    Ledger.record("CORE_RULE_ADD", rule_id, {"signer": signer})
    log(f"StorageCore: Core rule added {rule_id}")

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

# decay worker
def _decay_worker(self):
    while not self._stop:
        try:
            time.sleep(DECAY_INTERVAL)
            now = time.time()
            to_decay = []
            for mid, u in list(self.hot.items()):
                age = now - u.last_active
                u.decay_score += DECAY_RATE * (age / max(1.0, DECAY_INTERVAL))
                if u.decay_score > 0.5 and u.importance < 0.1:
                    to_decay.append(mid)
            for mid in to_decay:
                u = self.hot.get(mid)
                if not u:
                    continue
                u.xi = max(0.0, u.xi - 0.1)
                Ledger.record("DECAY_APPLIED", mid, {"decay_score": u.decay_score, "xi": u.xi})
                if u.xi <= 0.0 and not u.core_protected:
                    self.request_self_delete(mid, requester="decay_worker", hold_seconds=QUARANTINE_HOLD)
            log(f"StorageCore: Decay pass done, decayed={len(to_decay)}")
        except Exception as e:
            log(f"StorageCore: decay worker error: {e}")
            time.sleep(1.0)

# quarantine / delete
def request_self_delete(self, mem_id: str, requester: str, hold_seconds: float = QUARANTINE_HOLD):
    u = self.retrieve_any(mem_id)
    if not u:
        return {"status": "not_found"}
    if u.core_protected:
        Ledger.record("DELETE_REJECT_CORE", mem_id, {"requester": requester})
        return {"status": "rejected_core_protected"}
    snap_hash = sha256_hex(mem_id + ":" + str(time.time()))
    expire_ts = time.time() + hold_seconds
    self.quarantine[mem_id] = {"snapshot_hash": snap_hash, "expire_ts": expire_ts, "requester": requester, "reason": "self_delete"}
    u.quarantined = True
    u.delete_requester = requester
    u.delete_request_ts = time.time()
    if mem_id in self.index:
        self.index.pop(mem_id, None)
    Ledger.record("QUARANTINE", mem_id, {"requester": requester, "expire_ts": expire_ts})
    log(f"StorageCore: Quarantined {mem_id[:8]} by {requester} until {expire_ts}")
    threading.Thread(target=self._delayed_permanent_delete, args=(mem_id, expire_ts), daemon=True).start()
    MET_DELETE.inc()
    return {"status": "quarantined", "mem_id": mem_id, "hold_until": expire_ts}

def undo_delete(self, mem_id: str, requester: str):
    info = self.quarantine.get(mem_id)
    if not info:
        return {"status": "not_quarantined"}
    if info["requester"] != requester and requester != "admin":
        return {"status": "not_authorized"}
    u = self.retrieve_any(mem_id)
    if not u:
        return {"status": "unit_missing"}
    u.quarantined = False
    u.delete_requester = None
    u.delete_request_ts = None
    self.index[mem_id] = u.embedding.copy()
    self.quarantine.pop(mem_id, None)
    Ledger.record("UNDO_QUARANTINE", mem_id, {"requester": requester})
    log(f"StorageCore: Undo quarantine {mem_id[:8]} by {requester}")
    return {"status": "restored", "mem_id": mem_id}

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

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

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

def _populate_demo(self):
    log(f"Demo: populating {self.demo_units} memory units")
    for i in range(self.demo_units):
        emb = np.random.normal(scale=1.0, size=(self.dim,)).astype('float32')
        payload = f"demo-{i}".encode('utf-8')
        res = self.storage.pocket_put(payload, emb, xi=0.5)
        # occasionally push for consolidation
        if isinstance(res, dict) and res.get("status") == "ok":
            vaddr = res["vaddr"]
            mem_id = self.storage.page_table.get(vaddr, {}).get("mem_id")
            if mem_id and i % max(1, CONSOLIDATION_BATCH // 4) == 0:
                self.storage.push_consolidation(mem_id)
        if PREFETCH_ENABLED:
            self.fusion.prefetch(emb)
        time.sleep(0.002)
    log("Demo population done")
    Ledger.record("DEMO_POPULATED", uid(), {"units": self.demo_units})
    self.storage.rebuild_event.set()

def _auto_project_batcher(self):
    while not self._stop:
        try:
            batch = []
            for _ in range(AUTO_PROJECT_BATCH):
                q = np.random.normal(scale=1.0, size=(self.dim,)).astype('float32')
                batch.append(q)
            if not self.storage.token_bucket.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.project_batch(batch, background=self.storage.background, alpha=0.6, projection_mode=mode)
            for holo in holos:
                res = self.storage.negentropy_read(holo)
                Ledger.record("AUTO_PROJECT", holo.id, {"status": res.get("status"), "conf": holo.confidence, "toxic": holo.toxic_score, "proj_mode": mode})
            time.sleep(max(0.05, BACKGROUND_REBUILD_INTERVAL / 8.0))
        except Exception as e:
            log(f"Auto project batcher error: {e}")
            time.sleep(0.5)

def _start_http(self):
    app = FastAPI()
    @app.get("/health")
    def health():
        return {"status": "ok", "time": now_ts()}
    @app.get("/metrics")
    def metrics():
        if ENABLE_PROM:
            return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
        else:
            return {"metrics": "disabled"}
    @app.post("/put")
    def put_item(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.pocket_put(text, emb, xi=float(payload.get("xi", 0.5)))
            return res
        except Exception as e:
            raise HTTPException(status_code=500, detail=str(e))
    @app.post("/query")
    def query_item(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("proj_mode", "micro")
            res = self.storage.pocket_query(emb, topk=topk, projection_mode=mode)
            return {"hits": res}
        except Exception as e:
            raise HTTPException(status_code=500, detail=str(e))
    @app.get("/ledger")
    def ledger_count():
        return {"entries": len(Ledger.chain)}
    @app.post("/admin/snapshot")
    def admin_snapshot():
        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 _signal_handler(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")