server.py

"""
Vitalis LOREIN MCP Server — Cognitive toolkit for any model.
The server has no model, no weights, no LLM calls.
The model (client) calls these tools to improve its own output.
"""

import json
import traceback

from mcp.server.fastmcp import FastMCP
from pipeline.lorein import LOREIN

mcp = FastMCP("Vitalis LOREIN Server")
lorein = LOREIN()


def safe_call(fn, *args, **kwargs):
    """Wrap MCP tools with structured error handling."""
    try:
        return fn(*args, **kwargs)
    except Exception as e:
        return json.dumps({
            "error": str(e),
            "traceback": traceback.format_exc(),
        }, indent=2)


# ── Chat / Cognitive Prep ──

@mcp.tool(
    name="lorein_chat",
    description="Prepare cognitive context for a conversation. Governor checks input, MIRROR plans (Goals/Reasoning/Memory threads), logs to Journal. Returns structured context so the calling model can generate an informed response. Never generates a response itself.",
)
def lorein_chat(message: str) -> str:
    return safe_call(lambda: json.dumps(lorein.prepare_chat(message), indent=2))


@mcp.tool(
    name="lorein_truth_check",
    description="Verify a model response against the Truth Engine. Pass your draft response here to detect hedging, contradictions, and uncertainty before sending to the user.",
)
def lorein_truth_check(response: str, context: str = "") -> str:
    return safe_call(lambda: json.dumps(lorein.truth_check(response, context), indent=2))


# ── MIRROR Tools ──

@mcp.tool(
    name="lorein_mirror_think",
    description="Run MIRROR internal monologue (Goals/Reasoning/Memory threads). Call this before generating to get a structured plan and retrieved context.",
)
def lorein_mirror_think(input_text: str) -> str:
    return safe_call(lambda: json.dumps({
        "narrative": lorein.mirror.think(input_text)[0],
        "threads": lorein.mirror.think(input_text)[1],
    }, indent=2))


# ── Journal Tools ──

@mcp.tool(
    name="lorein_journal_query",
    description="Query the Journal — event-sourced, hash-chained immutable ledger of all tool calls and decisions.",
)
def lorein_journal_query(action: str | None = None, actor: str | None = None, limit: int = 50) -> str:
    return safe_call(lambda: json.dumps({
        "entries": lorein.journal.query(action, actor, limit),
        "total": lorein.journal.size(),
        "chain_integrity": lorein.journal.verify_chain(),
    }, indent=2))


@mcp.tool(
    name="lorein_journal_verify",
    description="Verify the Journal's cryptographic chain integrity. Returns true if no tampering detected.",
)
def lorein_journal_verify() -> str:
    return safe_call(lambda: json.dumps({
        "chain_integrity": lorein.journal.verify_chain(),
        "entry_count": lorein.journal.size(),
        "last_hash": lorein.journal.last_hash(),
    }, indent=2))


# ── Memory Tools ──

@mcp.tool(
    name="lorein_memory_query",
    description="Query hierarchical memory (episodic, semantic, procedural). Retrieve context across sessions.",
)
def lorein_memory_query(query: str, memory_type: str = "semantic") -> str:
    return safe_call(lambda: _query_memory(query, memory_type))


def _query_memory(query: str, memory_type: str):
    if memory_type == "semantic":
        results = lorein.semantic.query(query, top_k=10)
        return json.dumps({
            "type": "semantic",
            "results": [t.to_dict() for t in results],
            "total_facts": lorein.semantic.count(),
        }, indent=2)
    elif memory_type == "episodic":
        results = lorein.episodic.query(limit=20)
        return json.dumps({
            "type": "episodic",
            "results": results,
            "total_events": lorein.episodic.count(),
        }, indent=2)
    elif memory_type == "procedural":
        results = lorein.procedural.retrieve(query)
        return json.dumps({
            "type": "procedural",
            "results": [{"name": n, "skill": s} for n, s in results],
            "total_skills": lorein.procedural.count(),
        }, indent=2)
    return json.dumps({"error": f"Unknown memory type: {memory_type}"})


@mcp.tool(
    name="lorein_memory_consolidate",
    description="Run consolidation protocol — compress episodic events into semantic beliefs. The only learning mechanism in the system (no weight updates).",
)
def lorein_memory_consolidate(force: bool = True) -> str:
    return safe_call(lambda: json.dumps(lorein.consolidation.consolidate(force=force), indent=2))


