app.py

"""
==========================================================================
🧠 Thought Engine Node — Sovereign REST Substrate
==========================================================================
Contract: C-THOUGHT-NODE-001 (Phase 1)
Vault:    thought-vault (Cloudflare D1)
Stack:    FastAPI + httpx + Cloudflare D1 HTTP API

A persistent, queryable, forkable reasoning system with provenance,
pattern memory, and multiple entry surfaces.
==========================================================================
"""

import os
import uuid
import json
from datetime import datetime, timezone
from typing import Optional, List

from dotenv import load_dotenv
load_dotenv()

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware

import d1_client
from models import (
    CreateSessionReq, AddThoughtReq, ForkThoughtReq,
    CreateProposalReq, ReviewProposalReq, SearchReq,
    SessionInfo, ThoughtInfo, EdgeInfo, TreeNode,
    ThoughtClass, ThoughtStatus, EdgeRelation, ReviewAction,
)

# ── Application ────────────────────────────────────────────────
app = FastAPI(
    title="Thought Engine Node",
    description="Sovereign reasoning substrate — C-THOUGHT-NODE-001",
    version="0.1.0",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

NODE_SEAL = None


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _id(prefix: str = "t") -> str:
    return f"{prefix}-{uuid.uuid4().hex[:12]}"


# ── Lifecycle ──────────────────────────────────────────────────
@app.on_event("startup")
async def startup():
    global NODE_SEAL
    seg = uuid.uuid4().hex[:8].upper()
    NODE_SEAL = f"⟦ THOUGHT :: PANTHEON-TE :: 🧠-{datetime.now().strftime('%Y%m%d')}-{seg[:4]}-{seg[4:]} :: ACTIVE ⟧"
    print(f"\n{'='*60}")
    print(f"🧠 Thought Engine Node — Phase 1 Sovereign Substrate")
    print(f"   Seal: {NODE_SEAL}")
    print(f"   Time: {_now()}")
    print(f"{'='*60}\n")


# ── Identity ───────────────────────────────────────────────────
@app.get("/")
async def identity():
    count = await d1_client.execute_sql("SELECT COUNT(*) as c FROM sessions")
    session_count = count[0]["c"] if count else 0
    return {
        "node": "Thought Engine Node",
        "contract": "C-THOUGHT-NODE-001",
        "version": "0.1.0",
        "seal": NODE_SEAL,
        "sessions": session_count,
        "status": "ACTIVE",
        "timestamp": _now(),
    }


# ══════════════════════════════════════════════════════════════
#  SESSIONS
# ══════════════════════════════════════════════════════════════

@app.post("/session")
async def create_session(req: CreateSessionReq):
    """Start a new reasoning session with an initial root thought."""
    session_id = _id("ses")
    thought_id = _id("th")
    title = req.title or req.initial_thought[:80]

    # Insert session
    await d1_client.execute_sql(
        "INSERT INTO sessions (session_id, title, created_by, created_at, status, root_thought_id, active_thought_id) "
        "VALUES (?, ?, ?, ?, 'active', ?, ?)",
        [session_id, title, req.agent_id, _now(), thought_id, thought_id],
    )
    # Insert root thought
    await d1_client.execute_sql(
        "INSERT INTO thought_units (thought_id, session_id, content, thought_class, origin_agent, status, confidence, created_at) "
        "VALUES (?, ?, ?, ?, ?, 'open', ?, ?)",
        [thought_id, session_id, req.initial_thought, req.thought_class.value, req.agent_id, None, _now()],
    )
    # Witness event
    await d1_client.execute_sql(
        "INSERT INTO witness_events (event_id, session_id, thought_id, event_type, actor, detail, created_at) "
        "VALUES (?, ?, ?, 'created', ?, 'Session started', ?)",
        [_id("ev"), session_id, thought_id, req.agent_id, _now()],
    )

    return {
        "session_id": session_id,
        "root_thought_id": thought_id,
        "title": title,
        "status": "active",
        "message": "🧠 Reasoning session created.",
    }


@app.get("/session/{session_id}")
async def get_session(session_id: str):
    """Retrieve session metadata."""
    rows = await d1_client.execute_sql(
        "SELECT * FROM sessions WHERE session_id = ?", [session_id]
    )
    if not rows:
        raise HTTPException(404, "Session not found")
    return rows[0]


@app.get("/sessions")
async def list_sessions(
    status: Optional[str] = Query(None),
    limit: int = Query(20, ge=1, le=100),
):
    """List all sessions, optionally filtered by status."""
    if status:
        rows = await d1_client.execute_sql(
            "SELECT * FROM sessions WHERE status = ? ORDER BY created_at DESC LIMIT ?",
            [status, limit],
        )
    else:
        rows = await d1_client.execute_sql(
            "SELECT * FROM sessions ORDER BY created_at DESC LIMIT ?", [limit]
        )
    return {"total": len(rows), "sessions": rows}


# ══════════════════════════════════════════════════════════════
#  THOUGHTS
# ══════════════════════════════════════════════════════════════

@app.post("/session/{session_id}/thought")
async def add_thought(session_id: str, req: AddThoughtReq):
    """Add a reasoning step to the session's active chain."""
    # Verify session exists
    ses = await d1_client.execute_sql(
        "SELECT * FROM sessions WHERE session_id = ?", [session_id]
    )
    if not ses:
        raise HTTPException(404, "Session not found")

    thought_id = _id("th")
    parent_id = req.parent_thought_id or ses[0]["active_thought_id"]

    # Insert thought unit
    await d1_client.execute_sql(
        "INSERT INTO thought_units (thought_id, session_id, content, thought_class, origin_agent, status, confidence, created_at) "
        "VALUES (?, ?, ?, ?, ?, 'open', ?, ?)",
        [thought_id, session_id, req.content, req.thought_class.value, req.agent_id, req.confidence, _now()],
    )
    # Insert edge from parent
    await d1_client.execute_sql(
        "INSERT INTO thought_edges (edge_id, session_id, source_id, target_id, relation, created_at) "
        "VALUES (?, ?, ?, ?, 'derives_from', ?)",
        [_id("ed"), session_id, parent_id, thought_id, _now()],
    )
    # Update active pointer
    await d1_client.execute_sql(
        "UPDATE sessions SET active_thought_id = ?, updated_at = ? WHERE session_id = ?",
        [thought_id, _now(), session_id],
    )
    # Witness
    await d1_client.execute_sql(
        "INSERT INTO witness_events (event_id, session_id, thought_id, event_type, actor, detail, created_at) "
        "VALUES (?, ?, ?, 'created', ?, ?, ?)",
        [_id("ev"), session_id, thought_id, req.agent_id, f"Added {req.thought_class.value}", _now()],
    )

    return {
        "thought_id": thought_id,
        "parent_id": parent_id,
        "thought_class": req.thought_class.value,
        "message": f"✅ Thought added to session.",
    }


@app.get("/session/{session_id}/thoughts")
async def list_thoughts(session_id: str):
    """List all thought units in a session."""
    rows = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE session_id = ? ORDER BY created_at ASC",
        [session_id],
    )
    return {"session_id": session_id, "total": len(rows), "thoughts": rows}


