modules/cypher_tools.py

"""CYPHER V12 M11 — TOOLS_REGISTRY (central tool registry, NEXUS pattern).

Aggregates all CYPHER tools from M2-M10 modules into a single dict
TOOLS_REGISTRY. Lazy imports (try/except) so missing optional deps don't
break the bridge.

Public API:
  - TOOLS_REGISTRY: dict[str, callable]
  - call_tool(name, **kwargs): execute a tool by name
  - list_available_tools(): introspection
  - get_tool_signature(name): runtime signature peek
"""
from __future__ import annotations

import inspect
import logging
import os
import sys
from pathlib import Path
from typing import Any, Callable

logger = logging.getLogger(__name__)


# ──────────────────────────── HELPERS ───────────────────────────────


def _env(key: str, default: str | None = None) -> str | None:
    return os.environ.get(key, default)


def _key_available(env_key: str) -> bool:
    return bool(os.environ.get(env_key))


# ──────────────────────────── LAZY MODULES ──────────────────────────


_MODULE_DIR = Path(__file__).parent
if str(_MODULE_DIR) not in sys.path:
    sys.path.insert(0, str(_MODULE_DIR))


def _lazy_import(module_name: str) -> Any:
    try:
        return __import__(module_name)
    except ImportError as e:
        logger.warning(f"_lazy_import {module_name} failed: {e}")
        return None


# Cache singletons to avoid re-instantiation per call
_singletons: dict[str, Any] = {}


def _get_singleton(key: str, factory: Callable[[], Any]) -> Any:
    if key not in _singletons:
        try:
            _singletons[key] = factory()
        except Exception as e:
            logger.error(f"singleton {key} init failed: {type(e).__name__}: {e}")
            _singletons[key] = None
    return _singletons[key]


# ──────────────────────────── TOOL ADAPTERS ─────────────────────────


# M2 familybus_client
def _t_family_pull_recent(n: int = 20, since_ts: int | None = None) -> list[dict]:
    fb = _lazy_import("familybus_client")
    if fb is None:
        return [{"error": "familybus_client unavailable"}]
    client = _get_singleton("familybus", lambda: fb.FamilyBusClient())
    if client is None:
        return [{"error": "FamilyBusClient init failed"}]
    return client.pull_recent(n=n, since_ts=since_ts)


def _t_family_broadcast(lesson: str, category: str = "CYBERSEC",
                        trade_summary: dict | None = None) -> bool:
    fb = _lazy_import("familybus_client")
    if fb is None:
        return False
    client = _get_singleton("familybus", lambda: fb.FamilyBusClient())
    if client is None:
        return False
    return client.broadcast(lesson=lesson, asi="CYPHER",
                            category=category, trade_summary=trade_summary)


def _t_family_stats() -> dict:
    fb = _lazy_import("familybus_client")
    if fb is None:
        return {"error": "familybus_client unavailable"}
    client = _get_singleton("familybus", lambda: fb.FamilyBusClient())
    if client is None:
        return {"error": "init failed"}
    return client.get_stats()


# M3 cve_refresh_pipeline
def _t_cisa_kev_refresh(force: bool = False) -> dict:
    cve_mod = _lazy_import("cve_refresh_pipeline")
    if cve_mod is None:
        return {"error": "cve_refresh_pipeline unavailable"}
    kev = _get_singleton("kev", lambda: cve_mod.CISAKevRefresher())
    if kev is None:
        return {"error": "init failed"}
    return {"count": kev.refresh(force=force)}


def _t_cisa_kev_lookup(cve_id: str) -> dict | None:
    cve_mod = _lazy_import("cve_refresh_pipeline")
    if cve_mod is None:
        return None
    kev = _get_singleton("kev", lambda: cve_mod.CISAKevRefresher())
    if kev is None:
        return None
    return kev.lookup(cve_id)


def _t_nvd_cve_lookup(cve_id: str) -> dict | None:
    cve_mod = _lazy_import("cve_refresh_pipeline")
    if cve_mod is None:
        return None
    nvd = _get_singleton("nvd", lambda: cve_mod.NVDLookup())
    if nvd is None:
        return None
    return nvd.lookup_cve(cve_id)


