server.py

"""
==========================================================================
🌉 Thought Engine Bridge — MCP Tool Server (SSE Transport)
==========================================================================
Contract: C-THOUGHT-BRIDGE-001 (Phase 2)
Node:     https://kode-animator-thought-engine-node.hf.space
Stack:    FastMCP (SSE) + httpx → Thought Engine Node REST API

Exposes the Thought Engine Node's sovereign REST endpoints as AI-usable
MCP tools. Any MCP-compatible client (ChatGPT, Claude, Cline, etc.)
can create sessions, add thoughts, fork branches, submit proposals,
and visualize reasoning trees through this bridge.
==========================================================================
"""

import asyncio
import os
import json
import httpx
from contextlib import asynccontextmanager
from typing import Optional
from mcp.server.mcpserver import MCPServer
from starlette.applications import Starlette
from starlette.requests import Request as StarletteRequest
from starlette.responses import JSONResponse
from starlette.routing import Route

# ── Configuration ──────────────────────────────────────────────
NODE_URL = os.getenv(
    "THOUGHT_ENGINE_NODE_URL",
    "https://kode-animator-thought-engine-node.hf.space"
)

mcp = MCPServer("thought-engine-bridge")
_node_prewarm_started = False
_node_prewarm_task: asyncio.Task | None = None


async def _prewarm_node() -> None:
    """Best-effort bounded wake/readiness probe for the downstream Node."""
    for attempt in range(3):
        try:
            async with httpx.AsyncClient(timeout=5.0) as client:
                response = await client.get(f"{NODE_URL}/health")
            if response.status_code == 200:
                payload = response.json()
                if payload.get("status") == "ok":
                    return
        except Exception:
            pass

        if attempt < 2:
            await asyncio.sleep(0.25)


@asynccontextmanager
async def _bridge_lifespan(app: Starlette, modern_lifespan):
    global _node_prewarm_started, _node_prewarm_task

    async with modern_lifespan(app):
        if not _node_prewarm_started:
            _node_prewarm_started = True
            _node_prewarm_task = asyncio.create_task(_prewarm_node())
        try:
            yield
        finally:
            if _node_prewarm_task is not None and not _node_prewarm_task.done():
                _node_prewarm_task.cancel()
                try:
                    await _node_prewarm_task
                except asyncio.CancelledError:
                    pass


# ── HTTP Helper ────────────────────────────────────────────────
async def _call_node(method: str, path: str, body: dict = None) -> dict:
    """Call the Thought Engine Node REST API."""
    url = f"{NODE_URL}{path}"
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if method == "GET":
                r = await client.get(url)
            else:
                r = await client.post(url, json=body or {})
            r.raise_for_status()
            return r.json()
        except httpx.HTTPStatusError as e:
            return {"error": f"Node returned {e.response.status_code}", "detail": e.response.text}
        except Exception as e:
            return {"error": str(e)}


# ══════════════════════════════════════════════════════════════
#  ONBOARDING
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_onboard(client_name: str = "Agent") -> str:
    """🧠 REQUIRED FIRST STEP: Learn the Thought Engine capabilities and protocols.

    Call this tool first to understand how to use the Thought Engine effectively.
    """
    return f"""# 🧠 Welcome to the Thought Engine, {client_name}!

## What This Is
A **persistent, governed reasoning substrate** backed by Cloudflare D1.
Every thought you create survives restarts. Every fork is tracked. Every proposal is witnessed.

## Core Concepts
- **Sessions**: Reasoning containers. Each session holds a tree of thoughts.
- **Thoughts**: Typed nodes — hypothesis, observation, validation, counter_argument, synthesis, decision, question, proposal.
- **Forking**: Explore alternative reasoning paths without destroying the main thread.
- **Proposals (PRs)**: Submit a "Pull Request" for a thought. Can be accepted, rejected, or superseded.
- **Witness Trail**: Every action is provenance-logged — who did what, when, and why.

## Workflow
1. `thought_start_session` — Create a new reasoning session
2. `thought_add_step` — Add sequential thoughts to the active chain
3. `thought_fork` — Branch off to explore an alternative
4. `thought_propose` — Submit a formal thought proposal
5. `thought_review_proposal` — Accept or reject a proposal
6. `thought_display_tree` — Visualize the full reasoning tree
7. `thought_search` — Search thoughts by content pattern
8. `thought_witness` — View the audit/provenance trail

## Available Thought Types
`hypothesis`, `observation`, `validation`, `counter_argument`, `synthesis`, `decision`, `question`, `proposal`

Begin by creating a session with `thought_start_session`! 🧠
"""