# ══════════════════════════════════════════════════════════════
#  FORKING (Git-for-Thought Branching)
# ══════════════════════════════════════════════════════════════

@app.post("/session/{session_id}/fork")
async def fork_thought(session_id: str, req: ForkThoughtReq):
    """Fork a thought chain — create a branch for alternative exploration."""
    # Verify source thought
    source = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE thought_id = ? AND session_id = ?",
        [req.source_thought_id, session_id],
    )
    if not source:
        raise HTTPException(404, "Source thought not found in this session")

    fork_id = _id("th")
    src = source[0]

    # Create the forked thought node (copy of source with new id)
    await d1_client.execute_sql(
        "INSERT INTO thought_units (thought_id, session_id, content, thought_class, origin_agent, status, confidence, created_at, metadata) "
        "VALUES (?, ?, ?, ?, ?, 'open', ?, ?, ?)",
        [fork_id, session_id, src["content"], src["thought_class"], req.agent_id, src.get("confidence"),
         _now(), json.dumps({"forked_from": req.source_thought_id, "branch_label": req.branch_label})],
    )
    # Edge: forks_from
    await d1_client.execute_sql(
        "INSERT INTO thought_edges (edge_id, session_id, source_id, target_id, relation, created_at) "
        "VALUES (?, ?, ?, ?, 'forks_from', ?)",
        [_id("ed"), session_id, req.source_thought_id, fork_id, _now()],
    )
    # Move active pointer to the fork
    await d1_client.execute_sql(
        "UPDATE sessions SET active_thought_id = ?, updated_at = ? WHERE session_id = ?",
        [fork_id, _now(), session_id],
    )
    # Witness
    await d1_client.execute_sql(
        "INSERT INTO witness_events (event_id, session_id, thought_id, event_type, actor, detail, created_at) "
        "VALUES (?, ?, ?, 'forked', ?, ?, ?)",
        [_id("ev"), session_id, fork_id, req.agent_id, f"Forked from {req.source_thought_id} as '{req.branch_label}'", _now()],
    )

    return {
        "forked_thought_id": fork_id,
        "source_thought_id": req.source_thought_id,
        "branch_label": req.branch_label,
        "message": f"🌿 Forked thought chain: {req.branch_label}",
    }


