Building Local-First AI Apps: MCP Integration, Offline Memory & Cost Optimization

┌─────────────────────────────────────────────────────┐
│                   Your Application                   │
│  ┌──────────┐   ┌──────────────┐   ┌─────────────┐  │
│  │  UI/CLI  │──▶│  App Logic   │──▶│  Memory     │  │
│  └──────────┘   └──────────────┘   │  Store      │  │
└─────────────────────────────────────────────────────┘
                      │
              ┌───────▼────────┐
              │  MCP Client    │◀── Handles tool calls, resource reads
              │  (Embedded)    │
              └───────┬────────┘
                      │ JSON-RPC over stdio/SSE/WebSocket
        ┌─────────────┼─────────────┐
        │             │             │
   ┌────▼────┐  ┌────▼────┐  ┌────▼────┐
   │ MCP     │  │ MCP     │  │ MCP     │
   │ Server  │  │ Server  │  │ Server  │
   │ (Files) │  │ (Search)│  │(Memory) │
   └─────────┘  └─────────┘  └─────────┘
        │
        ▼
   ┌──────────────┐
   │ Local LLM    │  OpenWebUI, llama.cpp, Ollama
   │ (Primary)    │
   └──────────────┘
        │ (fallback)
        ▼
   ┌──────────────┐
   │ Cloud API    │  OpenAI, Anthropic, etc.
   └──────────────┘

# mcp_server/local_first_server.py
import json
import asyncio
from pathlib import Path
from typing import Any
from mcp.server import Server
from mcp.types import Tool, Resource, TextContent
import sqlite_vec


class LocalFirstMCPServer:
    """
    MCP server providing file system access, local search,
    and persistent vector memory for local-first AI apps.
    """

    def __init__(self, data_dir: Path = Path("~/.localfirst")):
        self.data_dir = data_dir.expanduser()
        self.data_dir.mkdir(parents=True, exist_ok=True)
        self.db_path = self.data_dir / "memory.db"
        self._init_db()

    def _init_db(self):
        """Initialize SQLite with vec extension for embeddings."""
        import sqlite3
        self.conn = sqlite3.connect(self.db_path)
        self.conn.enable_load_extension(True)
        self.conn.load_extension("sqlite_vec")
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                content TEXT NOT NULL,
                embedding BLOB,
                source TEXT,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                metadata JSON
            )
        """)
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS memories_vec
            USING vec0(embedding FLOAT(1536))
        """)
        self.conn.commit()

    async def list_tools(self) -> list[Tool]:
        return [
            Tool(
                name="remember",
                description="Store a fact or piece of context for future retrieval",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "content": {"type": "string", "description": "The fact to remember"},
                        "source": {"type": "string", "description": "Where this came from (user, system, document)"},
                        "metadata": {"type": "object", "description": "Optional tags and labels"}
                    },
                    "required": ["content"]
                }
            ),
            Tool(
                name="recall",
                description="Retrieve relevant memories by semantic search",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Search query"},
                        "limit": {"type": "integer", "default": 5, "description": "Max results"}
                    },
                    "required": ["query"]
                }
            ),
            Tool(
                name="read_file",
                description="Read a file from the local filesystem",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "Absolute or relative path"},
                        "lines": {"type": "integer", "default": 100, "description": "Max lines to read"}
                    },
                    "required": ["path"]
                }
            ),
        ]

    async def call_tool(self, name: str, args: dict) -> list[TextContent]:
        import numpy as np

        if name == "remember":
            return await self._remember(args)
        elif name == "recall":
            return await self._recall(args)
        elif name == "read_file":
            return await self._read_file(args)
        else:
            raise ValueError(f"Unknown tool: {name}")

    async def _remember(self, args: dict) -> list[TextContent]:
        """Store content with embedding for semantic search."""
        content = args["content"]
        source = args.get("source", "user")
        metadata = json.dumps(args.get("metadata", {}))

        # Generate embedding using local model (e.g., via local Ollama endpoint)
        embedding = await self._embed(content)

        cursor = self.conn.execute(
            "INSERT INTO memories (content, embedding, source, metadata) VALUES (?, ?, ?, ?)",
            (content, embedding.tobytes(), source, metadata)
        )
        mem_id = cursor.lastrowid

        # Sync to vector index
        self.conn.execute(
            "INSERT INTO memories_vec (rowid, embedding) VALUES (?, ?)",
            (mem_id, embedding.tobytes())
        )
        self.conn.commit()

        return [TextContent(type="text", text=f"Stored memory #{mem_id}: {content[:80]}...")]

    async def _recall(self, args: dict) -> list[TextContent]:
        """Semantic search over stored memories."""
        query = args["query"]
        limit = args.get("limit", 5)

        query_embedding = await self._embed(query)

        results = self.conn.execute("""
            SELECT m.id, m.content, m.source, m.created_at, m.metadata,
                   vec.distance
            FROM memories_vec AS v
            JOIN memories AS m ON m.id = v.rowid
            WHERE v.embedding MATCH ? AND k = ?
            ORDER BY vec.distance
            LIMIT ?
        """, (query_embedding.tobytes(), limit, limit)).fetchall()

        if not results:
            return [TextContent(type="text", text="No relevant memories found.")]

        snippets = []
        for row in results:
            mem_id, content, source, created_at, metadata, distance = row
            snippets.append(f"#{mem_id} [{source}] ({created_at}): {content[:200]}")

        return [TextContent(type="text", text="\n\n".join(snippets))]

    async def _read_file(self, args: dict) -> list[TextContent]:
        """Safely read a file from the filesystem."""
        path = Path(args["path"]).expanduser().resolve()
        max_lines = args.get("lines", 100)

        if not str(path).startswith(str(self.data_dir.parent)):
            return [TextContent(type="text", text="Error: Access denied to path outside allowed tree.")]

        try:
            text = path.read_text(encoding="utf-8")
            lines = text.splitlines()[:max_lines]
            return [TextContent(type="text", text="\n".join(lines))]
        except FileNotFoundError:
            return [TextContent(type="text", text=f"File not found: {path}")]
        except Exception as e:
            return [TextContent(type="text", text=f"Error reading file: {e}")]

    async def _embed(self, text: str) -> np.ndarray:
        """Generate embedding using local Ollama endpoint."""
        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "http://localhost:11434/api/embeddings",
                json={"model": "nomic-embed-text", "prompt": text}
            )
            resp.raise_for_status()
            return np.array(resp.json()["embedding"])

    def close(self):
        self.conn.close()


