Shuiquan-QEM-Cognitive-Field

#!/usr/bin/env python3

-- coding: utf-8 --

""" qem_autonomous_final_embedded.py Ultimate improved single-file — QEM Autonomous Final (with quarantine sublimation retry embedded)

• Based on qem_autonomous_final, embeds retry_quarantine_and_sublimate
• Automatically retries top-K pairs from quarantine after a single run (configurable)
• Supports --debug-run-once, --relax-sim, --force-cluster, --cluster-eps
• Writes detailed debug logs to qem_cloud_data/debug_log.txt

Notes:

• Place this file in the project root and run: python qem_autonomous_final_embedded.py [--debug-run-once]
• To force cluster merges: --force-cluster --cluster-eps 0.25
• Environment variables can override some parameters (see Config) """ from future import annotations import os import sys import time import json import uuid import math import random import threading import traceback import shutil import http.server import socketserver import argparse from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple

Optional acceleration

try: import numpy as np except Exception: np = None

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

Config

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

@dataclass class Config: base_dir: str = os.path.abspath(".") data_dir: str = os.path.join(base_dir, "qem_cloud_data") shard_dir: str = field(init=False) ent_path: str = field(init=False) seed_path: str = field(init=False) ledger_path: str = field(init=False) comp_log: str = field(init=False) backup_dir: str = field(init=False) debug_log: str = field(init=False) quarantine_path: str = field(init=False) auto_disable_file: str = field(init=False) shutdown_signal: str = field(init=False)

dim: int = int(os.environ.get("QEM_DIM", 64))
default_sim: float = float(os.environ.get("QEM_SIM", 0.70))
default_iters: int = int(os.environ.get("QEM_ITERS", 4))
quant_bits: int = int(os.environ.get("QEM_QUANT_BITS", 8))
min_group: int = int(os.environ.get("QEM_MIN_GROUP", 2))
freq_alpha: float = float(os.environ.get("QEM_FREQ_ALPHA", 1.0))
freq_beta: float = float(os.environ.get("QEM_FREQ_BETA", 1.0))
pair_sim_factor: float = float(os.environ.get("QEM_PAIR_SIM_FACTOR", 1.0))
status_port: int = int(os.environ.get("QEM_STATUS_PORT", 0))
max_auto_inject: int = int(os.environ.get("QEM_MAX_AUTO_INJECT", 128))
poll_interval: float = float(os.environ.get("QEM_POLL_INTERVAL", 5.0))
idle_run_seconds: int = int(os.environ.get("QEM_IDLE_RUN_SECONDS", 120))
log_prefix: str = "[QEM-Auto-Final-Embedded]"
sim_min: float = float(os.environ.get("QEM_SIM_MIN", 0.35))
quarantine_retry_limit: int = int(os.environ.get("QEM_QUARANTINE_RETRY", 3))
max_backup_keep: int = int(os.environ.get("QEM_MAX_BACKUP_KEEP", 10))

def __post_init__(self):
    self.shard_dir = os.path.join(self.data_dir, "shards")
    self.ent_path = os.path.join(self.data_dir, "ents.json")
    self.seed_path = os.path.join(self.data_dir, "seeds.json")
    self.ledger_path = os.path.join(self.data_dir, "ledger.json")
    self.comp_log = os.path.join(self.data_dir, "compression_log.json")
    self.backup_dir = os.path.join(self.data_dir, "backups")
    self.debug_log = os.path.join(self.data_dir, "debug_log.txt")
    self.quarantine_path = os.path.join(self.data_dir, "quarantine.json")
    self.auto_disable_file = os.path.join(self.data_dir, "auto_run.disable")
    self.shutdown_signal = os.path.join(self.data_dir, "shutdown.signal")
    os.makedirs(self.data_dir, exist_ok=True)
    os.makedirs(self.shard_dir, exist_ok=True)
    os.makedirs(self.backup_dir, exist_ok=True)

cfg = Config()

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

Utilities & logging

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

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

def uid(prefix: str = "") -> str: return prefix + str(uuid.uuid4())[:12]

def safe_write_json(path: str, obj: Any): tmp = path + ".tmp" try: with open(tmp, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False, indent=2) os.replace(tmp, path) except Exception: try: with open(path, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False, indent=2) except Exception: pass

def _append_debug(s: str): try: with open(cfg.debug_log, "a", encoding="utf-8") as f: f.write(f"{now_ts()} {s}\n") except Exception: pass

def log(msg: str): line = f"{cfg.log_prefix} [{now_ts()}] {msg}" print(line, flush=True) _append_debug(msg)

def log_exc(prefix: str = "EXC"): tb = traceback.format_exc() log(f"{prefix}: {tb}")

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

Ledger & comp log

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

class Ledger: _lock = threading.RLock() chain: List[Dict[str, Any]] = []

@classmethod
def record(cls, op: str, obj_id: str, info: Dict[str, Any]):
    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}
        try:
            s = json.dumps(entry, sort_keys=True, ensure_ascii=False)
            import hashlib
            entry['hash'] = hashlib.sha256(s.encode('utf-8')).hexdigest()
        except Exception:
            entry['hash'] = uid("h-")
        cls.chain.append(entry)