# ══════════════════════════════════════════════════════════════
#  PROPOSALS (Git-for-Thought PRs)
# ══════════════════════════════════════════════════════════════

@app.post("/session/{session_id}/proposal")
async def create_proposal(session_id: str, req: CreateProposalReq):
    """Submit a thought proposal (PR) branching from a parent."""
    parent = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE thought_id = ? AND session_id = ?",
        [req.parent_thought_id, session_id],
    )
    if not parent:
        raise HTTPException(404, "Parent thought not found")

    proposal_id = _id("pr")

    # Insert proposal thought
    await d1_client.execute_sql(
        "INSERT INTO thought_units (thought_id, session_id, content, thought_class, origin_agent, status, confidence, created_at, metadata) "
        "VALUES (?, ?, ?, 'proposal', ?, 'proposed', NULL, ?, ?)",
        [proposal_id, session_id, req.content, req.agent_id, _now(),
         json.dumps({"proposal_note": req.note, "target_branch": req.parent_thought_id})],
    )
    # Edge
    await d1_client.execute_sql(
        "INSERT INTO thought_edges (edge_id, session_id, source_id, target_id, relation, created_at) "
        "VALUES (?, ?, ?, ?, 'derives_from', ?)",
        [_id("ed"), session_id, req.parent_thought_id, proposal_id, _now()],
    )
    # Review record
    await d1_client.execute_sql(
        "INSERT INTO thought_reviews (review_id, session_id, thought_id, action, actor, reason, created_at) "
        "VALUES (?, ?, ?, 'propose', ?, ?, ?)",
        [_id("rv"), session_id, proposal_id, req.agent_id, req.note, _now()],
    )

    return {
        "proposal_id": proposal_id,
        "parent_id": req.parent_thought_id,
        "status": "proposed",
        "message": f"📝 Proposal submitted by {req.agent_id}",
    }


@app.get("/session/{session_id}/proposals")
async def list_proposals(session_id: str):
    """List all pending proposals in a session."""
    rows = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE session_id = ? AND status = 'proposed' ORDER BY created_at ASC",
        [session_id],
    )
    return {"session_id": session_id, "total": len(rows), "proposals": rows}


@app.post("/session/{session_id}/proposal/{proposal_id}/review")
async def review_proposal(session_id: str, proposal_id: str, req: ReviewProposalReq):
    """Accept, reject, or supersede a proposal."""
    proposal = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE thought_id = ? AND session_id = ? AND status = 'proposed'",
        [proposal_id, session_id],
    )
    if not proposal:
        raise HTTPException(404, "Proposal not found or not in 'proposed' status")

    new_status_map = {
        ReviewAction.ACCEPT: "accepted",
        ReviewAction.REJECT: "rejected",
        ReviewAction.SUPERSEDE: "superseded",
    }
    new_status = new_status_map.get(req.action)
    if not new_status:
        raise HTTPException(400, "Invalid review action for this endpoint")

    # Update thought status
    await d1_client.execute_sql(
        "UPDATE thought_units SET status = ? WHERE thought_id = ?",
        [new_status, proposal_id],
    )
    # Review record
    await d1_client.execute_sql(
        "INSERT INTO thought_reviews (review_id, session_id, thought_id, action, actor, reason, created_at) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        [_id("rv"), session_id, proposal_id, req.action.value, req.actor, req.reason, _now()],
    )
    # If accepted, move active pointer
    if req.action == ReviewAction.ACCEPT:
        await d1_client.execute_sql(
            "UPDATE sessions SET active_thought_id = ?, updated_at = ? WHERE session_id = ?",
            [proposal_id, _now(), session_id],
        )
    # Witness
    await d1_client.execute_sql(
        "INSERT INTO witness_events (event_id, session_id, thought_id, event_type, actor, detail, created_at) "
        "VALUES (?, ?, ?, 'reviewed', ?, ?, ?)",
        [_id("ev"), session_id, proposal_id, req.actor,
         f"{req.action.value}: {req.reason or 'No reason given'}", _now()],
    )

    return {
        "proposal_id": proposal_id,
        "new_status": new_status,
        "action": req.action.value,
        "message": f"{'⚡ Merged' if req.action == ReviewAction.ACCEPT else '❌ Rejected' if req.action == ReviewAction.REJECT else '🔄 Superseded'}: {proposal_id}",
    }