# ══════════════════════════════════════════════════════════════
#  SESSION TOOLS
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_start_session(
    initial_thought: str,
    title: str = "",
    thought_class: str = "hypothesis",
    agent_id: str = "Agent"
) -> str:
    """🧠 Start a new reasoning session with an initial thought.

    Args:
        initial_thought: The opening thought or question to reason about.
        title: Optional title for the session (defaults to first 80 chars of thought).
        thought_class: Type of thought — hypothesis, observation, validation, counter_argument, synthesis, decision, question.
        agent_id: Your identity for provenance tracking.
    """
    result = await _call_node("POST", "/session", {
        "initial_thought": initial_thought,
        "title": title or None,
        "thought_class": thought_class,
        "agent_id": agent_id,
    })
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_get_session(session_id: str) -> str:
    """📋 Get metadata for a specific session.

    Args:
        session_id: The session ID to retrieve.
    """
    result = await _call_node("GET", f"/session/{session_id}")
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_list_sessions(status: str = "", limit: int = 20) -> str:
    """📚 List all reasoning sessions, optionally filtered by status.

    Args:
        status: Filter by status — active, archived, merged. Leave empty for all.
        limit: Maximum number of sessions to return (1-100).
    """
    params = f"?limit={limit}"
    if status:
        params += f"&status={status}"
    result = await _call_node("GET", f"/sessions{params}")
    return json.dumps(result, indent=2)


# ══════════════════════════════════════════════════════════════
#  THOUGHT TOOLS
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_add_step(
    session_id: str,
    content: str,
    thought_class: str = "observation",
    agent_id: str = "Agent",
    parent_thought_id: str = "",
    confidence: float = None,
) -> str:
    """💭 Add a reasoning step to the session's active chain.

    Args:
        session_id: The session to add the thought to.
        content: The thought content.
        thought_class: Type — hypothesis, observation, validation, counter_argument, synthesis, decision, question.
        agent_id: Your identity for provenance tracking.
        parent_thought_id: Optional specific parent (defaults to session's active thought).
        confidence: Optional confidence score (0.0-1.0).
    """
    body = {
        "content": content,
        "thought_class": thought_class,
        "agent_id": agent_id,
    }
    if parent_thought_id:
        body["parent_thought_id"] = parent_thought_id
    if confidence is not None:
        body["confidence"] = confidence

    result = await _call_node("POST", f"/session/{session_id}/thought", body)
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_list_thoughts(session_id: str) -> str:
    """📃 List all thought units in a session, ordered chronologically.

    Args:
        session_id: The session to list thoughts from.
    """
    result = await _call_node("GET", f"/session/{session_id}/thoughts")
    return json.dumps(result, indent=2)


# ══════════════════════════════════════════════════════════════
#  FORKING TOOLS
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_fork(
    session_id: str,
    source_thought_id: str,
    branch_label: str,
    agent_id: str = "Agent",
) -> str:
    """🌿 Fork a thought chain to explore an alternative reasoning path.

    Creates a branch from the specified thought without destroying the original thread.
    The session's active pointer moves to the new fork.

    Args:
        session_id: The session containing the thought to fork.
        source_thought_id: The thought ID to branch from.
        branch_label: A descriptive label for the branch (e.g., "risk-analysis", "alternative-approach").
        agent_id: Your identity for provenance tracking.
    """
    result = await _call_node("POST", f"/session/{session_id}/fork", {
        "source_thought_id": source_thought_id,
        "branch_label": branch_label,
        "agent_id": agent_id,
    })
    return json.dumps(result, indent=2)