@classmethod
def dump(cls):
    with cls._lock:
        try:
            safe_write_json(cfg.ledger_path, cls.chain)
            log(f"Ledger dumped to {cfg.ledger_path}")
        except Exception:
            log_exc("Ledger.dump failed")

def append_comp_log(rec: Dict[str, Any]): arr = [] try: if os.path.exists(cfg.comp_log): with open(cfg.comp_log, "r", encoding="utf-8") as f: arr = json.load(f) except Exception: arr = [] arr.append(rec) safe_write_json(cfg.comp_log, arr)

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

Frequency store

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

class FrequencyStore: def init(self, path=os.path.join(cfg.data_dir, "freq.json")): self.path = path self.lock = threading.RLock() self.counts: Dict[str,int] = {} self._load()

def _load(self):
    if os.path.exists(self.path):
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                self.counts = json.load(f)
        except Exception:
            self.counts = {}

def inc(self, k:str, d:int=1):
    with self.lock:
        self.counts[k] = self.counts.get(k,0) + d
        if self.counts[k] % 50 == 0:
            safe_write_json(self.path, self.counts)

def get(self,k:str)->int:
    with self.lock:
        return self.counts.get(k,0)

def snapshot(self):
    with self.lock:
        return dict(self.counts)

freq_store = FrequencyStore()

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

Data models

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

@dataclass class EntNode: id: str vec: Optional[List[float]] shards: List[str] score: float = 1.0 ts: float = field(default_factory=time.time)

def to_dict(self):
    return {"id":self.id,"vec":self.vec,"shards":self.shards,"score":self.score,"ts":self.ts}

@staticmethod
def from_dict(d):
    return EntNode(id=d["id"], vec=d.get("vec"), shards=d.get("shards",[]), score=d.get("score",1.0), ts=d.get("ts", time.time()))

@dataclass class SeedNode: id: str seed_vec: Optional[List[float]] members: List[str] diffs: Dict[str, List[int]] = field(default_factory=dict) quant_meta: Dict[str,Any] = field(default_factory=dict) ts: float = field(default_factory=time.time)

def to_dict(self):
    return {"id":self.id,"seed_vec":self.seed_vec,"members":self.members,"diffs":self.diffs,"quant_meta":self.quant_meta,"ts":self.ts}

@staticmethod
def from_dict(d):
    return SeedNode(id=d["id"], seed_vec=d.get("seed_vec"), members=d.get("members",[]), diffs=d.get("diffs",{}), quant_meta=d.get("quant_meta",{}), ts=d.get("ts", time.time()))

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

Registry & SeedIndex

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

class EntRegistry: def init(self, path=cfg.ent_path): self.path = path self.lock = threading.RLock() self.nodes: Dict[str,EntNode] = {} self._load()

def _load(self):
    if os.path.exists(self.path):
        try:
            with open(self.path,"r",encoding="utf-8") as f:
                data = json.load(f)
            for nid, nd in data.get("nodes",{}).items():
                self.nodes[nid] = EntNode.from_dict(nd)
        except Exception:
            self.nodes = {}

def save(self):
    with self.lock:
        data = {"nodes": {nid:n.to_dict() for nid,n in self.nodes.items()}}
        safe_write_json(self.path, data)

def register(self,node:EntNode):
    with self.lock:
        self.nodes[node.id] = node
        try:
            safe_write_json(self.path, {"nodes": {nid:n.to_dict() for nid,n in self.nodes.items()}})
        except Exception:
            pass
    Ledger.record("ENT_REGISTER", node.id, {"shards": len(node.shards)})

class SeedIndex: def init(self, path=cfg.seed_path): self.path = path self.lock = threading.RLock() self.seeds: Dict[str,SeedNode] = {} self._load()

def _load(self):
    if os.path.exists(self.path):
        try:
            with open(self.path,"r",encoding="utf-8") as f:
                data = json.load(f)
            for sid, sd in data.get("seeds",{}).items():
                self.seeds[sid] = SeedNode.from_dict(sd)
        except Exception:
            self.seeds = {}

def save(self):
    with self.lock:
        data = {"seeds": {sid:s.to_dict() for sid,s in self.seeds.items()}}
        safe_write_json(self.path, data)

def register(self, seed:SeedNode):
    with self.lock:
        self.seeds[seed.id] = seed
        try:
            safe_write_json(self.path, {"seeds": {sid:s.to_dict() for sid,s in self.seeds.items()}})
        except Exception:
            pass
    Ledger.record("SEED_REGISTER", seed.id, {"members": len(seed.members)})

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

Vector helpers

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

def _cosine(a,b): if not a or not b: return 0.0 try: if np is not None: aa = np.array(a, dtype=float); bb = np.array(b, dtype=float) an = np.linalg.norm(aa) + 1e-12; bn = np.linalg.norm(bb) + 1e-12 return float(np.dot(aa, bb) / (an * bn)) except Exception: pass m = min(len(a), len(b)) dot = sum((a[i] * b[i]) for i in range(m)) an = math.sqrt(sum(xx for x in a)) + 1e-12 bn = math.sqrt(sum(yy for y in b)) + 1e-12 return dot / (an * bn)

