# ================================================================ # === 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=""): """ Lightweight wrapper: aggregates shared_reports_state, calls LLM if available, and returns doctrinal consolidated brief (non-operational). """ 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 # ================================================================