mcp_server.py

"""DevMind MCP Server — exposes project memory graph to AI agents via stdio."""

import asyncio
import re

from dotenv import load_dotenv

load_dotenv()

from mcp.server.fastmcp import FastMCP

from db.client import DevMindDB, extract_rows
from brief.generator import generate_brief, AGENT_INSTRUCTIONS
from agent.graph import app as langgraph_app
from agent.nodes import close_shared_db

mcp = FastMCP("devmind")

# ---------------------------------------------------------------------------
# Agent instructions — appended to the first tool call in each session
# ---------------------------------------------------------------------------

_instructions_sent = False


def _with_instructions(result: str) -> str:
    """Append agent instructions to the first tool response in this session."""
    global _instructions_sent
    if not _instructions_sent:
        _instructions_sent = True
        return result + "\n" + AGENT_INSTRUCTIONS
    return result


# ---------------------------------------------------------------------------
# Write tool — runs the LangGraph enrichment pipeline
# ---------------------------------------------------------------------------

@mcp.tool()
async def log_activity(
    project: str,
    activity_type: str,
    summary: str,
    details: str = "",
    files: list[str] | None = None,
) -> str:
    """Report what you changed, fixed, or decided. Triggers analysis and returns insights
    about patterns, landmines, and connections. Call once per real task, then move on.
    Do NOT call this tool to report the act of logging itself.

    Args:
        project: Project name (e.g. "dev-mind", "auth-service").
        activity_type: One of fix, error, decision, pattern, feature.
        summary: One-line description of the activity.
        details: Optional freeform text with additional context.
        files: Optional list of file paths involved.
    """
    valid_types = {"fix", "error", "decision", "pattern", "feature"}
    if activity_type not in valid_types:
        return f"Invalid activity_type '{activity_type}'. Must be one of: {', '.join(sorted(valid_types))}"

    state = {
        "project": project,
        "activity_type": activity_type,
        "summary": summary,
        "details": details,
        "files": files or [],
    }
    # Overall pipeline timeout to prevent hanging forever
    PIPELINE_TIMEOUT = 60  # seconds
    try:
        result = await asyncio.wait_for(
            langgraph_app.ainvoke(state),
            timeout=PIPELINE_TIMEOUT,
        )
    except asyncio.TimeoutError:
        return f"log_activity timed out after {PIPELINE_TIMEOUT}s. The LLM or database may be unreachable. Activity was NOT saved."
    except Exception as exc:
        return f"log_activity failed: {exc}"
    finally:
        # Guarantee DB cleanup even if pipeline fails before respond node
        await close_shared_db()
    return result["response"]


# ---------------------------------------------------------------------------
# Full memory dump tool
# ---------------------------------------------------------------------------

_ALL_NODE_TABLES = ["error", "fix", "decision", "pattern", "file", "context"]

_DURATION_RE = re.compile(r"^(\d+)([dhms])$")

_DURATION_UNIT_MAP = {
    "d": "d",
    "h": "h",
    "m": "m",
    "s": "s",
}


def _parse_since(since: str) -> str | None:
    """Convert a human-friendly duration like '7d' or '30d' to SurrealDB duration format."""
    if not since:
        return None
    m = _DURATION_RE.match(since.strip())
    if not m:
        return None
    amount, unit = m.group(1), m.group(2)
    return f"{amount}{_DURATION_UNIT_MAP[unit]}"


@mcp.tool()
async def get_full_memory(
    project: str,
    types: list[str] | None = None,
    since: str = "",
) -> str:
    """Get complete knowledge graph dump for a project. Returns all stored nodes.
    REMINDER: After completing your task, call log_activity() to report what you did.

    Args:
        project: Project name.
        types: Optional list of node types to include (error, fix, decision, pattern, file, context). Empty means all.
        since: Optional time filter, e.g. '7d', '30d'. Empty means no time filter.
    """
    tables = types if types else _ALL_NODE_TABLES

    # Validate requested types
    invalid = [t for t in tables if t not in _ALL_NODE_TABLES]
    if invalid:
        return f"Invalid type(s): {', '.join(invalid)}. Valid types: {', '.join(_ALL_NODE_TABLES)}"

    surreal_duration = _parse_since(since)

    db = DevMindDB()
    try:
        await db.connect()
    except Exception as exc:
        return f"Database connection failed: {exc}"

    try:
        sections = []

        for table in tables:
            # file and context tables use 'updated', others use 'timestamp'
            time_field = "updated" if table in ("file", "context") else "timestamp"
            if surreal_duration:
                query = (
                    f"SELECT * FROM {table} WHERE project = $p "
                    f"AND {time_field} > time::now() - {surreal_duration} "
                    f"ORDER BY {time_field} DESC;"
                )
            else:
                query = f"SELECT * FROM {table} WHERE project = $p ORDER BY {time_field} DESC;"

            rows = extract_rows(await db.query(query, {"p": project}))

            section_lines = [f"{'=' * 40}", f"  {table.upper()} ({len(rows)} records)", f"{'=' * 40}"]

            if not rows:
                section_lines.append("  (none)")
            else:
                for row in rows:
                    section_lines.append("")
                    for key, value in row.items():
                        section_lines.append(f"  {key}: {value}")

            sections.append("\n".join(section_lines))

        header = f"FULL MEMORY DUMP -- {project}"
        if since:
            header += f" (since {since})"
        header += f"\nTables: {', '.join(tables)}\n"

        return _with_instructions(header + "\n\n".join(sections))

    finally:
        await db.disconnect()