def mean_vec(vecs): if not vecs: return None try: if np is not None: arr = np.stack([np.array(v, dtype=float) for v in vecs], axis=0) return np.mean(arr, axis=0).tolist() except Exception: pass dim = max(len(v) for v in vecs) res = [0.0] * dim for v in vecs: for i, x in enumerate(v): res[i] += x n = len(vecs) return [x / n for x in res]

def vec_sub(a, b): if a is None or b is None: return None dim = max(len(a), len(b)) res = [] for i in range(dim): ai = a[i] if i < len(a) else 0.0 bi = b[i] if i < len(b) else 0.0 res.append(ai - bi) return res

def vec_add(a, b): if a is None: return b if b is None: return a dim = max(len(a), len(b)) res = [] for i in range(dim): ai = a[i] if i < len(a) else 0.0 bi = b[i] if i < len(b) else 0.0 res.append(ai + bi) return res

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

Quantization helpers

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

def quantize_list(vecs: List[List[float]], bits:int=cfg.quant_bits): flat = [x for v in vecs for x in v] if vecs else [] if not flat: return [], {} mn = min(flat); mx = max(flat) if mn == mx: q = [[0] * len(vecs[0]) for _ in vecs] return q, {"min": mn, "max": mx, "bits": bits} levels = (1 << bits) - 1 meta = {"min": mn, "max": mx, "bits": bits} qvecs = [] for v in vecs: qv = [int(round((x - mn) / (mx - mn) * levels)) for x in v] qvecs.append(qv) return qvecs, meta

def dequantize(qvec, meta): mn = meta.get("min", 0.0); mx = meta.get("max", 0.0); bits = meta.get("bits", cfg.quant_bits) levels = (1 << bits) - 1 if levels == 0: return [mn for _ in qvec] return [mn + (x / levels) * (mx - mn) for x in qvec]

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

Compressor with both strategies

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

class Compressor: def init(self, registry:EntRegistry, seed_index:SeedIndex): self.registry = registry self.seed_index = seed_index

# greedy cluster used by build_seeds and force-cluster
def greedy_cluster(self, nodes:List[EntNode], sim_thresh:float, min_group:int):
    groups = []; used = set()
    for i, a in enumerate(nodes):
        if a.id in used:
            continue
        group = [a]; used.add(a.id)
        for b in nodes[i+1:]:
            if b.id in used:
                continue
            try:
                if _cosine(a.vec, b.vec) >= sim_thresh:
                    group.append(b); used.add(b.id)
            except Exception:
                continue
        if len(group) >= min_group:
            groups.append(group)
    return groups

def build_seeds(self, sim_thresh:float=cfg.default_sim, min_group:int=cfg.min_group, quant_bits:int=cfg.quant_bits):
    nodes = list(self.registry.nodes.values())
    if not nodes:
        return {"created": 0}
    groups = self.greedy_cluster(nodes, sim_thresh, min_group)
    created = 0
    for g in groups:
        created += self._create_seed(g, quant_bits)
    return {"created": created, "groups": len(groups)}

def _create_seed(self, group:List[EntNode], quant_bits:int):
    vecs = [n.vec for n in group if n.vec is not None]
    if not vecs:
        return 0
    seed_vec = mean_vec(vecs)
    member_ids = []
    diffs = []
    ent_ids = []
    for n in group:
        member_ids.extend(n.shards)
        if n.vec is not None:
            d = vec_sub(n.vec, seed_vec)
            if d is not None:
                diffs.append(d)
                ent_ids.append(n.shards[0] if n.shards else n.id)
    qvecs, meta = quantize_list(diffs, bits=quant_bits) if diffs else ([], {})
    diffs_map = {ent_ids[i]: qvecs[i] for i in range(len(ent_ids))} if qvecs else {}
    seed_node = SeedNode(id=uid("seed-"), seed_vec=seed_vec, members=member_ids, diffs=diffs_map, quant_meta=meta)
    self.seed_index.register(seed_node)
    with self.registry.lock:
        for n in group:
            self.registry.nodes.pop(n.id, None)
        self.registry.save()
    Ledger.record("SEED_CREATED", seed_node.id, {"from":[n.id for n in group], "members":len(member_ids)})
    append_comp_log({"ts": now_ts(), "seed": seed_node.id, "from":[n.id for n in group], "members": len(member_ids)})
    log(f"Created seed {seed_node.id} from {[n.id for n in group]}")
    return 1