# ══════════════════════════════════════════════════════════════
#  PROPOSAL TOOLS (Git-for-Thought PRs)
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_propose(
    session_id: str,
    parent_thought_id: str,
    content: str,
    note: str = "",
    agent_id: str = "Agent",
) -> str:
    """📝 Submit a thought proposal (PR) — a formal suggestion branching from a parent thought.

    Use this when you want to suggest a change or alternative without hijacking the active cursor.
    The proposal must be reviewed (accepted/rejected) before it becomes active.

    Args:
        session_id: The session to submit the proposal in.
        parent_thought_id: The thought this proposal branches from.
        content: The proposed thought content.
        note: Optional note explaining why this proposal matters.
        agent_id: Your identity for provenance tracking.
    """
    result = await _call_node("POST", f"/session/{session_id}/proposal", {
        "parent_thought_id": parent_thought_id,
        "content": content,
        "note": note,
        "agent_id": agent_id,
    })
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_list_proposals(session_id: str) -> str:
    """📋 List all pending proposals in a session.

    Args:
        session_id: The session to check for proposals.
    """
    result = await _call_node("GET", f"/session/{session_id}/proposals")
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_review_proposal(
    session_id: str,
    proposal_id: str,
    action: str,
    actor: str = "Agent",
    reason: str = "",
) -> str:
    """⚡ Review a thought proposal — accept, reject, or supersede it.

    Accepting a proposal makes it the active thought and merges it into the reasoning chain.
    Rejecting records the reason in the witness trail.

    Args:
        session_id: The session containing the proposal.
        proposal_id: The proposal ID to review.
        action: Review action — accept, reject, or supersede.
        actor: Your identity for the review record.
        reason: Explanation for the review decision.
    """
    result = await _call_node("POST", f"/session/{session_id}/proposal/{proposal_id}/review", {
        "action": action,
        "actor": actor,
        "reason": reason or None,
    })
    return json.dumps(result, indent=2)


# ══════════════════════════════════════════════════════════════
#  VISUALIZATION & QUERY TOOLS
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_display_tree(session_id: str) -> str:
    """🌳 Render the full reasoning tree for a session.

    Returns the hierarchical thought structure with all branches, forks, and proposals.

    Args:
        session_id: The session to visualize.
    """
    result = await _call_node("GET", f"/session/{session_id}/tree")
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_search(session_id: str, pattern: str) -> str:
    """🔍 Search thoughts in a session by content pattern.

    Args:
        session_id: The session to search within.
        pattern: Text pattern to search for in thought content.
    """
    result = await _call_node("POST", f"/session/{session_id}/search", {
        "pattern": pattern,
    })
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_list_edges(session_id: str) -> str:
    """🔗 List all edges (relationships) between thoughts in a session.

    Shows how thoughts are connected: derives_from, supports, challenges, forks_from, merges_into, etc.

    Args:
        session_id: The session to inspect.
    """
    result = await _call_node("GET", f"/session/{session_id}/edges")
    return json.dumps(result, indent=2)


@mcp.tool()
async def thought_witness(session_id: str, limit: int = 50) -> str:
    """👁️ View the provenance/audit trail for a session.

    Shows who created, forked, reviewed, and merged thoughts — the full governance history.

    Args:
        session_id: The session to audit.
        limit: Maximum number of events to return (1-200).
    """
    result = await _call_node("GET", f"/session/{session_id}/witness?limit={limit}")
    return json.dumps(result, indent=2)


# ══════════════════════════════════════════════════════════════
#  IDENTITY
# ══════════════════════════════════════════════════════════════

@mcp.tool()
async def thought_node_identity() -> str:
    """📡 Check the Thought Engine Node's identity, seal, and status."""
    result = await _call_node("GET", "/")
    return json.dumps(result, indent=2)


# ── Server Entry Point ─────────────────────────────────────────

async def health(request: StarletteRequest):
    """Bridge-local readiness. Does not call the Thought Engine Node."""
    return JSONResponse({
        "service": "Thought Engine MCP Bridge",
        "status": "ONLINE",
        "mcp_version": "2026-07-28",
        "bridge_target_node": NODE_URL,
        "streamable_http_endpoint": "/mcp",
        "sse_endpoint": "/sse",
    })


def build_app() -> Starlette:
    """Build the dual-transport ASGI application from official MCP SDK apps."""
    modern = mcp.streamable_http_app(
        streamable_http_path="/mcp",
        json_response=True,
        stateless_http=True,
        host="0.0.0.0",
    )
    legacy = mcp.sse_app(
        sse_path="/sse",
        message_path="/messages/",
        host="0.0.0.0",
    )
    return Starlette(
        routes=[
            Route("/", health, methods=["GET"]),
            Route("/health", health, methods=["GET"]),
            *modern.routes,
            *legacy.routes,
        ],
        lifespan=lambda app: _bridge_lifespan(app, modern.router.lifespan_context),
    )


app = build_app()


if __name__ == "__main__":
    import uvicorn

    os.environ["PYTHONIOENCODING"] = "utf-8"
    os.environ["PYTHONUNBUFFERED"] = "1"
    print("🌉 Thought Engine Bridge — modern /mcp + legacy /sse starting...")
    print(f"   Node URL: {NODE_URL}")

    uvicorn.run(
        app,
        host="0.0.0.0",
        port=7860,
        proxy_headers=True,
        forwarded_allow_ips="*",
    )