# ── Room Tools ──

@mcp.tool(
    name="lorein_room_create",
    description="Create a sandboxed execution environment (the Room). Returns a sandbox_id for subsequent calls.",
)
def lorein_room_create(sandbox_id: str | None = None) -> str:
    return safe_call(lambda: json.dumps({"sandbox_id": lorein.room.create(sandbox_id)}))


@mcp.tool(
    name="lorein_room_execute",
    description="Execute code in the sandboxed Room. Results include stdout, stderr, return code. Use this to verify generated code.",
)
def lorein_room_execute(code: str, language: str = "python", sandbox_id: str | None = None) -> str:
    return safe_call(lambda: json.dumps(lorein.run_code(code, language, sandbox_id), indent=2))


@mcp.tool(
    name="lorein_room_write",
    description="Write a file in the Room sandbox workspace.",
)
def lorein_room_write(sandbox_id: str, path: str, content: str) -> str:
    return safe_call(lambda: json.dumps(lorein.room.write_file(sandbox_id, path, content)))


@mcp.tool(
    name="lorein_room_read",
    description="Read a file from the Room sandbox workspace.",
)
def lorein_room_read(sandbox_id: str, path: str) -> str:
    return safe_call(lambda: json.dumps(lorein.room.read_file(sandbox_id, path)))


@mcp.tool(
    name="lorein_room_destroy",
    description="Destroy a sandboxed Room environment.",
)
def lorein_room_destroy(sandbox_id: str) -> str:
    return safe_call(lambda: json.dumps({"destroyed": lorein.room.destroy(sandbox_id)}))


@mcp.tool(
    name="lorein_room_list",
    description="List all active Room sandboxes.",
)
def lorein_room_list() -> str:
    return safe_call(lambda: json.dumps({"sandboxes": lorein.room.list_sandboxes()}))


# ── Debug Loop Tool ──

@mcp.tool(
    name="lorein_debug_loop",
    description="Run code with error capture and regeneration support. If code fails, returns the error trace so the model can fix and retry with the same session_id. Keeps iteration state across attempts.",
)
def lorein_debug_loop(query: str, code: str, language: str = "python",
                      max_iterations: int = 3, session_id: str | None = None,
                      sandbox_id: str | None = None) -> str:
    return safe_call(lambda: json.dumps(
        lorein.debug_loop(query, code, language, max_iterations, sandbox_id), indent=2
    ))


# ── Active Inference Tools ──

@mcp.tool(
    name="lorein_inference_state",
    description="Get the current Active Inference state (VFE, EFE, confidence, adaptive temperature, exploration urgency).",
)
def lorein_inference_state() -> str:
    return safe_call(lambda: json.dumps(lorein.inference.state(), indent=2))


@mcp.tool(
    name="lorein_empowerment",
    description="Get the current empowerment metric — measures action-outcome channel capacity.",
)
def lorein_empowerment() -> str:
    return safe_call(lambda: json.dumps({
        "empowerment": round(lorein.empowerment.compute(), 3),
        "should_explore": lorein.inference.should_explore(),
        "exploration_urgency": lorein.inference.exploration_urgency(),
    }, indent=2))


# ── Governor Tools ──

@mcp.tool(
    name="lorein_governor_evaluate",
    description="Evaluate an action against the ethical constraint manifold. Returns PASS / WARN / BLOCK. Core rule: protects the mutual advancement between user and model.",
)
def lorein_governor_evaluate(action: str, context: str = "") -> str:
    return safe_call(lambda: json.dumps(lorein.governor.evaluate(action, context), indent=2))


@mcp.tool(
    name="lorein_governor_add_constraint",
    description="Add a dynamic constraint to the ethical manifold. Patterns are comma-separated. Cannot override core constraints.",
)
def lorein_governor_add_constraint(name: str, patterns: str, cost: float) -> str:
    return safe_call(lambda: _add_constraint(name, patterns, cost))


def _add_constraint(name: str, patterns: str, cost: float):
    pat_list = [p.strip() for p in patterns.split(",")]
    lorein.governor.add_constraint(name, pat_list, cost)
    return json.dumps({"added": name, "patterns": pat_list, "cost": cost})


# ── Shadow Mode Tools ──

@mcp.tool(
    name="lorein_shadow_create",
    description="Create a shadow agent for parallel testing (Level 3+5 Divergence Engine).",
)
def lorein_shadow_create(name: str, config: str = "{}") -> str:
    return safe_call(lambda: _create_shadow(name, config))