# Method A: sign-agnostic + low-sim sublimation + quarantine
def complementary_sublimate_flexible(self, sim_thresh:float=cfg.default_sim, sim_min:float=cfg.sim_min, allow_sign_flip:bool=True, alpha:float=cfg.freq_alpha, beta:float=cfg.freq_beta, quant_bits:int=cfg.quant_bits, max_iters:int=cfg.default_iters, target_nodes:Optional[int]=None):
    quarantine_path = cfg.quarantine_path

    def load_quarantine():
        try:
            if os.path.exists(quarantine_path):
                return json.load(open(quarantine_path, "r", encoding="utf-8"))
        except Exception:
            pass
        return []

    def save_quarantine(q):
        try:
            safe_write_json(quarantine_path, q)
        except Exception:
            pass

    def node_freq_score(node: EntNode) -> float:
        vals = [freq_store.get(s) for s in node.shards] if node.shards else [0]
        mean = sum(vals) / max(1, len(vals))
        try:
            return math.tanh(alpha * (math.log1p(mean) - beta))
        except Exception:
            return 0.0

    def pair_metric(a: EntNode, b: EntNode, sa: float, sb: float) -> float:
        sim = _cosine(a.vec, b.vec) if (a.vec and b.vec) else 0.0
        sign_bonus = 1.0
        if sa * sb < 0:
            sign_bonus = 1.2
        elif allow_sign_flip:
            sign_bonus = 1.05
        sign_term = 1.0 - abs(sa + sb)
        return sim * (abs(sa) + abs(sb) + 1e-6) * sign_term * sign_bonus * cfg.pair_sim_factor

    quarantine = load_quarantine()
    merged_total = 0
    it = 0
    while it < max_iters:
        it += 1
        nodes = list(self.registry.nodes.values())
        if target_nodes and len(nodes) <= target_nodes:
            break
        if len(nodes) < 2:
            break
        scores = {n.id: node_freq_score(n) for n in nodes}
        nodes_sorted = sorted(nodes, key=lambda x: abs(scores.get(x.id, 0.0)), reverse=True)
        used = set()
        pairs = []
        candidate_list = []
        for i, a in enumerate(nodes_sorted):
            if a.id in used:
                continue
            sa = scores.get(a.id, 0.0)
            best = None; best_metric = 0.0; best_sim = 0.0
            for b in nodes_sorted[i+1:]:
                if b.id in used:
                    continue
                sb = scores.get(b.id, 0.0)
                sim = _cosine(a.vec, b.vec) if (a.vec and b.vec) else 0.0
                m = pair_metric(a, b, sa, sb)
                candidate_list.append((a.id, b.id, sim, sa, sb, m))
                if sim >= sim_thresh and m > best_metric:
                    best_metric = m; best = b; best_sim = sim
                elif best is None and m > best_metric:
                    best_metric = m; best = b; best_sim = sim
            if best:
                if best_sim < sim_min:
                    quarantine.append({"a": a.id, "b": best.id, "sim": best_sim, "metric": best_metric, "iter": it, "retries": 0})
                    continue
                if best_sim < sim_thresh:
                    pairs.append((a, best, best_metric, "low-sim"))
                else:
                    pairs.append((a, best, best_metric, "high-sim"))
                used.add(a.id); used.add(best.id)
        if candidate_list:
            candidate_list_sorted = sorted(candidate_list, key=lambda x: -x[5])[:50]
            log(f"Candidate summary (top 50) iter={it} total_candidates={len(candidate_list)}")
            for ca, cb, sim, sa, sb, m in candidate_list_sorted[:20]:
                _append_debug(f"CAND {ca} {cb} sim={sim:.4f} sa={sa:.4f} sb={sb:.4f} metric={m:.6f}")
        else:
            log("No candidate pairs generated this iteration")
        if not pairs:
            log("No pairs selected this iteration; will attempt quarantine retry or relax-run later")
            break
        for a_node, b_node, metric, tag in pairs:
            try:
                vecs = [v for v in (a_node.vec, b_node.vec) if v is not None]
                if not vecs:
                    continue
                seed_vec = mean_vec(vecs)
                shards = sorted(set(a_node.shards + b_node.shards))
                diffs = []; ids = []
                for n in (a_node, b_node):
                    if n.vec is not None:
                        d = vec_sub(n.vec, seed_vec)
                        if d is not None:
                            diffs.append(d); ids.append(n.shards[0] if n.shards else n.id)
                qvecs, meta = quantize_list(diffs, bits=quant_bits) if diffs else ([], {})
                diffs_map = {ids[i]: qvecs[i] for i in range(len(ids))} if qvecs else {}
                seed = SeedNode(id=uid("seed-"), seed_vec=seed_vec, members=shards, diffs=diffs_map, quant_meta=meta)
                if tag == "low-sim":
                    seed.quant_meta["low_sim_flag"] = True
                    seed.quant_meta["orig_metric"] = metric
                self.seed_index.register(seed)
                with self.registry.lock:
                    self.registry.nodes.pop(a_node.id, None)
                    self.registry.nodes.pop(b_node.id, None)
                    merged_ent = EntNode(id=uid("ent-"), vec=seed_vec, shards=shards, score=(a_node.score + b_node.score))
                    self.registry.nodes[merged_ent.id] = merged_ent
                    try:
                        safe_write_json(self.registry.path, {"nodes": {nid:n.to_dict() for nid,n in self.registry.nodes.items()}})
                    except Exception:
                        pass
                Ledger.record("1PLUSNEG1_MERGE", seed.id, {"from":[a_node.id, b_node.id], "shards": len(shards), "metric": metric, "tag": tag})
                append_comp_log({"ts": now_ts(), "seed": seed.id, "from":[a_node.id, b_node.id], "members": len(shards), "type":"1+-1", "metric": metric, "tag": tag})
                log(f"Merged {a_node.id}+{b_node.id} -> {seed.id} tag={tag} metric={metric:.6f}")
                merged_total += 1
            except Exception:
                log_exc("merge error")
        try:
            self.registry.save(); self.seed_index.save(); Ledger.dump()
        except Exception:
            log_exc("post-merge persist error")
    # dedupe and save quarantine
    if quarantine:
        uniq = {}
        for q in quarantine:
            key = f"{q['a']}::{q['b']}"
            if key not in uniq:
                uniq[key] = q
            else:
                uniq[key]["retries"] = uniq[key].get("retries",0) + 1
        save_quarantine(list(uniq.values()))
        log(f"Quarantine saved {len(uniq)} pairs for retry")
    return {"merged": merged_total, "iters": it, "remaining": len(self.registry.nodes)}