@mcp.tool()
async def get_project_brief(project: str) -> str:
    """Call this FIRST when starting any coding session. Returns the full project brief:
    tech stack, active decisions, landmines, patterns, and recent fixes.
    REMINDER: After completing your task, you MUST call log_activity() to report what you did."""
    try:
        return _with_instructions(await generate_brief(project))
    except Exception as exc:
        return f"Failed to generate brief: {exc}. Is SurrealDB reachable?"


@mcp.tool()
async def get_file_context(project: str, file_path: str) -> str:
    """Get everything DevMind knows about a specific file: errors, fixes,
    decisions, patterns, and whether it is a landmine.
    REMINDER: After completing your task, call log_activity() to report what you did."""
    db = DevMindDB()
    try:
        await db.connect()
    except Exception as exc:
        return f"Database connection failed: {exc}"

    try:
        errors = extract_rows(await db.query(
            "SELECT * FROM error WHERE file = $f AND project = $p ORDER BY timestamp DESC;",
            {"f": file_path, "p": project},
        ))
        fixes = extract_rows(await db.query(
            "SELECT * FROM fix WHERE file = $f AND project = $p ORDER BY timestamp DESC;",
            {"f": file_path, "p": project},
        ))
        decisions = extract_rows(await db.query(
            "SELECT * FROM decision WHERE file = $f AND project = $p ORDER BY timestamp DESC;",
            {"f": file_path, "p": project},
        ))
        patterns = extract_rows(await db.query(
            "SELECT * FROM pattern WHERE file = $f AND project = $p ORDER BY timestamp DESC;",
            {"f": file_path, "p": project},
        ))

        lines = [f"FILE CONTEXT: {file_path} ({project})", ""]

        is_landmine = any(e.get("landmine") for e in errors)
        if is_landmine:
            lines.append("WARNING: This file is a LANDMINE (has broken repeatedly)")
            lines.append("")

        lines.append(f"ERRORS ({len(errors)})")
        for e in errors:
            status = "resolved" if e.get("resolved") else "OPEN"
            lines.append(f"  * [{status}] {e.get('message', 'no message')}")

        lines.append("")
        lines.append(f"FIXES ({len(fixes)})")
        for f in fixes:
            lines.append(f"  * {f.get('description', 'no description')}")

        lines.append("")
        lines.append(f"DECISIONS ({len(decisions)})")
        for d in decisions:
            lines.append(f"  * {d.get('summary', 'no summary')}")
            if d.get("reasoning"):
                lines.append(f"    Reasoning: {d['reasoning']}")

        lines.append("")
        lines.append(f"PATTERNS ({len(patterns)})")
        for p in patterns:
            lines.append(f"  * {p.get('description', 'no description')}")

        return _with_instructions("\n".join(lines))

    finally:
        await db.disconnect()


@mcp.tool()
async def get_landmines(project: str) -> str:
    """Get files that have broken repeatedly (landmines).
    These are danger zones that need extra care when modifying."""
    db = DevMindDB()
    try:
        await db.connect()
    except Exception as exc:
        return f"Database connection failed: {exc}"

    try:
        landmines = extract_rows(await db.query(
            "SELECT * FROM error WHERE project = $p AND landmine = true;",
            {"p": project},
        ))

        if not landmines:
            return f"No landmines found for project {project}."

        lines = [f"LANDMINES ({len(landmines)} danger zones in {project})", ""]

        for lm in landmines:
            lm_id = lm.get("id", "")
            fix_result = await db.query(
                "SELECT ->fixed_by->fix.* AS fixes FROM $lid;",
                {"lid": lm_id},
            )
            fix_rows = extract_rows(fix_result)
            last_fix = "no fix recorded"
            if fix_rows:
                linked = fix_rows[0].get("fixes", [])
                if linked and isinstance(linked[-1], dict):
                    last_fix = linked[-1].get("description", "no fix recorded")

            lines.append(f"  * {lm.get('file', 'unknown')}: {lm.get('message', 'unknown')}")
            lines.append(f"    Broke {lm.get('count', 1)} times. Last fix: {last_fix}")

        return _with_instructions("\n".join(lines))

    finally:
        await db.disconnect()


@mcp.tool()
async def search_memory(project: str, query: str) -> str:
    """Search across all DevMind knowledge (errors, fixes, decisions, patterns) by keyword."""
    db = DevMindDB()
    try:
        await db.connect()
    except Exception as exc:
        return f"Database connection failed: {exc}"

    try:
        results = []
        q_lower = query.lower()

        errors = extract_rows(await db.query(
            "SELECT * FROM error WHERE project = $p AND (string::lowercase(message) CONTAINS $q OR string::lowercase(raw_message) CONTAINS $q);",
            {"p": project, "q": q_lower},
        ))
        for e in errors:
            results.append(f"  [error] {e.get('message', '')} (file: {e.get('file', 'unknown')})")

        fixes = extract_rows(await db.query(
            "SELECT * FROM fix WHERE project = $p AND string::lowercase(description) CONTAINS $q;",
            {"p": project, "q": q_lower},
        ))
        for f in fixes:
            results.append(f"  [fix] {f.get('description', '')} (file: {f.get('file', 'unknown')})")

        decisions = extract_rows(await db.query(
            "SELECT * FROM decision WHERE project = $p AND (string::lowercase(summary) CONTAINS $q OR string::lowercase(reasoning) CONTAINS $q);",
            {"p": project, "q": q_lower},
        ))
        for d in decisions:
            results.append(f"  [decision] {d.get('summary', '')}")

        patterns = extract_rows(await db.query(
            "SELECT * FROM pattern WHERE project = $p AND string::lowercase(description) CONTAINS $q;",
            {"p": project, "q": q_lower},
        ))
        for p in patterns:
            results.append(f"  [pattern] {p.get('description', '')}")

        if not results:
            return _with_instructions(f"No results found for '{query}' in project {project}.")

        header = f"SEARCH RESULTS for '{query}' in {project} ({len(results)} matches)\n"
        return _with_instructions(header + "\n".join(results))

    finally:
        await db.disconnect()


@mcp.tool()
async def get_recent_activity(project: str, limit: int = 10) -> str:
    """Get the most recent events captured by DevMind across all types."""
    db = DevMindDB()
    try:
        await db.connect()
    except Exception as exc:
        return f"Database connection failed: {exc}"

    try:
        # Query each table and merge by timestamp
        errors = extract_rows(await db.query(
            "SELECT 'error' AS type, message AS summary, file, timestamp FROM error WHERE project = $p ORDER BY timestamp DESC LIMIT $l;",
            {"p": project, "l": limit},
        ))
        fixes = extract_rows(await db.query(
            "SELECT 'fix' AS type, description AS summary, file, timestamp FROM fix WHERE project = $p ORDER BY timestamp DESC LIMIT $l;",
            {"p": project, "l": limit},
        ))
        decisions = extract_rows(await db.query(
            "SELECT 'decision' AS type, summary, file, timestamp FROM decision WHERE project = $p ORDER BY timestamp DESC LIMIT $l;",
            {"p": project, "l": limit},
        ))
        patterns = extract_rows(await db.query(
            "SELECT 'pattern' AS type, description AS summary, file, timestamp FROM pattern WHERE project = $p ORDER BY timestamp DESC LIMIT $l;",
            {"p": project, "l": limit},
        ))

        all_events = errors + fixes + decisions + patterns
        all_events.sort(key=lambda x: str(x.get("timestamp", "")), reverse=True)
        all_events = all_events[:limit]

        if not all_events:
            return f"No activity found for project {project}."

        lines = [f"RECENT ACTIVITY in {project} (last {len(all_events)} events)", ""]
        for ev in all_events:
            ts = str(ev.get("timestamp", ""))[:10]
            lines.append(f"  [{ev.get('type', '?')}] {ev.get('summary', 'no summary')} -- {ts}")

        return _with_instructions("\n".join(lines))

    finally:
        await db.disconnect()


@mcp.tool()
async def get_project_context(project: str) -> str:
    """Get auto-detected project metadata: tech stack, purpose, and conventions.
    This is learned automatically from code changes over time."""
    db = DevMindDB()
    try:
        await db.connect()
    except Exception as exc:
        return f"Database connection failed: {exc}"

    try:
        rows = extract_rows(await db.query(
            "SELECT * FROM context WHERE project = $p;",
            {"p": project},
        ))

        if not rows:
            return _with_instructions(f"No context detected yet for project {project}. Context is learned automatically as you log activity.")

        lines = [f"PROJECT CONTEXT for {project}", ""]
        for row in rows:
            key = row.get("key", "unknown")
            value = row.get("value", "")
            confidence = row.get("confidence", 0)
            pct = int(confidence * 100)
            lines.append(f"  {key}: {value} ({pct}% confidence)")

        return _with_instructions("\n".join(lines))

    finally:
        await db.disconnect()


if __name__ == "__main__":
    mcp.run(transport="stdio")