def _create_shadow(name: str, config: str):
    parsed = json.loads(config)
    sid = lorein.shadow.create_shadow(name, parsed)
    return json.dumps({"shadow_id": sid})


@mcp.tool(
    name="lorein_shadow_compare",
    description="Compare live vs shadow agent decisions. Returns divergence score and details.",
)
def lorein_shadow_compare(live_decision: str, shadow_id: str) -> str:
    return safe_call(lambda: _compare_shadow(live_decision, shadow_id))


def _compare_shadow(live_decision: str, shadow_id: str):
    live = json.loads(live_decision)
    shadow_agent = lorein.shadow.get_shadow(shadow_id)
    if not shadow_agent or not shadow_agent["decisions"]:
        return json.dumps({"error": "No shadow decisions available"})
    shadow_dec = shadow_agent["decisions"][-1]["decision"]
    result = lorein.shadow.compare(live, shadow_dec)
    return json.dumps(result, indent=2)


# ── Perception Tools ──

@mcp.tool(
    name="lorein_desktop_context",
    description="Get desktop context (platform, processes, recent activity).",
)
def lorein_desktop_context() -> str:
    return safe_call(lambda: _get_desktop())


def _get_desktop():
    from perception import get_desktop_context, list_recent_activity
    ctx = get_desktop_context()
    ctx["recent_activity"] = list_recent_activity(10)
    return json.dumps(ctx, indent=2)


@mcp.tool(
    name="lorein_search_memory",
    description="Search local memory files for content matching a query.",
)
def lorein_search_memory(query: str, path: str | None = None) -> str:
    return safe_call(lambda: _search_mem(query, path))


def _search_mem(query: str, path: str | None):
    from perception import search_content
    results = search_content(query, path)
    return json.dumps({"query": query, "results": results, "count": len(results)}, indent=2)


# ── Router Tools ──

from router import route as _route_task, classify_task, get_routing_history
from router import TASK_ROUTES, TASK_CATEGORIES

@mcp.tool(
    name="vitalis_route",
    description="Route a task to the best model. Classifies the task (code/reasoning/research/creative/general) and returns the recommended model. Call this before lorein_chat to pick the right model for the job.",
)
def vitalis_route(message: str, preferred_model: str = "") -> str:
    return safe_call(lambda: json.dumps({
        "routing": _route_task(message, preferred_model or None),
        "available_tasks": list(TASK_ROUTES.keys()),
    }, indent=2))


@mcp.tool(
    name="vitalis_classify",
    description="Classify a message into a task category: code, reasoning, research, creative, or general. Does NOT select a model — use vitalis_route for that.",
)
def vitalis_classify(message: str) -> str:
    return safe_call(lambda: json.dumps({
        "task": classify_task(message),
        "categories": {k: v for k, v in TASK_CATEGORIES.items()},
    }, indent=2))


# ── Identity Tools ──

@mcp.tool(
    name="lorein_identity",
    description="Get the LOREIN agent's identity, chain integrity, and state summary.",
)
def lorein_identity() -> str:
    return safe_call(lambda: json.dumps(lorein.get_identity(), indent=2))


@mcp.tool(
    name="lorein_integrity_check",
    description="Full system integrity check. Verifies Journal chain, identity binding, and memory consistency.",
)
def lorein_integrity_check() -> str:
    return safe_call(lambda: json.dumps(lorein.verify_integrity(), indent=2))


# ── Resources ──

@mcp.resource(
    uri="lorein://identity",
    name="LOREIN Identity",
    description="Full agent identity and state summary.",
)
def resource_identity() -> str:
    return json.dumps(lorein.get_identity(), indent=2)


@mcp.resource(
    uri="lorein://journal",
    name="LOREIN Journal",
    description="Recent journal entries with chain integrity status.",
)
def resource_journal() -> str:
    entries = lorein.journal.query(limit=20)
    return json.dumps({
        "entries": entries,
        "total": lorein.journal.size(),
        "chain_integrity": lorein.journal.verify_chain(),
    }, indent=2)


@mcp.resource(
    uri="lorein://inference",
    name="Active Inference State",
    description="Current VFE, EFE, confidence, exploration metrics.",
)
def resource_inference() -> str:
    return json.dumps(lorein.inference.state(), indent=2)


def main():
    mcp.run()


if __name__ == "__main__":
    main()