# Method B: force cluster and merge (aggressive)
def force_cluster_and_merge(self, eps:float=0.25, min_members:int=2, quant_bits:int=cfg.quant_bits):
    nodes = list(self.registry.nodes.values())
    if len(nodes) < 2:
        log("force_cluster: not enough nodes")
        return {"forced":0}
    clusters: List[List[EntNode]] = []
    used = set()
    for i, a in enumerate(nodes):
        if a.id in used:
            continue
        cluster = [a]; used.add(a.id)
        for b in nodes[i+1:]:
            if b.id in used:
                continue
            try:
                if _cosine(a.vec, b.vec) >= eps:
                    cluster.append(b); used.add(b.id)
            except Exception:
                continue
        if len(cluster) >= min_members:
            clusters.append(cluster)
    forced = 0
    for cl in clusters:
        try:
            vecs = [n.vec for n in cl if n.vec is not None]
            if not vecs:
                continue
            seed_vec = mean_vec(vecs)
            shards = []
            diffs = []; ids = []
            for n in cl:
                shards.extend(n.shards)
                if n.vec is not None:
                    d = vec_sub(n.vec, seed_vec)
                    if d is not None:
                        diffs.append(d); ids.append(n.shards[0] if n.shards else n.id)
            qvecs, meta = quantize_list(diffs, bits=quant_bits) if diffs else ([], {})
            diffs_map = {ids[i]: qvecs[i] for i in range(len(ids))} if qvecs else {}
            seed = SeedNode(id=uid("seed-"), seed_vec=seed_vec, members=sorted(set(shards)), diffs=diffs_map, quant_meta=meta)
            seed.quant_meta["force_cluster_flag"] = True
            self.seed_index.register(seed)
            with self.registry.lock:
                for n in cl:
                    self.registry.nodes.pop(n.id, None)
                merged_ent = EntNode(id=uid("ent-"), vec=seed_vec, shards=sorted(set(shards)), score=sum(n.score for n in cl))
                self.registry.nodes[merged_ent.id] = merged_ent
                try:
                    safe_write_json(self.registry.path, {"nodes": {nid:n.to_dict() for nid,n in self.registry.nodes.items()}})
                except Exception:
                    pass
            Ledger.record("FORCE_CLUSTER_MERGE", seed.id, {"from":[n.id for n in cl], "members": len(shards), "eps": eps})
            append_comp_log({"ts": now_ts(), "seed": seed.id, "from":[n.id for n in cl], "members": len(shards), "type":"force-cluster", "eps": eps})
            log(f"Force-cluster created seed {seed.id} from {[n.id for n in cl]} eps={eps}")
            forced += 1
        except Exception:
            log_exc("force-cluster merge error")
    try:
        self.registry.save(); self.seed_index.save(); Ledger.dump()
    except Exception:
        log_exc("force-cluster persist error")
    return {"forced": forced}

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

LazyExpander (unchanged)

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

class LazyExpander: def init(self, seed_index:SeedIndex): self.seed_index = seed_index self.cache = {} self.lock = threading.RLock()

def quick_holo(self, seed:SeedNode, query_vec:Optional[List[float]]=None, alpha:float=0.6):
    if query_vec is None:
        emb = seed.seed_vec
    else:
        emb = vec_add([x*alpha for x in query_vec], [x*(1-alpha) for x in seed.seed_vec]) if seed.seed_vec else query_vec
    delta = _cosine(seed.seed_vec, emb) if seed.seed_vec else 0.0
    holo = {"id": uid("h-"), "embedding": emb, "confidence": 0.75, "seed": seed.id, "delta": delta}
    Ledger.record("SEED_QUICK", holo["id"], {"seed":seed.id})
    return holo

def expand(self, seed:SeedNode, top_n:int=6):
    with self.lock:
        if seed.id in self.cache:
            return self.cache[seed.id]
    members = []
    if seed.quant_meta and seed.diffs:
        for mid in seed.members[:top_n]:
            q = seed.diffs.get(mid)
            if q:
                try:
                    diff = dequantize(q, seed.quant_meta)
                    emb = vec_add(seed.seed_vec, diff)
                    members.append({"id":mid,"embedding":emb})
                except Exception:
                    members.append({"id":mid})
            else:
                members.append({"id":mid})
    else:
        for mid in seed.members[:top_n]:
            members.append({"id":mid})
    res = {"seed":seed.id,"members":members,"ts":time.time()}
    with self.lock:
        self.cache[seed.id] = res
    Ledger.record("SEED_EXPAND", seed.id, {"members":len(members)})
    return res

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

Self-bootstrap: ingest shards or inject samples

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