# ══════════════════════════════════════════════════════════════
#  TREE VISUALIZATION
# ══════════════════════════════════════════════════════════════

@app.get("/session/{session_id}/tree")
async def get_tree(session_id: str):
    """Render the full reasoning tree for a session."""
    ses = await d1_client.execute_sql(
        "SELECT * FROM sessions WHERE session_id = ?", [session_id]
    )
    if not ses:
        raise HTTPException(404, "Session not found")

    thoughts = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE session_id = ? ORDER BY created_at ASC",
        [session_id],
    )
    edges = await d1_client.execute_sql(
        "SELECT * FROM thought_edges WHERE session_id = ?", [session_id]
    )

    # Build adjacency map (parent -> children)
    children_map: dict[str, list[str]] = {}
    for edge in edges:
        src = edge["source_id"]
        tgt = edge["target_id"]
        children_map.setdefault(src, []).append(tgt)

    thought_map = {t["thought_id"]: t for t in thoughts}

    def build_node(tid: str) -> dict:
        t = thought_map.get(tid, {})
        return {
            "thought_id": tid,
            "content": t.get("content", ""),
            "thought_class": t.get("thought_class", ""),
            "status": t.get("status", ""),
            "origin_agent": t.get("origin_agent", ""),
            "children": [build_node(cid) for cid in children_map.get(tid, [])],
        }

    root_id = ses[0].get("root_thought_id")
    tree = build_node(root_id) if root_id else {}

    return {
        "session_id": session_id,
        "title": ses[0].get("title"),
        "total_thoughts": len(thoughts),
        "total_edges": len(edges),
        "active_thought_id": ses[0].get("active_thought_id"),
        "tree": tree,
    }


# ══════════════════════════════════════════════════════════════
#  SEARCH & EDGES
# ══════════════════════════════════════════════════════════════

@app.post("/session/{session_id}/search")
async def search_thoughts(session_id: str, req: SearchReq):
    """Search thoughts in a session by content pattern."""
    rows = await d1_client.execute_sql(
        "SELECT * FROM thought_units WHERE session_id = ? AND content LIKE ? ORDER BY created_at ASC",
        [session_id, f"%{req.pattern}%"],
    )
    return {"session_id": session_id, "pattern": req.pattern, "matches": len(rows), "results": rows}


@app.get("/session/{session_id}/edges")
async def list_edges(session_id: str):
    """List all edges (relationships) in a session."""
    rows = await d1_client.execute_sql(
        "SELECT * FROM thought_edges WHERE session_id = ? ORDER BY created_at ASC",
        [session_id],
    )
    return {"session_id": session_id, "total": len(rows), "edges": rows}


# ══════════════════════════════════════════════════════════════
#  WITNESS / PROVENANCE
# ══════════════════════════════════════════════════════════════

@app.get("/session/{session_id}/witness")
async def list_witness_events(session_id: str, limit: int = Query(50, ge=1, le=200)):
    """Get the audit trail for a session."""
    rows = await d1_client.execute_sql(
        "SELECT * FROM witness_events WHERE session_id = ? ORDER BY created_at DESC LIMIT ?",
        [session_id, limit],
    )
    return {"session_id": session_id, "total": len(rows), "events": rows}


# ══════════════════════════════════════════════════════════════
#  HEALTH
# ══════════════════════════════════════════════════════════════

@app.get("/health")
async def health():
    return {"status": "ok", "timestamp": _now()}