def _t_get_cve_context(cve_id: str) -> str:
    cve_mod = _lazy_import("cve_refresh_pipeline")
    if cve_mod is None:
        return f"[CONTEXT: CVE={cve_id} (cve module unavailable)]"
    kev = _get_singleton("kev", lambda: cve_mod.CISAKevRefresher())
    nvd = _get_singleton("nvd", lambda: cve_mod.NVDLookup())
    return cve_mod.get_cve_context(cve_id, kev=kev, nvd=nvd)


# M4 cypher_memory
def _t_store_memory(content: str, category: str,
                    metadata: dict | None = None) -> str | None:
    mem_mod = _lazy_import("cypher_memory")
    if mem_mod is None:
        return None
    mem = _get_singleton("memory", lambda: mem_mod.CypherMemory())
    if mem is None:
        return None
    try:
        return mem.store_memory(content, category, metadata)
    except (ValueError, Exception) as e:
        logger.error(f"store_memory: {type(e).__name__}: {e}")
        return None


def _t_recall_memory(query: str, k: int = 3,
                     category: str | None = None) -> list[dict]:
    mem_mod = _lazy_import("cypher_memory")
    if mem_mod is None:
        return []
    mem = _get_singleton("memory", lambda: mem_mod.CypherMemory())
    if mem is None:
        return []
    return mem.recall_memory(query, k=k, category=category)


def _t_list_memories(category: str | None = None, limit: int = 50) -> list[dict]:
    mem_mod = _lazy_import("cypher_memory")
    if mem_mod is None:
        return []
    mem = _get_singleton("memory", lambda: mem_mod.CypherMemory())
    if mem is None:
        return []
    return mem.list_memories(category=category, limit=limit)


def _t_delete_memory(memory_id: str) -> bool:
    mem_mod = _lazy_import("cypher_memory")
    if mem_mod is None:
        return False
    mem = _get_singleton("memory", lambda: mem_mod.CypherMemory())
    if mem is None:
        return False
    return mem.delete_memory(memory_id)


def _t_memory_stats() -> dict:
    mem_mod = _lazy_import("cypher_memory")
    if mem_mod is None:
        return {"error": "memory module unavailable"}
    mem = _get_singleton("memory", lambda: mem_mod.CypherMemory())
    if mem is None:
        return {"error": "init failed"}
    return mem.memory_stats()


def _t_clear_category(category: str) -> int:
    mem_mod = _lazy_import("cypher_memory")
    if mem_mod is None:
        return 0
    mem = _get_singleton("memory", lambda: mem_mod.CypherMemory())
    if mem is None:
        return 0
    return mem.clear_category(category)


# M5 cypher_rule_validator
def _t_yara_validate(rule_str: str) -> dict:
    rv = _lazy_import("cypher_rule_validator")
    if rv is None:
        return {"valid": None, "error": "rule_validator unavailable", "rule_name": None}
    return rv.YaraValidator().validate(rule_str)


def _t_sigma_validate(yaml_str: str) -> dict:
    rv = _lazy_import("cypher_rule_validator")
    if rv is None:
        return {"valid": None, "error": "rule_validator unavailable", "title": None}
    return rv.SigmaValidator().validate(yaml_str)


def _t_suricata_validate(rule_str: str) -> dict:
    rv = _lazy_import("cypher_rule_validator")
    if rv is None:
        return {"valid": None, "error": "rule_validator unavailable", "sid": None}
    return rv.SuricataValidator().validate(rule_str)


def _t_extract_rules_from_text(text: str) -> dict:
    rv = _lazy_import("cypher_rule_validator")
    if rv is None:
        return {"yara": [], "sigma": [], "suricata": []}
    return rv.extract_rules_from_text(text)


# M6 conversation_context_bank
def _t_conv_add_turn(user_msg: str, cypher_response: str,
                     category: str | None = None) -> bool:
    cb = _lazy_import("conversation_context_bank")
    if cb is None:
        return False
    bank = _get_singleton("conv", lambda: cb.ConversationContextBank())
    if bank is None:
        return False
    bank.add_turn(user_msg, cypher_response, category=category)
    return True


def _t_conv_get_context(format: str = "compact") -> str:
    cb = _lazy_import("conversation_context_bank")
    if cb is None:
        return ""
    bank = _get_singleton("conv", lambda: cb.ConversationContextBank())
    if bank is None:
        return ""
    return bank.get_context(format=format)


def _t_conv_clear() -> bool:
    cb = _lazy_import("conversation_context_bank")
    if cb is None:
        return False
    bank = _get_singleton("conv", lambda: cb.ConversationContextBank())
    if bank is None:
        return False
    bank.clear()
    return True


# M10 multimodal_stub
def _t_describe_image(image_src: str, prompt: str = "Describe briefly.") -> str:
    mm = _lazy_import("multimodal_stub")
    if mm is None:
        return "[multimodal_stub unavailable]"
    backend = "anthropic" if _key_available("ANTHROPIC_API_KEY") else "mock"
    stub = _get_singleton(f"mm_{backend}", lambda: mm.MultimodalStub(backend=backend))
    if stub is None:
        return "[init failed]"
    return stub.describe_image(image_src, prompt=prompt)


def _t_is_image_request(prompt: str) -> bool:
    mm = _lazy_import("multimodal_stub")
    if mm is None:
        return False
    return mm.is_image_request(prompt)


# ──────────────────────────── REGISTRY ──────────────────────────────


TOOLS_REGISTRY: dict[str, dict] = {
    # M2 FamilyBus
    "family_pull_recent": {
        "fn": _t_family_pull_recent,
        "module": "M2",
        "desc": "Pull recent broadcasts from aether-family-bus HF dataset",
    },
    "family_broadcast": {
        "fn": _t_family_broadcast,
        "module": "M2",
        "desc": "Broadcast a CYPHER lesson to the family (async)",
    },
    "family_stats": {
        "fn": _t_family_stats,
        "module": "M2",
        "desc": "Get FamilyBus cache stats (producers, latest)",
    },
    # M3 CVE
    "cisa_kev_refresh": {
        "fn": _t_cisa_kev_refresh,
        "module": "M3",
        "desc": "Refresh CISA KEV catalog from cisa.gov",
    },
    "cisa_kev_lookup": {
        "fn": _t_cisa_kev_lookup,
        "module": "M3",
        "desc": "Lookup a CVE in CISA KEV",
    },
    "nvd_cve_lookup": {
        "fn": _t_nvd_cve_lookup,
        "module": "M3",
        "desc": "Lookup a CVE in NVD API v2 (rate limited)",
    },
    "get_cve_context": {
        "fn": _t_get_cve_context,
        "module": "M3",
        "desc": "Build [CONTEXT: ...] string for a CVE (KEV + NVD merged)",
    },
    # M4 Memory
    "store_memory": {
        "fn": _t_store_memory,
        "module": "M4",
        "desc": "Persist a memory in chromadb (categories: cve_analysis, trade_postmortem, ioc_confirmed, cybersec_insight, conversation, ecosystem_fact)",
    },
    "recall_memory": {
        "fn": _t_recall_memory,
        "module": "M4",
        "desc": "Recall top-k memories by similarity (filter by category)",
    },
    "list_memories": {
        "fn": _t_list_memories,
        "module": "M4",
        "desc": "List stored memories (filter by category)",
    },
    "delete_memory": {
        "fn": _t_delete_memory,
        "module": "M4",
        "desc": "Delete a memory by ID",
    },
    "memory_stats": {
        "fn": _t_memory_stats,
        "module": "M4",
        "desc": "Memory storage stats per category",
    },
    "clear_category": {
        "fn": _t_clear_category,
        "module": "M4",
        "desc": "Clear all memories of a category",
    },
    # M5 Rule validator
    "yara_validate": {
        "fn": _t_yara_validate,
        "module": "M5",
        "desc": "Validate YARA rule via yara-python compile",
    },
    "sigma_validate": {
        "fn": _t_sigma_validate,
        "module": "M5",
        "desc": "Validate Sigma rule YAML structure",
    },
    "suricata_validate": {
        "fn": _t_suricata_validate,
        "module": "M5",
        "desc": "Validate Suricata/Snort rule syntax",
    },
    "extract_rules_from_text": {
        "fn": _t_extract_rules_from_text,
        "module": "M5",
        "desc": "Detect YARA/Sigma/Suricata blocks in arbitrary text",
    },
    # M6 Conversation
    "conv_add_turn": {
        "fn": _t_conv_add_turn,
        "module": "M6",
        "desc": "Append a turn to the 3-window conversation context bank",
    },
    "conv_get_context": {
        "fn": _t_conv_get_context,
        "module": "M6",
        "desc": "Get formatted conversation context for encoder injection",
    },
    "conv_clear": {
        "fn": _t_conv_clear,
        "module": "M6",
        "desc": "Clear conversation context window",
    },
    # M10 Multimodal
    "describe_image": {
        "fn": _t_describe_image,
        "module": "M10",
        "desc": "Describe an image via Anthropic/OpenAI vision API",
    },
    "is_image_request": {
        "fn": _t_is_image_request,
        "module": "M10",
        "desc": "Heuristic: does prompt require image analysis?",
    },
}


def call_tool(name: str, **kwargs) -> Any:
    """Invoke a registered tool by name."""
    if name not in TOOLS_REGISTRY:
        raise KeyError(f"Tool not registered: {name!r}. Available: {sorted(TOOLS_REGISTRY)}")
    fn = TOOLS_REGISTRY[name]["fn"]
    try:
        return fn(**kwargs)
    except TypeError as e:
        # Caller passed wrong kwargs
        sig = inspect.signature(fn)
        raise TypeError(
            f"call_tool({name!r}, **kwargs) signature mismatch. "
            f"Expected: {sig}. Got kwargs={list(kwargs)}. Error: {e}"
        ) from e


def list_available_tools() -> list[dict]:
    """Return list of tool descriptors."""
    out: list[dict] = []
    for name, info in TOOLS_REGISTRY.items():
        try:
            sig = str(inspect.signature(info["fn"]))
        except (TypeError, ValueError):
            sig = "(?)"
        out.append({
            "name": name,
            "module": info["module"],
            "desc": info["desc"],
            "signature": sig,
        })
    return sorted(out, key=lambda x: (x["module"], x["name"]))


def get_tool_signature(name: str) -> str:
    if name not in TOOLS_REGISTRY:
        return ""
    try:
        return str(inspect.signature(TOOLS_REGISTRY[name]["fn"]))
    except (TypeError, ValueError):
        return "(?)"


__all__ = [
    "TOOLS_REGISTRY",
    "call_tool",
    "list_available_tools",
    "get_tool_signature",
]


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    print("=== M11 cypher_tools SMOKE TEST ===")

    tools = list_available_tools()
    print(f"Total tools registered: {len(tools)}")
    by_mod: dict[str, list[str]] = {}
    for t in tools:
        by_mod.setdefault(t["module"], []).append(t["name"])
    for mod in sorted(by_mod):
        print(f"  {mod} ({len(by_mod[mod])} tools): {by_mod[mod]}")

    # Test a simple tool: extract_rules_from_text (no IO needed)
    rules = call_tool("extract_rules_from_text",
                       text="rule Test { strings: $a = \"x\" condition: $a }")
    print(f"\nextract_rules_from_text: yara={len(rules['yara'])} sigma={len(rules['sigma'])} suricata={len(rules['suricata'])}")

    # Test family_stats (no network)
    stats = call_tool("family_stats")
    print(f"family_stats: {stats}")

    # Test invalid kwargs error
    try:
        call_tool("yara_validate", wrong_kwarg="oops")
    except TypeError as e:
        print(f"Type error caught as expected: {str(e)[:200]}")

    # Test unknown tool
    try:
        call_tool("nonexistent_tool")
    except KeyError as e:
        print(f"KeyError caught: {str(e)[:120]}")

    # Test get_tool_signature
    sig = get_tool_signature("yara_validate")
    print(f"yara_validate signature: {sig}")

    print("=== SMOKE PASS ===")