def _ingest_from_shards(registry:EntRegistry, max_import:int=1024): if not os.path.isdir(cfg.shard_dir): return 0 files = sorted(os.listdir(cfg.shard_dir)) imported = 0 seen_hashes = set() for fname in files[:max_import]: fpath = os.path.join(cfg.shard_dir, fname) try: with open(fpath, "rb") as f: payload = f.read() except Exception: continue import hashlib h = hashlib.sha256(payload).hexdigest() if h in seen_hashes: continue seen_hashes.add(h) vec = [] for i in range(cfg.dim): idx = (i * 2) % len(h) try: b = int(h[idx:idx+2], 16) except Exception: b = 0 val = ((b / 255.0) * 0.6) - 0.3 vec.append(val) nid = uid("ent-") shard_id = fname node = EntNode(id=nid, vec=vec, shards=[shard_id]) registry.register(node) imported += 1 return imported

_injected_counter_lock = threading.RLock() _injected_counter_path = os.path.join(cfg.data_dir, "injected_count.json")

def _get_injected_count(): try: if os.path.exists(_injected_counter_path): with open(_injected_counter_path, "r", encoding="utf-8") as f: return int(json.load(f).get("count", 0)) except Exception: pass return 0

def _set_injected_count(n:int): try: safe_write_json(_injected_counter_path, {"count": n}) except Exception: pass

def _inject_animation_samples(registry:EntRegistry, count:int = 24): samples = [ "pocket infinite storage", "memory bread copy restore", "memory camera snapshot replay", "time cloth restore state", "memory disk compress replay", "memory capsule compress small", "holographic pocket seed aggregator", "seed singularity compressed origin", "lazy expansion reconstruct" ] injected = 0 i = 0 count = min(count, cfg.max_auto_inject) with _injected_counter_lock: already = _get_injected_count() to_inject = max(0, count - already) while injected < to_inject: s = samples[i % len(samples)] + f" sample-{already+injected}" import hashlib h = hashlib.sha256(s.encode('utf-8')).digest() vec = [] for k in range(cfg.dim): b = h[k % len(h)] val = ((b / 255.0) * 0.6) - 0.3 vec.append(val) nid = uid("ent-") registry.register(EntNode(id=nid, vec=vec, shards=[f"anim-{already+injected}"])) injected += 1 i += 1 if injected > 0: _set_injected_count(already + injected) return injected

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

Backup & rollback helpers

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

def _prune_backups(): try: items = sorted(os.listdir(cfg.backup_dir)) if len(items) <= cfg.max_backup_keep: return for old in items[:-cfg.max_backup_keep]: p = os.path.join(cfg.backup_dir, old) try: if os.path.isdir(p): shutil.rmtree(p) else: os.remove(p) except Exception: pass except Exception: pass

def backup_state(): try: ts = int(time.time()) dest = os.path.join(cfg.backup_dir, f"backup{ts}") os.makedirs(dest, exist_ok=True) for p in (cfg.ent_path, cfg.seed_path, cfg.ledger_path, cfg.comp_log): if os.path.exists(p): try: shutil.copy2(p, dest) except Exception: pass log(f"Backup created at {dest}") _prune_backups() return dest except Exception as e: log(f"Backup failed: {e}") return None

def rollback_from_backup(backup_dir: str): try: for fname in ("ents.json","seeds.json","ledger.json","compression_log.json"): src = os.path.join(backup_dir, fname) dst = os.path.join(cfg.data_dir, fname) if os.path.exists(src): try: shutil.copy2(src, dst) except Exception: pass log(f"Rollback applied from {backup_dir}") except Exception as e: log(f"Rollback failed: {e}")

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

Quarantine retry & sublimation (embedded)

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

def retry_quarantine_and_sublimate(compressor: Compressor, top_k:int=20, relax_steps:List[float]=[0.55,0.50], interp_steps:int=5, perturb_sigma:float=0.02, cluster_merge:bool=False): """ Retry and 'sublimate' pairs from quarantine. Parameters: compressor: Compressor instance top_k: process top_k pairs by metric relax_steps: sequence of sim thresholds to try interp_steps: number of interpolation steps between vectors (inclusive) perturb_sigma: gaussian noise stddev for augmentation cluster_merge: if True, cluster related quarantine pairs first (not implemented here) """ qpath = cfg.quarantine_path if not os.path.exists(qpath): log("retry_quarantine: no quarantine file") return {"tried":0,"merged":0} try: q = json.load(open(qpath, "r", encoding="utf-8")) except Exception: log("retry_quarantine: failed to load quarantine") return {"tried":0,"merged":0} q_sorted = sorted(q, key=lambda x: -float(x.get("metric",0)))[:top_k] tried = 0; merged = 0; new_quarantine = []