async def main():
    server = Server("local-first")
    mcp = LocalFirstMCPServer()

    @server.list_tools()
    async def handle_list_tools():
        return await mcp.list_tools()

    @server.call_tool()
    async def handle_call_tool(name: str, args: dict):
        return await mcp.call_tool(name, args)

    async with server.run_stdio_server():
        await asyncio.Future()  # run forever


if __name__ == "__main__":
    asyncio.run(main())

// .ollama/config.json or equivalent
{
  "mcpServers": {
    "local-first": {
      "command": "python3",
      "args": ["mcp_server/local_first_server.py"],
      "env": {
        "OLLAMA_HOST": "http://localhost:11434"
      }
    }
  }
}

┌─────────────────────────────────────────────────┐
│                  SHORT-TERM BUFFER              │
│  Last N messages (raw text, in-context)        │
│  Size: ~2K-4K tokens, fresh in every request    │
├─────────────────────────────────────────────────┤
│               RECENT MEMORY (vector)            │
│  Last ~2 weeks of interactions, embedded        │
│  Retrieved via semantic search when relevant    │
│  Compressed to ~500 tokens max per query        │
├─────────────────────────────────────────────────┤
│              LONG-TERM STORE (summarized)       │
│  Old interactions → summaries + key facts       │
│  Stored as structured records with embeddings   │
│  Never injected raw — only summaries fetched    │
└─────────────────────────────────────────────────┘

# memory/store.py
import json
import asyncio
from pathlib import Path
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional
import numpy as np

