Update app.py

app.py

@@ -1,70 +1,225 @@
-import gradio as gr
-from huggingface_hub import InferenceClient
-
-
-def respond(
-    message,
-    history: list[dict[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-    hf_token: gr.OAuthToken,
-):
+# ================================================================
+# === INT STRATEGIC BRAIN v4 — Enemy Analysis & Tab Integration ===
+# ================================================================
+# Paste this block inside your app. For UI, insert the Gradio tab block
+# within your existing `with gr.Blocks(...) as demo:` where other tabs appear.
+# ================================================================
+
+import math
+import random
+import datetime
+import json
+from collections import Counter
+
+# -------------------------
+# INTBrainV4 core extension
+# -------------------------
+class INTBrainV4:
+    def __init__(self, base=None, memory_file="int_brain_v4.json"):
+        self.base = base
+        self.memory_file = memory_file
+        self.seed = 42
+        random.seed(self.seed)
+
+    def _now(self):
+        return datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
+
+    def infer_intent(self, text):
+        """Lightweight intent inference (doctrinal, non-operational)."""
+        t = text.lower()
+        intents = []
+        if any(k in t for k in ["recon","observe","surveil","watch"]):
+            intents.append("Reconnaissance / Surveillance")
+        if any(k in t for k in ["attack","ambush","assault","blast","gunfire","strike"]):
+            intents.append("Direct Attack / Ambush")
+        if any(k in t for k in ["probe","test","drill","feint"]):
+            intents.append("Probe / Feint")
+        if any(k in t for k in ["disrupt","sabotage","deny service","block"]):
+            intents.append("Disruption / Sabotage")
+        if not intents:
+            intents.append("Unknown / Recon Required")
+        return intents
+
+    def extract_order_of_battle(self, text):
+        """Very simple OOB extractor: counts actors, platforms, activity tokens."""
+        tokens = [w.strip(".,:;()[]\"'").lower() for w in text.split()]
+        counts = Counter(tokens)
+        # heuristic features to present
+        features = {
+            "actors_mentioned": sorted(set([w for w in tokens if len(w)>3]) )[:12],
+            "mention_count": sum(v for k,v in counts.items() if len(k)>3),
+        }
+        return features
+
+    def cluster_ttp(self, reports, threshold=0.33):
+        """Group short TTP-like fragments into rough clusters (lexical heuristic)."""
+        clusters = []
+        for r in reports:
+            placed = False
+            for c in clusters:
+                if len(set(r.split()) & set(c["centroid"].split()))/max(1,len(set(r.split())|set(c["centroid"].split()))) > threshold:
+                    c["members"].append(r)
+                    # pick longest as centroid
+                    c["centroid"] = max(c["members"], key=lambda s: len(s.split()))
+                    placed = True
+                    break
+            if not placed:
+                clusters.append({"centroid": r, "members": [r]})
+        clusters.sort(key=lambda c: len(c["members"]), reverse=True)
+        return clusters
+
+    def enemy_analysis(self, text, extra_reports=None):
+        """
+        High-level enemy analysis (doctrinal). Returns JSON-friendly dict:
+        - inferred intent
+        - OOB features
+        - TTP clusters
+        - quick doctrinal assessment (risk_level)
+        """
+        res = {"timestamp": self._now(), "input": text}
+        res["intent"] = self.infer_intent(text)
+        res["oob"] = self.extract_order_of_battle(text)
+
+        ttp_fragments = []
+        if extra_reports:
+            for er in extra_reports:
+                if isinstance(er, str):
+                    ttp_fragments.extend([s.strip() for s in er.splitlines() if s.strip()])
+        # include short sentences from text as potential TTP fragments
+        ttp_fragments.extend([s.strip() for s in text.split(".") if s.strip()])
+        ttp_fragments = list(dict.fromkeys(ttp_fragments))[:60]
+        res["ttp_clusters"] = self.cluster_ttp(ttp_fragments)
+
+        # Simple, safe risk-level heuristic
+        t = text.lower()
+        if any(k in t for k in ["mass", "large", "suicide", "23 terrorists", "dozen", "multiple attackers", "armored"]):
+            risk = "High"
+        elif any(k in t for k in ["attack","ambush","firefight","assault","blast","gunfire"]):
+            risk = "Medium-High"
+        elif any(k in t for k in ["threat","probe","suspicious","riot","breach"]):
+            risk = "Medium"
+        else:
+            risk = "Low"
+        res["risk_level"] = risk
+
+        # Human-friendly doctrinal blurb (non-operational)
+        res["doctrinal_summary"] = (
+            f"At {res['timestamp']}: Inferred intent(s): {', '.join(res['intent'])}. "
+            f"Risk level (doctrinal heuristic): {res['risk_level']}. "
+            f"Top OOB markers: {', '.join(res['oob'].get('actors_mentioned',[])[:6]) or '[none]'}. "
+            "TTP clusters and reasoning provided for planning-level assessment only."
+        )
+
+        return res
+
+# wrapper instance and runner
+INTBrainV4 = INTBrainV4(base=globals().get("INTStrategicBrainV3", None))
+
+def run_int_brain_v4(text, extra_reports=None):
+    try:
+        out = INTBrainV4.enemy_analysis(text, extra_reports=extra_reports or [])
+        return out
+    except Exception as e:
+        return {"error": str(e), "timestamp": datetime.datetime.utcnow().isoformat()}
+
+# -------------------------
+# UI: Gradio Tab block (paste inside your build_interface demo)
+# -------------------------
+# Insert this block inside your with gr.Blocks(...) as demo: where other `with gr.Tab(...)` are defined.
+
+with gr.Tab("Enemy Analysis"):
+    gr.Markdown("""
+    ### Enemy Analysis (Doctrinal)
+    Paste reports, sightings, or chatter. The system produces a doctrinal, non-operational
+    Enemy Analysis: inferred intent, order-of-battle markers, TTP clusters, and a planning-level risk indicator.
+    """)
+
+    enemy_input = gr.Textbox(label="Paste Enemy-Related Text / Logs / Reports", lines=6,
+                             placeholder="e.g., '23 attackers attacked convoy at 0300, RPGs observed, vehicles spotted near waypoint X'")
+
+    extra_reports = gr.Textbox(label="Optional: Additional short reports (one per line)", lines=4,
+                               placeholder="Paste supporting notes or HUMINT fragments (optional).")
+
+    enemy_run = gr.Button("Run Enemy Analysis")
+    enemy_out_json = gr.JSON(label="Enemy Analysis (structured)")   # shows JSON
+    enemy_out_text = gr.Textbox(label="Enemy Analysis (human-friendly)", lines=12, interactive=False)
+
+    def _enemy_handler(text, extras):
+        txt = (text or "").strip()
+        if not txt:
+            return {"error":"No input provided."}, "No input provided."
+
+        try:
+            extra_list = [s for s in (extras or "").splitlines() if s.strip()]
+            analysis = run_int_brain_v4(txt, extra_reports=extra_list)
+
+            # Derive a polished, non-actionable brief if enrich available
+            brief = None
+            try:
+                prompt = (
+                    "You are a doctrinal security planner. Produce a concise, non-operational planning brief "
+                    "from this structured enemy analysis. Keep it high-level and non-actionable.\n\n"
+                    + json.dumps(analysis, indent=2, ensure_ascii=False)
+                )
+                brief = enrich_with_llm(prompt, context_title="Enemy Analysis Brief") if "enrich_with_llm" in globals() else None
+            except Exception:
+                brief = None
+
+            human = brief if brief else analysis.get("doctrinal_summary", "[No summary available]")
+
+            # Register with director shared state for later fusion
+            try:
+                update_shared_state("Enemy Analysis", {
+                    "input": txt[:800],
+                    "summary": analysis.get("doctrinal_summary", "")[:800],
+                    "risk_level": analysis.get("risk_level", "Unknown"),
+                    "timestamp": analysis.get("timestamp", "")
+                })
+            except Exception:
+                pass
+
+            return analysis, human
+
+        except Exception as e:
+            return {"error": str(e)}, f"[Error] {e}"
+
+    enemy_run.click(_enemy_handler, inputs=[enemy_input, extra_reports], outputs=[enemy_out_json, enemy_out_text])
+
+# -------------------------
+# Director hook snippet (small) - safe to paste before your Director button handler
+# -------------------------
+def director_collect_and_summarize(extra_context=""):
     """
-    For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
+    Lightweight wrapper: aggregates shared_reports_state, calls LLM if available,
+    and returns doctrinal consolidated brief (non-operational).
     """
-    client = InferenceClient(token=hf_token.token, model="openai/gpt-oss-20b")
-
-    messages = [{"role": "system", "content": system_message}]
-
-    messages.extend(history)
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        choices = message.choices
-        token = ""
-        if len(choices) and choices[0].delta.content:
-            token = choices[0].delta.content
-
-        response += token
-        yield response
-
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-chatbot = gr.ChatInterface(
-    respond,
-    type="messages",
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
-
-with gr.Blocks() as demo:
-    with gr.Sidebar():
-        gr.LoginButton()
-    chatbot.render()
-
-
-if __name__ == "__main__":
-    demo.launch()
+    try:
+        reports = shared_reports_state.copy() if isinstance(shared_reports_state, dict) else {}
+        if not reports:
+            return "⚠️ No departmental reports available yet."
+
+        compiled = "DIRECTOR CONSOLIDATED (DOCTRINAL)\nGenerated: " + datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") + "\n\n"
+        for dept, r in reports.items():
+            compiled += f"--- {dept} ---\n"
+            if isinstance(r, dict):
+                compiled += f"Risk: {r.get('risk_level','Unknown')}\nSummary: {r.get('summary','')[:600]}\n\n"
+            else:
+                compiled += str(r)[:800] + "\n\n"
+
+        # Optional LLM polish (safe, non-operational)
+        try:
+            if "enrich_with_llm" in globals():
+                prompt = "Polish the following doctrinal consolidated report. Keep it non-operational.\n\n" + compiled + "\n\nContext: " + (extra_context or "")
+                polished = enrich_with_llm(prompt, context_title="Director Consolidated Brief")
+                return polished or compiled
+        except Exception:
+            pass
+
+        return compiled
+
+    except Exception as e:
+        return f"[Director aggregation error] {e}"
+
+# ================================================================
+# === END OF INT v4 Enemy Analysis block
+# ================================================================