def attempt_merge_by_vec(a_id, b_id, cand_vec, sim_target):
    nonlocal merged
    reg = compressor.registry
    a = reg.nodes.get(a_id); b = reg.nodes.get(b_id)
    if not a or not b:
        return False
    sim_a = _cosine(a.vec, cand_vec)
    sim_b = _cosine(b.vec, cand_vec)
    if sim_a >= sim_target and sim_b >= sim_target:
        shards = sorted(set(a.shards + b.shards))
        diffs=[]; ids=[]
        for n in (a,b):
            if n.vec is not None:
                d = vec_sub(n.vec, cand_vec)
                if d is not None:
                    diffs.append(d); ids.append(n.shards[0] if n.shards else n.id)
        qvecs, meta = quantize_list(diffs, bits=cfg.quant_bits) if diffs else ([],{})
        diffs_map = {ids[i]: qvecs[i] for i in range(len(ids))} if qvecs else {}
        seed = SeedNode(id=uid("seed-"), seed_vec=cand_vec, members=shards, diffs=diffs_map, quant_meta=meta)
        seed.quant_meta["sublimated_flag"] = True
        compressor.seed_index.register(seed)
        with compressor.registry.lock:
            compressor.registry.nodes.pop(a.id, None)
            compressor.registry.nodes.pop(b.id, None)
            merged_ent = EntNode(id=uid("ent-"), vec=cand_vec, shards=shards, score=(a.score + b.score))
            compressor.registry.nodes[merged_ent.id] = merged_ent
            try:
                safe_write_json(compressor.registry.path, {"nodes": {nid:n.to_dict() for nid,n in compressor.registry.nodes.items()}})
            except Exception:
                pass
        Ledger.record("SUBLIMATE_MERGE", seed.id, {"from":[a.id,b.id],"sim_a":sim_a,"sim_b":sim_b,"target":sim_target})
        append_comp_log({"ts": now_ts(), "seed": seed.id, "from":[a.id,b.id], "members": len(shards), "type":"sublimate", "sim_a":sim_a, "sim_b":sim_b, "target":sim_target})
        log(f"Sublimated merge {a.id}+{b.id} -> {seed.id} sim_a={sim_a:.4f} sim_b={sim_b:.4f} target={sim_target:.3f}")
        merged += 1
        return True
    return False

for item in q_sorted:
    tried += 1
    a = item.get("a"); b = item.get("b")
    base_metric = float(item.get("metric",0))
    reg = compressor.registry
    na = reg.nodes.get(a); nb = reg.nodes.get(b)
    if not na or not nb:
        continue
    vecs = []
    for t in range(interp_steps+1):
        alpha = t / max(1, interp_steps)
        cand = [(1-alpha)*x + alpha*y for x,y in zip(na.vec, nb.vec)]
        vecs.append(cand)
        for p in range(2):
            pert = [v + random.gauss(0, perturb_sigma) for v in cand]
            vecs.append(pert)
    merged_flag = False
    for sim_target in relax_steps:
        if merged_flag: break
        for cand in vecs:
            if attempt_merge_by_vec(a, b, cand, sim_target):
                merged_flag = True
                break
    if not merged_flag:
        item["retries"] = item.get("retries",0) + 1
        if item["retries"] < cfg.quarantine_retry_limit:
            new_quarantine.append(item)
        else:
            item["exhausted"] = True
            new_quarantine.append(item)
try:
    safe_write_json(cfg.quarantine_path, new_quarantine)
except Exception:
    pass
try:
    compressor.registry.save(); compressor.seed_index.save(); Ledger.dump()
except Exception:
    pass
return {"tried":tried,"merged":merged}

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

Autonomous pipeline driver & watcher

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

class AutoDriver: def init(self): self.registry = EntRegistry() self.seed_index = SeedIndex() self.comp = Compressor(self.registry, self.seed_index) self.lock = threading.RLock() self.running = False self.sim = cfg.default_sim self.iters = cfg.default_iters self.min_group = cfg.min_group self.last_activity = time.time() self.metrics = {"runs":0,"merged":0,"seeds":0,"injected":_get_injected_count()} self._shutdown_requested = False

def request_shutdown(self):
    self._shutdown_requested = True

def ensure_data(self):
    if not self.registry.nodes:
        imported = _ingest_from_shards(self.registry, max_import=1024)
        if imported:
            log(f"AutoDriver: imported {imported} shards")
        else:
            injected = _inject_animation_samples(self.registry, count=cfg.max_auto_inject)
            if injected:
                self.metrics["injected"] += injected
            log(f"AutoDriver: injected {injected} samples")

def run_once(self, relax_sim: Optional[float] = None, force_cluster: bool = False, cluster_eps: float = 0.25):
    with self.lock:
        if self._shutdown_requested:
            log("AutoDriver: shutdown requested; skipping run")
            return
        self.ensure_data()
        if not self.registry.nodes:
            log("AutoDriver: no nodes to process")
            return
        backup = _backup_state()
        try:
            sim_to_use = relax_sim if relax_sim is not None else self.sim
            log(f"AutoDriver: running fusion with sim={sim_to_use:.3f} nodes={len(self.registry.nodes)}")
            res1 = self.comp.complementary_sublimate_flexible(sim_thresh=sim_to_use, sim_min=cfg.sim_min, max_iters=2)
            res2 = self.comp.build_seeds(sim_thresh=sim_to_use, min_group=self.min_group, quant_bits=cfg.quant_bits)
            merged = res1.get("merged",0) + res2.get("created",0)
            forced = 0
            # If quarantine exists and no merges, attempt automatic quarantine retry/sublimation
            try:
                if merged == 0 and os.path.exists(cfg.quarantine_path):
                    log("AutoDriver: attempting automatic quarantine retry/sublimation")
                    retry_res = retry_quarantine_and_sublimate(self.comp, top_k=60, relax_steps=[0.55,0.50,0.45], interp_steps=6, perturb_sigma=0.03)
                    log(f"AutoDriver: quarantine retry result: {retry_res}")
                    merged += retry_res.get("merged",0)
            except Exception:
                log_exc("auto quarantine retry error")
            if merged == 0 and force_cluster:
                log("AutoDriver: no merges from flexible path; running force-cluster as backup")
                forced_res = self.comp.force_cluster_and_merge(eps=cluster_eps, min_members=2, quant_bits=cfg.quant_bits)
                forced = forced_res.get("forced", 0)
            self.metrics["runs"] += 1
            self.metrics["merged"] += merged + forced
            self.metrics["seeds"] += res2.get("created",0)
            self.last_activity = time.time()
            log(f"AutoDriver: run completed merged={merged} forced={forced} seeds_created={res2.get('created',0)} remaining_nodes={len(self.registry.nodes)}")
            if merged + forced > 0:
                self.sim = min(0.995, self.sim + 0.01)
            else:
                self.sim = max(0.50, self.sim - 0.02)
            self.registry.save(); self.seed_index.save(); Ledger.dump()
        except Exception as e:
            log(f"AutoDriver: error during run: {e}")
            log_exc("run_once exception")
            if backup:
                rollback_from_backup(backup)