@dataclass
class MemoryRecord:
    id: str
    content: str
    role: str  # "user", "assistant", "system"
    timestamp: datetime
    embedding: Optional[np.ndarray] = None
    summary: Optional[str] = None
    importance: float = 1.0  # 0.0 to 1.0, set by model or heuristics

    def to_context_snippet(self, max_tokens: int = 200) -> str:
        if self.summary:
            return f"[{self.role}] Summary: {self.summary}"
        truncated = self.content[:max_tokens * 4]
        return f"[{self.role}] {truncated}"


class LocalMemoryStore:
    """
    Tiered local memory with vector search, automatic summarization,
    and recency-weighted context injection.

    Designed for local-first AI apps where every request must work
    without network connectivity.
    """

    def __init__(self, db_path: Path = Path("~/.localfirst/memory.db")):
        self.db_path = db_path.expanduser()
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()
        self._session_messages: list[MemoryRecord] = []
        self._embed_model = self._load_embed_model()

    def _init_db(self):
        import sqlite3
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row
        self.conn.enable_load_extension(True)
        self.conn.load_extension("sqlite_vec")

        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                id TEXT PRIMARY KEY,
                content TEXT NOT NULL,
                role TEXT NOT NULL,
                timestamp DATETIME NOT NULL,
                embedding BLOB,
                summary TEXT,
                importance REAL DEFAULT 1.0,
                session_id TEXT
            )
        """)
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS memories_vec
            USING vec0(embedding FLOAT(768))
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_memories_session
            ON memories(session_id, timestamp DESC)
        """)
        self.conn.commit()

    def add(self, role: str, content: str, session_id: str = "default") -> str:
        """Add a message to short-term buffer and persist to long-term store."""
        import uuid
        mem_id = str(uuid.uuid4())
        record = MemoryRecord(
            id=mem_id,
            content=content,
            role=role,
            timestamp=datetime.utcnow(),
            session_id=session_id
        )
        self._session_messages.append(record)
        self._persist(record)
        return mem_id

    def _persist(self, record: MemoryRecord):
        embedding = self._embed(record.content)
        record.embedding = embedding

        self.conn.execute("""
            INSERT OR REPLACE INTO memories
            (id, content, role, timestamp, embedding, session_id, importance)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (
            record.id, record.content, record.role,
            record.timestamp.isoformat(),
            embedding.tobytes(), record.session_id, record.importance
        ))

        self.conn.execute("""
            INSERT OR REPLACE INTO memories_vec (rowid, embedding)
            VALUES ((SELECT id FROM memories WHERE id=?), ?)
        """, (record.id, embedding.tobytes()))
        self.conn.commit()

    def get_context(self, query: str, max_tokens: int = 1500) -> str:
        """
        Build a context prompt by combining short-term messages
        with semantically relevant long-term memories.
        """
        # 1. Short-term: recent conversation history
        recent = self._session_messages[-8:]  # last 8 messages
        short_term = "\n".join(m.to_context_snippet(150) for m in recent)

        # 2. Long-term: semantic recall
        long_term = self._recall(query, max_results=3)

        # 3. Combine with priority ordering
        context_parts = []
        if long_term:
            context_parts.append("## Relevant Past Context\n" + long_term)
        if short_term:
            context_parts.append("## Recent Conversation\n" + short_term)

        full_context = "\n\n".join(context_parts)

        # 4. Trim to token budget (rough char-based estimate)
        max_chars = max_tokens * 4
        if len(full_context) > max_chars:
            full_context = full_context[:max_chars] + "\n...[context truncated]"

        return full_context

    def _recall(self, query: str, max_results: int = 3) -> str:
        """Semantic search over long-term memory."""
        query_emb = self._embed(query)

        results = self.conn.execute("""
            SELECT m.id, m.content, m.role, m.summary,
                   m.timestamp, m.importance, v.distance
            FROM memories_vec AS v
            JOIN memories AS m ON m.id = v.rowid
            WHERE v.embedding MATCH ? AND k = ?
            ORDER BY v.distance ASC, m.importance DESC
            LIMIT ?
        """, (query_emb.tobytes(), max_results, max_results)).fetchall()

        snippets = []
        for r in results:
            if r["summary"]:
                snippets.append(f"• [{r['role']}] Summary: {r['summary']}")
            else:
                snippets.append(f"• [{r['role']}] {r['content'][:200]}")

        return "\n".join(snippets)

    def summarize_old_sessions(self, older_than_days: int = 14):
        """
        Replace old raw memories with AI-generated summaries.
        Called periodically to reclaim context window space.
        """
        cutoff = (datetime.utcnow() - timedelta(days=older_than_days)).isoformat()

        old_records = self.conn.execute(
            "SELECT id, content, role FROM memories WHERE timestamp < ? ORDER BY timestamp ASC",
            (cutoff,)
        ).fetchall()

        # Group by session and summarize in batches
        batches = self._chunk(old_records, size=10)
        for batch in batches:
            summary_text = self._generate_summary(batch)
            self._save_summary(batch[0]["id"], summary_text)

    def _generate_summary(self, records: list) -> str:
        """Use local LLM to summarize a batch of messages."""
        import httpx
        messages_text = "\n".join(f"[{r[1]}] {r[2][:300]}" for r in records[:5])

        prompt = f"""Summarize these conversation excerpts in 2-3 sentences.\nExtract key facts and decisions.\n\n{messages_text}\n\nSummary:"""

        # Call local Ollama for summarization
        resp = httpx.post(
            "http://localhost:11434/api/generate",
            json={"model": "qwen2.5:7b", "prompt": prompt, "stream": False},
            timeout=30.0
        )
        return resp.json()["response"].strip()

    def _save_summary(self, anchor_id: str, summary: str):
        import uuid
        new_id = str(uuid.uuid4())
        record = MemoryRecord(
            id=new_id,
            content=summary,
            role="system",
            timestamp=datetime.utcnow(),
            summary=summary,
            importance=0.7
        )
        self._persist(record)
        # Remove old records
        self.conn.execute("DELETE FROM memories WHERE id = ?", (anchor_id,))
        self.conn.commit()

    def _embed(self, text: str) -> np.ndarray:
        """Local embedding via Ollama (no network to external services)."""
        import httpx
        resp = httpx.post(
            "http://localhost:11434/api/embeddings",
            json={"model": "nomic-embed-text", "prompt": text},
            timeout=15.0
        )
        return np.array(resp.json()["embedding"])

    def _load_embed_model(self):
        """Ensure embedding model is available locally."""
        import httpx
        try:
            httpx.get("http://localhost:11434/api/tags", timeout=5.0)
        except httpx.ConnectError:
            raise RuntimeError(
                "Ollama not running. Start with: ollama pull nomic-embed-text && ollama serve"
            )
        return "loaded"

    def _chunk(self, items: list, size: int) -> list:
        return [items[i:i + size] for i in range(0, len(items), size)]

    def close(self):
        self.conn.close()

User Request
     │
     ▼
┌──────────────┐
│  Classifier  │  Lightweight model decides routing
│  (local,     │  Runs on smaller model (e.g. Phi-3)
│  ~0.5B params)│
└──────┬───────┘
       │
   ┌───┼───┐
   ▼   ▼   ▼
┌────┐┌────┐┌──────┐
│Tier││Tier││Tier  │
│  A ││  B ││  C   │
│Simple││Medium││Complex│
└────┘└────┘└──────┘
  │      │       │
  ▼      ▼       ▼
Local  Local   Cloud
Phi-3  Qwen   GPT-4o/
7B     14B     Claude
 cost   cost    premium
  $0     $0      $$$

# routing/cost_aware_router.py
import json
import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Optional


class ModelTier(Enum):
    FREE = "free"        # Local models, zero cost
    LOW = "low"          # Local models, minimal energy cost
    MEDIUM = "medium"    # Small cloud models (if needed)
    HIGH = "high"        # Premium cloud APIs


@dataclass
class RouteDecision:
    tier: ModelTier
    model: str
    estimated_cost_per_1k_tokens: float
    reason: str


class CostAwareRouter:
    """
    Routes LLM requests to the cheapest appropriate model.
    Uses a two-stage classification: keyword heuristic + lightweight
    model judgment, with caching for repeated patterns.
    """

    # Keyword-based tier assignments (fast path)
    SIMPLE_PATTERNS = [
        (r"hello|hi|hey", "greeting"),
        (r"thanks|thank you", "acknowledgment"),
        (r"what time|what's the time", "fact"),
        (r"translate (.+?) to", "translation"),
        (r"summarize(?:\s+this)?", "summarization"),
        (r"explain\s+(?:the\s+)?(?:basic|simple|what is)", "explanation"),
        (r"list|give me.*examples?", "enumeration"),
        (r"convert\s+(?:json|yaml|toml)", "format-conversion"),
    ]

    COMPLEX_PATTERNS = [
        (r"write\s+a\s+(?:full|complete|production)", "code-generation"),
        (r"debug|fix\s+(?:this\s+)?(?:error|bug|issue)", "debugging"),
        (r"architect|design\s+a\s+(?:system|api|architecture)", "architecture"),
        (r"analyze\s+(?:the\s+)?(?:code|architecture|system)", "analysis"),
        (r"create\s+a\s+(?:test|suite|benchmark)", "test-generation"),
        (r"compare|contrast\s+(?:these|the)", "comparison"),
        (r"optimize|improve\s+(?:this\s+)?(?:code|performance)", "optimization"),
        (r"review\s+(?:the\s+)?(?:code|PR|pull request)", "code-review"),
    ]

    # Model registry with costs (per 1M tokens input/output)
    MODEL_REGISTRY = {
        # Local models — effectively free
        "phi-3-mini": {"tier": ModelTier.FREE, "context_window": 4096, "input_cost": 0.0, "output_cost": 0.0},
        "qwen2.5:7b": {"tier": ModelTier.FREE, "context_window": 32768, "input_cost": 0.0, "output_cost": 0.0},
        "llama3.1:8b": {" tier": ModelTier.FREE, "context_window": 128000, "input_cost": 0.0, "output_cost": 0.0},
        "qwen2.5:14b": {"tier": ModelTier.LOW, "context_window": 32768, "input_cost": 0.0, "output_cost": 0.0},
        "command-r": {"tier": ModelTier.LOW, "context_window": 128000, "input_cost": 0.0, "output_cost": 0.0},

        # Cloud models with costs
        "claude-3-haiku": {"tier": ModelTier.MEDIUM, "context_window": 200000, "input_cost": 0.25, "output_cost": 1.25},
        "claude-3.5-sonnet": {"tier": ModelTier.HIGH, "context_window": 200000, "input_cost": 3.0, "output_cost": 15.0},
        "gpt-4o-mini": {"tier": ModelTier.MEDIUM, "context_window": 128000, "input_cost": 0.15, "output_cost": 0.60},
        "gpt-4o": {"tier": ModelTier.HIGH, "context_window": 128000, "input_cost": 2.50, "output_cost": 10.0},
    }

    def __init__(self, available_models: Optional[dict] = None):
        self.available = available_models or self.MODEL_REGISTRY.copy()
        self._cache: dict[str, RouteDecision] = {}
        self._cache_ttl = 300  # seconds

    def route(self, user_input: str, system_context: str = "", max_tokens_budget: int = 4096) -> RouteDecision:
        """
        Determine the optimal model for a given request.
        Two-stage: heuristic classification → model selection.
        """
        cache_key = f"{hash(user_input[:100])}:{max_tokens_budget}"
        if cache_key in self._cache:
            cached = self._cache[cache_key]
            if (asyncio.get_event_loop().time() - cached._cached_time) < self._cache_ttl:
                return cached.decision

        # Stage 1: Keyword heuristic classification
        task_type = self._classify_task(user_input)

        # Stage 2: Select model based on task type + budget
        decision = self._select_model(task_type, max_tokens_budget, user_input)

        self._cache[cache_key] = _CacheEntry(decision, asyncio.get_event_loop().time())
        return decision

    def _classify_task(self, text: str) -> str:
        """Classify the task type using pattern matching."""
        text_lower = text.lower()

        for pattern, task_type in self.COMPLEX_PATTERNS:
            if __import__("re").search(pattern, text_lower):
                return task_type

        for pattern, task_type in self.SIMPLE_PATTERNS:
            if __import__("re").search(pattern, text_lower):
                return task_type

        return "general"  # Default: try local first

    def _select_model(self, task_type: str, max_tokens: int, raw_input: str) -> RouteDecision:
        """
        Select the cheapest model that can handle the task.
        Strategy: try local first, escalate only when necessary.
        """
        # Prefer local models for known simple tasks
        if task_type in ("greeting", "acknowledgment", "fact"):
            model = self._find_local_with_context(4096)
            return RouteDecision(
                tier=ModelTier.FREE, model=model,
                estimated_cost_per_1k_tokens=0.0,
                reason=f"Simple {task_type} — local model sufficient"
            )

        # For complex tasks, try local first, escalate on failure
        preferred_local = self._find_best_local(max_tokens)

        if preferred_local:
            # Test local model capability with a lightweight probe
            if self._is_local_capable(preferred_local, task_type, raw_input):
                return RouteDecision(
                    tier=ModelTier.FREE,
                    model=preferred_local,
                    estimated_cost_per_1k_tokens=0.0,
                    reason=f"Local model {preferred_local} can handle {task_type}"
                )

        # Fall back to cloud — pick cheapest capable model
        cloud_model = self._find_cheapest_cloud(task_type, max_tokens)
        if cloud_model:
            cfg = self.available[cloud_model]
            return RouteDecision(
                tier=cfg["tier"],
                model=cloud_model,
                estimated_cost_per_1k_tokens=(cfg["input_cost"] + cfg["output_cost"]) / 2,
                reason=f"Escalated to cloud for {task_type} (local insufficient)"
            )

        raise RuntimeError(f"No suitable model found for task type: {task_type}")

    def _find_local_with_context(self, min_ctx: int) -> Optional[str]:
        """Find any local model with sufficient context window."""
        for name, cfg in self.available.items():
            if cfg["tier"] in (ModelTier.FREE, ModelTier.LOW) and cfg["context_window"] >= min_ctx:
                return name
        return None

    def _find_best_local(self, max_tokens: int) -> Optional[str]:
        """Find the most capable local model with enough context."""
        candidates = [
            (name, cfg)
            for name, cfg in self.available.items()
            if cfg["tier"] in (ModelTier.FREE, ModelTier.LOW)
            and cfg["context_window"] >= max_tokens
        ]
        # Prefer larger, more capable models first
        candidates.sort(key=lambda x: x[1]["context_window"], reverse=True)
        return candidates[0][0] if candidates else None

    def _find_cheapest_cloud(self, task_type: str, max_tokens: int) -> Optional[str]:
        """Find cheapest cloud model that can handle the task."""
        candidates = [
            (name, cfg)
            for name, cfg in self.available.items()
            if cfg["tier"] in (ModelTier.MEDIUM, ModelTier.HIGH)
            and cfg["context_window"] >= max_tokens
        ]
        candidates.sort(key=lambda x: x[1]["input_cost"])
        return candidates[0][0] if candidates else None

    def _is_local_capable(self, model: str, task_type: str, input_text: str) -> bool:
        """
        Determine if the local model is likely capable of this task.
        Uses a combination of model size heuristics and task complexity.
        """
        # Larger local models handle more complex tasks
        model_params = self._estimate_params(model)

        complex_tasks = {"architecture", "code-generation", "debugging", "code-review"}
        if task_type in complex_tasks and model_params < 8:
            return False  # Small model, complex task — escalate

        # Check input length — very long inputs may exceed local context
        if len(input_text) > 5000 and model_params < 14:
            return False

        return True  # Default: trust local for most things

    def _estimate_params(self, model_name: str) -> int:
        """Rough estimate of model parameter count from name."""
        import re
        match = re.search(r'(\d+)\.?(\d*)b?', model_name)
        if match:
            base = int(match.group(1))
            frac = int(match.group(2)) if match.group(2) else 0
            return base + frac / 10
        return 7  # Default assumption

    def get_cost_estimate(self, decision: RouteDecision, input_tokens: int, output_tokens: int) -> float:
        """Calculate estimated cost for a route decision."""
        cfg = self.available.get(decision.model, {})
        input_cost = cfg.get("input_cost", 0) / 1_000_000 * input_tokens
        output_cost = cfg.get("output_cost", 0) / 1_000_000 * output_tokens
        return round(input_cost + output_cost, 6)


class _CacheEntry:
    def __init__(self, decision: RouteDecision, timestamp: float):
        self.decision = decision
        self._cached_time = timestamp

# monitoring/metrics.py
import json
from datetime import datetime
from pathlib import Path
from collections import defaultdict


class CostTracker:
    """Track routing decisions and costs over time."""

    def __init__(self, log_path: Path = Path("~/.localfirst/routing_log.jsonl")):
        self.log_path = log_path.expanduser()
        self._stats = defaultdict(lambda: {"count": 0, "total_cost": 0.0, "by_tier": defaultdict(int)})

    def log(self, decision, input_tokens: int, output_tokens: int, success: bool = True):
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "model": decision.model,
            "tier": decision.tier.value,
            "reason": decision.reason,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": decision.estimated_cost_per_1k_tokens * (input_tokens + output_tokens) / 1000,
            "success": success
        }
        with open(self.log_path, "a") as f:
            f.write(json.dumps(record) + "\n")

        self._stats[decision.model]["count"] += 1
        self._stats[decision.model]["total_cost"] += record["cost"]
        self._stats[decision.model]["by_tier"][decision.tier.value] += 1

    def report(self) -> dict:
        """Generate a summary report."""
        total_requests = sum(s["count"] for s in self._stats.values())
        total_cost = sum(s["total_cost"] for s in self._stats.values())

        tier_breakdown = defaultdict(int)
        for stats in self._stats.values():
            for tier, count in stats["by_tier"].items():
                tier_breakdown[tier] += count

        return {
            "total_requests": total_requests,
            "total_cost_usd": round(total_cost, 4),
            "avg_cost_per_request": round(total_cost / max(total_requests, 1), 6),
            "tier_distribution": dict(tier_breakdown),
            "models_used": {
                model: {
                    "requests": stats["count"],
                    "total_cost": round(stats["total_cost"], 4)
                }
                for model, stats in self._stats.items()
            }
        }

┌─────────────────────────────────────────────────────────────┐
│                      USER INTERFACE                         │
│  (Web, CLI, Desktop — any frontend)                         │
└─────────────────────────────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                    APP ORCHESTRATOR                         │
│  ┌─────────────┐  ┌─────────────┐  ┌──────────────────┐   │
│  │  Memory     │  │  Cost       │  │  MCP             │   │
│  │  Manager    │  │  Router     │  │  Client          │   │
│  │             │  │             │  │                  │   │
│  │ • Add ctx   │  │ • Classify  │  │ • Tool calls     │   │
│  │ • Recall    │  │ • Route     │  │ • Resource read  │   │
│  │ • Summarize │  │ • Estimate  │  │ • Prompt templates│   │
│  └─────────────┘  └─────────────┘  └──────────────────┘   │
└─────────────────────────────────────────────────────────────┘
          │              │              │
    ┌─────▼──────┐  ┌────▼─────┐  ┌───▼─────────┐
    │  Local     │  │ Local    │  │ MCP Servers  │
    │  Memory    │  │ LLM      │  │ (stdio/SSE)  │
    │  Store     │  │ (Ollama) │  │              │
    │  (SQLite+  │  │          │  │ • Filesystem │
    │   sqlite_  │  │  Primary │  │ • Search     │
    │   vec)     │  │  Compute │  │ • Database   │
    │            │  │          │  │ • Knowledge  │
    │            │  │  Fallback│  │   Base       │
    │            │  │  (cloud) │  └──────────────┘
    │            │  └──────────┘
    └────────────┘

# app/orchestrator.py
import asyncio
from pathlib import Path
from typing import Optional

from memory.store import LocalMemoryStore
from routing.cost_aware_router import CostAwareRouter, ModelTier
from monitoring.metrics import CostTracker


class LocalFirstApp:
    """
    Complete local-first AI application.
    All operations work offline. Cloud is fallback only.
    """

    def __init__(self, data_dir: Path = Path("~/.localfirst")):
        self.data_dir = data_dir.expanduser()
        self.data_dir.mkdir(parents=True, exist_ok=True)

        self.memory = LocalMemoryStore(self.data_dir / "memory.db")
        self.router = CostAwareRouter()
        self.tracker = CostTracker(self.data_dir / "routing_log.jsonl")
        self.session_id = "default"

    async def chat(self, user_input: str) -> dict:
        """
        Process a user message through the full local-first pipeline.
        Returns response + metadata about routing decision.
        """
        # 1. Build context from memory
        context = self.memory.get_context(user_input, max_tokens=1500)

        # 2. Classify and route
        decision = self.router.route(user_input, system_context=context)

        # 3. Build the prompt
        prompt = self._build_prompt(user_input, context)

        # 4. Execute via chosen model
        result = await self._execute_with_model(prompt, decision)

        # 5. Store the exchange in memory
        self.memory.add("user", user_input, self.session_id)
        self.memory.add("assistant", result["response"], self.session_id)

        # 6. Track costs
        self.tracker.log(decision, result["input_tokens"], result["output_tokens"])

        return {
            "response": result["response"],
            "routing": {
                "model": decision.model,
                "tier": decision.tier.value,
                "reason": decision.reason,
                "estimated_cost": self.tracker.get_cost_estimate(
                    decision, result["input_tokens"], result["output_tokens"]
                )
            }
        }

    def _build_prompt(self, user_input: str, context: str) -> str:
        """Construct the full prompt with context injection."""
        system_prompt = """You are a helpful AI assistant running in a local-first environment.
You have access to tools (memory, file system, search) via MCP.
Be concise, accurate, and respect the user's privacy — all data stays local."""

        parts = [f"System: {system_prompt}"]
        if context:
            parts.append(f"\n{context}")
        parts.append(f"\nUser: {user_input}")
        parts.append("\nAssistant:")

        return "\n".join(parts)

    async def _execute_with_model(self, prompt: str, decision) -> dict:
        """Execute the prompt using the routed model."""
        import httpx

        if decision.tier in (ModelTier.FREE, ModelTier.LOW):
            # Local execution via Ollama
            async with httpx.AsyncClient(timeout=120.0) as client:
                resp = await client.post(
                    "http://localhost:11434/api/generate",
                    json={
                        "model": decision.model,
                        "prompt": prompt,
                        "stream": False,
                        "options": {"num_ctx": 4096}
                    }
                )
                data = resp.json()
                return {
                    "response": data["response"],
                    "input_tokens": data.get("prompt_eval_count", 0),
                    "output_tokens": data.get("eval_count", 0)
                }
        else:
            # Cloud fallback
            async with httpx.AsyncClient(timeout=60.0) as client:
                # This would call OpenAI/Anthropic in production
                # For now, simulate with local as fallback
                return await self._execute_with_model(prompt, decision)

    def run_periodic_maintenance(self):
        """Run memory compaction and cache cleanup."""
        self.memory.summarize_old_sessions(older_than_days=14)
        # Clear stale routing cache
        self.router._cache.clear()
        print(f"Maintenance complete. Costs so far: {self.tracker.report()}")

    def close(self):
        self.memory.close()


async def main():
    app = LocalFirstApp()

    print("Local-First AI App (type 'quit' to exit)\n")
    while True:
        try:
            user_input = input("> ").strip()
            if user_input.lower() in ("quit", "exit"):
                break
            if not user_input:
                continue

            result = await app.chat(user_input)
            print(f"\n{result['response']}\n")
            print(f"[Routed: {result['routing']['model']} ({result['routing']['tier']})]")

        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"Error: {e}")

    app.run_periodic_maintenance()
    app.close()
    print(f"\nFinal report: {app.tracker.report()}")


if __name__ == "__main__":
    asyncio.run(main())

# docker-compose.yml
version: "3.9"
services:
  ollama:
    image: ollama/ollama:latest
    container_mode: true
    ports:
      - "11434:11434"
    volumes:
      - ollama_models:/root/.ollama
    command: ["serve"]

  # MCP servers as separate containers (optional for isolation)
  # Uncomment for production-grade isolation
  # mcp-files:
  #   build: ./mcp-servers/files
  #   volumes:
  #     - ./data:/data
  #   environment:
  #     - DATA_DIR=/data

volumes:
  ollama_models: