app.py

"""
🧠 LHC-0 — Bi-Hemispheric Brain (Demo v2 — reliability edition)
الشق الأيسر (دلالي/رمزي — مرمّز P2 slot-aware) + الشق الأيمن (إنجرامات 5M)
+ جسم ثفني بقناتين + **ثقة معايرة (vmPFC) + امتناع «لا أعرف» (ACC) + بوابة ترسيخ v2**.

Run:  python app.py
"""
import json
import os
import re

import numpy as np

from lhc_core import LHC0
from phase_c_encoder import TrainableEncoder, hashed_features
from phase_e_callosum import BiCallosum, fuse
from phase_p1_reliability import (ConsolidationGateV2, ConfidenceCalibrator,
                                  build_canonical, OOD_TOPICS_EN, OOD_TOPICS_AR)
from phase_p2_slot import hashed_features_slot
from phase_w2_episodic import EpisodicMemory, SleepConsolidation, SelfModel
from phase_w3_workspace import GlobalWorkspace, Proposal
from phase_c_augment import SYN, augment

F_DIM = 2 ** 15
HERE = os.path.dirname(os.path.abspath(__file__))
TAU = 0.5  # عتبة الامتناع (P1: عند τ=0.5 — دقة انتقائية 96.4% وصفر إجابات وهمية OOD)


# ============================================================
# Engine — the whole brain (load once, answer forever)
# ============================================================
class Brain:
    def __init__(self):
        self.kb = json.load(open(os.path.join(HERE, "knowledge_bank_superbrain.json"),
                                 encoding="utf-8"))
        # ---- primary left hemisphere: W1 encoder (big-data) with P2 fallback
        w1_path = os.path.join(HERE, "phase_w1_W.npz")
        p2_path = os.path.join(HERE, "phase_p2_W.npz")
        chosen = w1_path if os.path.exists(w1_path) else p2_path
        z2 = np.load(chosen)
        self.encoder_name = os.path.basename(chosen)
        self.bi = TrainableEncoder(F=int(z2["F"]), d=int(z2["d"]),
                                   tau=float(z2["tau"]), W=z2["W"],
                                   features_fn=hashed_features_slot)
        self.lhc = LHC0(kb=self.kb, encoder=self.bi)
        self.gate = ConsolidationGateV2(self.bi, threshold=0.687, content_tau=0.5,
                                        canon=build_canonical(SYN))
        self.cal = self._fit_calibrator()

        # ---- W2: episodic memory (hippocampus) + self-model (vmPFC) -----
        self.episodic = EpisodicMemory(encoder=self.bi)
        self.consolidation = SleepConsolidation(self.episodic, self.lhc, gate=self.gate)
        self.self_model = SelfModel(self.lhc, self.episodic,
                                    confidence_fn=lambda q: self.answer(q)["confidence"])

        # ---- W3: global workspace + attention (thalamus / FEF-IPS / ACC) -----
        self.workspace = GlobalWorkspace()
        self._depth = 0

        # ---- legacy bi-hemispheric vote panel (Phase E) --------------------
        z = np.load(os.path.join(HERE, "phase_e_bundle.npz"))
        self.legacy = TrainableEncoder(F=F_DIM, d=128, W=z["W"])
        self.lhc_old = LHC0(kb=self.kb, encoder=self.legacy)
        self.bical = BiCallosum()
        self.bical.dorsal.X, self.bical.dorsal.A = z["X_d"], z["A_d"]
        self.bical.ventral.X, self.bical.ventral.A = z["X_v"], z["A_v"]
        self.Y = z["Y"]
        self.n_right = self.Y.shape[0]

    # ---- vmPFC: fit P(correct | score, gap) on synthetic self-test ---------
    def _fit_calibrator(self):
        nrs = np.random.RandomState(0)
        S, G, Y = [], [], []
        for e in self.kb:
            for _ in range(2):
                q = augment(e["prompt"], nrs)
                ent, s, g = self.lhc.memory.read(q)
                S.append(s); G.append(g)
                Y.append(1.0 if ent["prompt"] == e["prompt"] else 0.0)
        for t_en, t_ar in zip(OOD_TOPICS_EN, OOD_TOPICS_AR):
            for q in (f"What is fact number 900001 about {t_en}?",
                      f"ما هي الحقيقة رقم 900002 حول {t_ar}؟"):
                _, s, g = self.lhc.memory.read(q)
                S.append(s); G.append(g); Y.append(0.0)
        feats = np.stack([S, G], axis=1)
        return ConfidenceCalibrator().fit(feats, np.array(Y))

    def _try_calc(self, q):
        """basal-ganglia calculator candidacy: only for real expressions."""
        s = (q or "").strip().rstrip("؟?!. ")
        for pre in ("احسب", "كم يساوي", "ما ناتج", "حل", "compute", "calculate", "ما هو"):
            if s.startswith(pre):
                s = s[len(pre):].strip()
        s = s.rstrip("؟?!. ").strip()
        if (re.fullmatch(r"[0-9\s\+\-\*/\^\(\)\.eE]+", s or "")
                and re.search(r"[0-9]", s) and re.search(r"[\+\-\*/\^]", s)):
            try:
                return str(self.lhc.compute(s)["value"])
            except Exception:
                return None
        return None

    def answer(self, q):
        """W3: every pathway proposes → the global workspace broadcasts ONE
        winner (honest dual report on ACC conflict, empty display on abstain).
        Dict contract preserved (+ source / workspace keys)."""
        self._depth += 1
        try:
            entry, score, gap = self.lhc.memory.read(q)
            conf = float(self.cal.predict(np.array([[score, gap]]))[0])
            knows = conf >= TAU
            props = []
            if self._depth == 1:                    # guard: no self-model re-entry
                s = self.self_model.answer(q)
                if s is not None:
                    props.append(Proposal("self_model", s, 0.95))
            cv = self._try_calc(q)
            if cv is not None:
                props.append(Proposal("calc", cv, 0.99))
            if knows:
                props.append(Proposal("semantic", entry["response"], conf))
            b = self.workspace.resolve(props, text=q)
            self.episodic.log("ask", q,
                              response=b.winner.content if b.winner else None,
                              salience=conf)
            out = {"response": b.winner.content if b.winner else None,
                   "abstain": b.winner is None,
                   "confidence": b.winner.confidence if b.winner else conf,
                   "raw_score": score, "gap": gap,
                   "prompt_matched": entry["prompt"],
                   "domain": entry.get("domain", "") if b.winner is None or
                             b.winner.source == "semantic" else
                             {"self_model": "🪞 ذات", "calc": "🧮 حساب"}.get(b.winner.source, ""),
                   "source": b.winner.source if b.winner else None,
                   "workspace": b.reason}
            if b.winner is not None and b.conflict_flag:
                others = [p for p in b.candidates if p.source != b.winner.source]
                extra = " ⚡ ".join(f"«{p.source}»: {p.content}" for p in others)
                out["response"] = (f"{b.winner.content}\n\n"
                                   f"⚠️ تعارض ACC — اقتراح آخر قريب يُعرض ولا يُخفى: {extra}")
                out["source"] = b.winner.source + " vs " + ",".join(p.source for p in others)
            return out
        finally:
            self._depth -= 1

    def votes(self, q):
        e = self.legacy.encode(q)
        x = np.zeros(F_DIM, dtype=np.float32)
        idx, val = hashed_features(q, F_DIM)
        x[idx] = val
        s_left = np.stack(self.lhc_old.memory.codes) @ e
        s_right = self.bical.right_scores(x, e, self.Y)
        if len(s_left) > self.n_right:
            pad = np.full(len(s_left) - self.n_right,
                          float(s_right.min()) - 1.0, dtype=s_right.dtype)
            s_right = np.concatenate([s_right, pad])
            has_right = np.zeros(len(s_left), dtype=bool)
            has_right[:self.n_right] = True
        else:
            has_right = np.ones(len(s_left), dtype=bool)
        s_fused = fuse(s_left, s_right, has_right)

        def top(s):
            i = int(np.argmax(s))
            return self.lhc_old.memory.entries[i]["prompt"]
        return {"left": top(s_left), "right": top(s_right), "fusion": top(s_fused)}

    def compute(self, expr):
        return self.lhc.compute(expr)

    def teach(self, prompt, response):
        """Learning through the hippocampal gate v2: merge duplicates, keep news."""
        d = self.gate.check(self.lhc.memory, prompt)
        self.episodic.log("learn", prompt, response=response,
                          salience=0.8 if d["is_new"] else 0.6)
        if not d["is_new"]:
            j = d["nearest"]
            self.lhc.memory.entries[j]["response"] = response   # تحديث المعنى
            return ("🔁 ذكرى مدموجة (تكرار) — حدّثت الإجابة القديمة",
                    f"تشابه={d['score']:.2f} • محتوى={d['content']:.2f} • سبب={d['reason']}")
        self.lhc.memory.write({"prompt": prompt, "response": response,
                               "domain": "🧠 ذاكرة مكتسبة لحظيًا"})
        return (f"✅ ذكرى جديدة مخزّنة ({len(self.lhc.memory.codes)} ذاكرة الآن)",
                f"تشابه={d['score']:.2f} • سبب={d['reason']}")


brain = Brain()


# ============================================================
# Gradio UI
# ============================================================
def ui_answer(q):
    if not q or not q.strip():
        return "اكتب سؤالًا أولًا...", "", "", "", ""
    a = brain.answer(q.strip())
    v = brain.votes(q.strip())
    conf_pct = f"**الثقة المعايرة: {a['confidence']:.0%}** • درجة خام={a['raw_score']:.2f} • فجوة={a['gap']:.2f}"
    if a["abstain"]:
        ans = ("🧠 **لا أعرف** — ثقتي منخفضة بهذا السؤال، فأمتنع عن الإجابة بدل التخمين.\n"
               "💡 يمكنك تعليمي إياه في تبويب «تعلّم ذكرى جديدة».")
    else:
        ans = a["response"]
    votes_md = (
        f"**اليسار (دلالي):** {v['left']}\n\n"
        f"**اليمين (إنجرامات):** {v['right']}\n\n"
        f"**الدمج (الجسم الثفني):** {v['fusion']}"
    )
    decision = "أجيب ✓" if not a["abstain"] else "أرفض (لا أعرف) ✗"
    return ans, votes_md, conf_pct, decision, a["domain"]


def ui_calc(expr):
    if not expr or not expr.strip():
        return "", ""
    try:
        r = brain.compute(expr.strip())
        return str(r["value"]), f"{r['steps']} خطوة زمنية (عمق تفكير تكيفي)"
    except Exception as ex:
        return f"خطأ: {ex}", ""


def ui_teach(prompt, response):
    if not prompt.strip() or not response.strip():
        return "اكتب السؤال والإجابة", ""
    return brain.teach(prompt.strip(), response.strip())


def ui_day():
    evs = [e for e in brain.episodic.events if not e["archived"]]
    today = brain.episodic.recall_today()
    lines = [f"📅 اليوم: {len(today)} حدثًا | الكل: {len(evs)} حدثًا حيًّا"]
    for e in today[-12:]:
        tag = {"learn": "🧠 تعلّم", "ask": "💬 سؤال", "event": "⚡ حدث"}.get(e["kind"], e["kind"])
        conf = f" (ثقة {e['salience']:.0%})" if e["kind"] == "ask" else ""
        lines.append(f"  [{e['t'][11:16]}] {tag}: {e['content'][:70]}{conf}")
    return "\n".join(lines)


def ui_replay():
    r = brain.consolidation.replay_cycle()
    return (f"🌙 دورة ترسيخ اكتملت: {r['promoted']} مُرسَّخ • {r['merged']} مدموج • "
            f"{r['decayed']} مُﲻ (تلاشٍ صحي) • الذاكرة الدلالية: "
            f"{r['semantic_before']}→{r['semantic_after']}")


def ui_self(q):
    if not q or not q.strip():
        return "اسألني: من أنت؟ / ماذا يمكنك؟ / ماذا تعلمت اليوم؟ / كيف حالك؟ / متى تعلمت ...؟"
    a = brain.self_model.answer(q.strip())
    return a if a is not None else "هذا ليس سؤالًا عن ذاتي — اسألني في تبويب «اسأل العقل»."


def ui_workspace(q):
    if not q or not q.strip():
        return "اكتب سؤالًا لترى التنافس على العرض العالمي.", ""
    a = brain.answer(q.strip())
    return brain.workspace.display_report(), (a["response"] or "— العرض فارغ: لا بث (امتناع صادق) —")


def build_demo():
    import gradio as gr
    RTL_CSS = "body, .gradio-container { direction: rtl; text-align: right; }"
    with gr.Blocks(title="🧠 LHC-0 — العقل ثنائي الشقين", css=RTL_CSS) as demo:
        gr.Markdown(
            "# 🧠 LHC-0 — العقل ثنائي الشقين (Bi-Hemispheric Brain)\n"
            "### الشق الأيسر (رمزي/دلالي — مرمّز **W1** المدرَّب على 359K ترجمة) + الشق الأيمن (إنجرامات 5M) "
            "+ جسم ثفني بقناتين — **بدون طبقات متتالية، ذاكرة مدى الحياة، وثقة تعرف متى تقول: لا أعرف**\n\n"
            "💡 **اسأل بأي لغة** — يفهم الترجمة الحيّة (سؤال عربي عن حقيقة مخزّنة بالإنجليزية والعكس)."
        )
        with gr.Tab("💬 اسأل العقل"):
            q = gr.Textbox(label="سؤالك (عربي أو إنجليزي)",
                           placeholder="ما هي عاصمة مصر؟ / Who created you؟")
            btn = gr.Button("اسأل", variant="primary")
            ans = gr.Textbox(label="الإجابة", lines=3)
            votes = gr.Markdown(label="تصويت النصفين (Phase E)")
            conf = gr.Markdown(label="الثقة")
            dec = gr.Textbox(label="قرار العقل")
            dom = gr.Textbox(label="المجال")
            btn.click(ui_answer, [q], [ans, votes, conf, dec, dom])
            q.submit(ui_answer, [q], [ans, votes, conf, dec, dom])
        with gr.Tab("🧮 الحساب الفعلي (BODMAS)"):
            e = gr.Textbox(label="تعبير حسابي", value="(12 + 8) * (7 - 5) / 4")
            b2 = gr.Button("احسب")
            val = gr.Textbox(label="النتيجة")
            stp = gr.Textbox(label="خطوات الزمن")
            b2.click(ui_calc, [e], [val, stp])
        with gr.Tab("🧠 تعلّم ذكرى جديدة (بوابة الحُصين v2)"):
            gr.Markdown("كل ذكرى تمرّ ببوابة الترسيخ: **التكرار يُدمج، والجديد يُخزَّن** — "
                        "مع عرض السبب (فصل الأنماط).")
            p = gr.Textbox(label="سؤال جديد", placeholder="ما هو أفضل مشروع؟")
            r = gr.Textbox(label="إجابته", placeholder="LHC-0 هو الأفضل!")
            b3 = gr.Button("علّم العقل", variant="primary")
            out = gr.Textbox(label="الحالة")
            why = gr.Textbox(label="تفاصيل البوابة")
            b3.click(ui_teach, [p, r], [out, why])
        with gr.Tab("📅 يوميات العقل (ذاكرة عرضية)"):
            gr.Markdown("الأحداث بختم زمني + **الترسيخ الليلي** (النوم ينقل المهم إلى الذاكرة الدلالية ويُخفي التافه).")
            b4 = gr.Button("🔄 اعرض يومي")
            day = gr.Textbox(label="السجل العرضي", lines=10)
            b4.click(ui_day, [], [day])
            b5 = gr.Button("🌙 شغّل دورة الترسيخ (نوم)", variant="primary")
            rep = gr.Textbox(label="تقرير الترسيخ")
            b5.click(ui_replay, [], [rep])
        with gr.Tab("🪞 اسأل الذات (وعي بالذات)"):
            gr.Markdown("اسأل العقل عن نفسه وقدراته وحدود معرفته — التقرير مولَّد من حالته **الحقيقية**.")
            q2 = gr.Textbox(label="سؤال عن الذات", placeholder="من أنت؟ / ماذا تعلمت اليوم؟ / متى تعلمت ...؟")
            b6 = gr.Button("اسأل", variant="primary")
            self_out = gr.Textbox(label="الجواب", lines=8)
            b6.click(ui_self, [q2], [self_out])
            q2.submit(ui_self, [q2], [self_out])
        with gr.Tab("📡 المساحة العالمية (W3)"):
            gr.Markdown("كل مسار يعرض اقتراحه → **تنافس** (الانتباه يرجّح كفة المهمة) → فائز واحد يُبثّ للكل. "
                        "مراقب التعارض (ACC) يكشف خلاف المسارات فيعرضها بصراحة — لا إخفاء ولا تخمين.")
            q3 = gr.Textbox(label="سؤال", placeholder="احسب 25*13 / من أنت؟ / ما هو أطول نهر؟")
            b7 = gr.Button("شغّل المساحة", variant="primary")
            wr = gr.Markdown(label="تقرير المساحة (التنافس)")
            wa = gr.Textbox(label="الإجابة المبثوثة", lines=3)
            b7.click(ui_workspace, [q3], [wr, wa])
            q3.submit(ui_workspace, [q3], [wr, wa])
        gr.Markdown(
            f"**إحصائيات:** مرمّز **{brain.encoder_name}** • {brain.n_right} إنجرام حقيقي • "
            f"{int(np.load(os.path.join(HERE, 'phase_e_bundle.npz'))['global_ids'].size)} عصيبة نشطة "
            "من فضاء 5,000,000 • مترجم ثفني بقناتين • **معايرة ثقة + امتناع (صفر إجابات وهمية OOD)** "
            "+ **ترجمة حية** (TED-محجوز 84%) + **ذاكرة عرضية ووعي بالذات** (W2) + **المساحة العالمية والانتباه** (W3)\n\n"
            "المشروع مبني على تشريح الشق الأيسر للمخ (بروكا/فيرنيكه/التلفيف الزاوي/العقد القاعدية/المخيخ/الحُصين/ACC/vmPFC) "
            "— راجع التقرير التقني (holdout صياغات 100% • تغطية 75% عند دقة 95% • 359K ترجمة من 8 مجالات)."
        )
    return demo


if __name__ == "__main__":
    demo = build_demo()
    demo.launch(server_name="0.0.0.0", server_port=7860)