def watch_loop(self, poll_interval: float = cfg.poll_interval):
    self.running = True
    log("AutoDriver: entering watch loop")
    self.run_once()
    while self.running and not self._shutdown_requested:
        if os.path.exists(cfg.shutdown_signal):
            log("AutoDriver: shutdown signal detected; exiting watch loop")
            break
        try:
            new_found = False
            for fname in os.listdir(cfg.shard_dir):
                fpath = os.path.join(cfg.shard_dir, fname)
                try:
                    mtime = os.path.getmtime(fpath)
                    if mtime > self.last_activity - 1:
                        new_found = True
                        break
                except Exception:
                    continue
            if new_found:
                log("AutoDriver: new shard detected -> triggering run")
                self.run_once()
            if time.time() - self.last_activity > cfg.idle_run_seconds:
                log("AutoDriver: idle timeout -> periodic run")
                self.run_once()
        except Exception as e:
            log(f"AutoDriver: watch loop error: {e}")
            log_exc("watch_loop exception")
        time.sleep(poll_interval)
    self.running = False
    log("AutoDriver: watch loop ended")

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

Optional minimal HTTP status server

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

class StatusHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path.startswith("/status"): try: body = { "ts": now_ts(), "ents": len(EntRegistry().nodes), "seeds": len(SeedIndex().seeds), "metrics": auto_driver.metrics if 'auto_driver' in globals() else {} } self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(body).encode("utf-8")) except Exception: self.send_response(500); self.end_headers() elif self.path.startswith("/metrics"): try: m = auto_driver.metrics if 'auto_driver' in globals() else {} lines = [] for k,v in m.items(): lines.append(f"qem_{k} {v}") self.send_response(200) self.send_header("Content-Type", "text/plain; version=0.0.4") self.end_headers() self.wfile.write("\n".join(lines).encode("utf-8")) except Exception: self.send_response(500); self.end_headers() else: self.send_response(404); self.end_headers()

def log_message(self, format, *args):
    return

def start_status_server(port:int): if port <= 0: return None try: server = socketserver.ThreadingTCPServer(("0.0.0.0", port), StatusHandler) t = threading.Thread(target=server.serve_forever, daemon=True) t.start() log(f"Status server listening on port {port}") return server except Exception as e: log(f"Failed to start status server: {e}") return None

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

Entrypoint

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

def main(argv): parser = argparse.ArgumentParser(description="QEM Autonomous Final Embedded") parser.add_argument("--debug-run-once", action="store_true", help="Run one detailed debug iteration then exit") parser.add_argument("--no-auto", action="store_true", help="Do not auto-run (exit)") parser.add_argument("--relax-sim", type=float, default=None, help="Temporarily relax sim threshold for debug run") parser.add_argument("--force-cluster", action="store_true", help="If no merges, run force-cluster as backup") parser.add_argument("--cluster-eps", type=float, default=0.25, help="Eps for force-cluster") args = parser.parse_args(argv)

global auto_driver
auto_driver = AutoDriver()
server = start_status_server(cfg.status_port)

if args.no_auto:
    log("Auto-run disabled by --no-auto; exiting")
    return

if args.debug_run_once:
    log("DEBUG RUN ONCE: starting detailed single run")
    try:
        with open(cfg.debug_log, "a", encoding="utf-8") as f:
            f.write(f"\n=== DEBUG RUN START {now_ts()} ===\n")
    except Exception:
        pass
    auto_driver.run_once(relax_sim=args.relax_sim, force_cluster=args.force_cluster, cluster_eps=args.cluster_eps)
    log("DEBUG RUN ONCE: finished; check qem_cloud_data/debug_log.txt for details")
    if server:
        try:
            server.shutdown(); server.server_close()
        except Exception:
            pass
    return

try:
    auto_driver.watch_loop(poll_interval=cfg.poll_interval)
except KeyboardInterrupt:
    log("KeyboardInterrupt received; shutting down")
finally:
    if server:
        try:
            server.shutdown(); server.server_close()
        except Exception:
            pass
    Ledger.dump()
    log("Autonomous process exiting")

if name == "main": main(sys.argv[1:])