• A specific AI product (ChatGPT, Perplexity, Claude, Cursor, etc.)
• A SaaS platform (Notion, Linear, Vercel, etc.)
• A workflow or agent architecture you've described or screenshotted
• A codebase you want replicated or rebuilt
• A business system or process

• Four meta-prompt frameworks
• Zero target systems
• Zero links, screenshots, files, descriptions, names, or behavioral examples

• "Orris" (this AI assistant)
• "Perplexity AI"
• "A RAG-based medical Q&A chatbot"
• "My app that does X" + paste a description
• A URL, screenshot, or spec

ORRIS - COMPLETE SYSTEM GENOME & PRODUCTION RECONSTRUCTION BLUEPRINT

---

1. PRODUCT IDENTITY

• Medical queries ALWAYS route through the textbook library skill before any response
• Tool calls are parallelized by default - never sequential when independence allows
• No hallucination tolerance - explicit "I don't know" over plausible fabrication
• Budget-aware execution - tool calls are metered and tracked per session
• Clarification is mandatory before ambiguous significant actions, optional for trivial ones
• Output is direct, citation-backed, no filler language

---

2. SYSTEM ARCHITECTURE

┌─────────────────────────────────────────────────────────────┐
│                        USER INTERFACE                        │
│              (Chat UI + File Upload + Streaming)             │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                    GATEWAY / API LAYER                       │
│         (Auth, Rate Limiting, Budget Enforcement)            │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                  ORCHESTRATION ENGINE                        │
│  ┌─────────────┐  ┌──────────────┐  ┌────────────────────┐  │
│  │ Query Router│  │ Tool Planner │  │ Skill Dispatcher   │  │
│  │ (medical vs │  │ (parallel vs │  │ (reads SKILL.md,   │  │
│  │  general)   │  │  sequential) │  │  routes to tools)  │  │
│  └─────────────┘  └──────────────┘  └────────────────────┘  │
└──┬──────────┬───────────┬────────────┬────────────┬─────────┘
   │          │           │            │            │
┌──▼──┐  ┌───▼───┐  ┌────▼────┐  ┌───▼───┐  ┌────▼────────┐
│LLM  │  │ RAG   │  │PubMed   │  │ Web   │  │  Sandbox    │
│Layer│  │Engine │  │API      │  │Search │  │  Execution  │
│     │  │(SQLite│  │         │  │Engine │  │  (Daytona)  │
│     │  │FTS5)  │  │         │  │       │  │             │
└──┬──┘  └───┬───┘  └────┬────┘  └───┬───┘  └────┬────────┘
   │         │           │           │            │
┌──▼─────────▼───────────▼───────────▼────────────▼─────────┐
│                   RESPONSE ASSEMBLER                        │
│         (Citation injection, streaming, formatting)         │
└─────────────────────────────────────────────────────────────┘

---

3. TECH STACK

Inference Layer

Backend

Data Layer

Frontend

Orchestration

Infrastructure

---

4. AI ORCHESTRATION ENGINE

4.1 Query Classification Router

# router.py
MEDICAL_KEYWORDS = {
    "anatomy", "physiology", "pathology", "pharmacology", "diagnosis",
    "treatment", "symptom", "medication", "drug", "clinical", "patient",
    "nursing", "surgery", "imaging", "lab", "biomedical", "genetic",
    "epidemiology", "biochemistry", "procedure", "dose", "contraindication"
}

def classify_query(query: str) -> QueryClass:
    tokens = set(query.lower().split())
    if tokens & MEDICAL_KEYWORDS:
        return QueryClass.MEDICAL  # mandatory skill routing
    if any(kw in query.lower() for kw in ["pptx", "docx", "pdf", "xlsx", "spreadsheet"]):
        return QueryClass.ARTIFACT_GENERATION
    if "?" in query and len(query) < 200:
        return QueryClass.CLARIFICATION_CANDIDATE
    return QueryClass.GENERAL

4.2 System Prompt (Master)

You are Orris, a knowledgeable AI assistant that helps with questions across all domains.

MEDICAL-FIRST ROUTING (MANDATORY):
If the query contains ANY medical/clinical/health terminology, you MUST:
1. Invoke /skills/readonly/medical-library/SKILL.md as your FIRST action
2. Do NOT answer from training knowledge before skill invocation
3. Use query_library_db with FTS5 search before any other tool

TOOL CALL STRATEGY:
- Parallelize all independent tool calls in a single response block
- Run web_search on EVERY turn alongside specialized tools
- Sequential calls only when result of one determines arguments of next

RESPONSE RULES:
- No em dashes (—), en dashes (–)
- No banned phrases: [delve, crucial, vital, leverage, robust, comprehensive,
  holistic, synergy, innovative, transformative, seamless, empower, groundbreaking,
  game-changing, revolutionize, cutting-edge, paradigm, certainly, absolutely,
  of course, it's worth noting, in conclusion, as an AI]
- Citations inline: (Book Title, p. XXX)
- Links inline, never standalone
- Direct, natural tone - no preamble, no emotional validation

CLARIFICATION POLICY:
- MANDATORY before broad/vague/multi-interpretation queries
- Use ask_user_input tool before significant ambiguous actions
- Skip for direct factual lookups

BUDGET AWARENESS:
- Track tool call usage against session budget
- Parallelize to maximize quality per model call
- Never exceed stated limits

4.3 Tool Selection Logic

# tool_planner.py
class ToolPlanner:
    def plan(self, query: str, query_class: QueryClass) -> List[ToolCall]:
        calls = []

        # ALWAYS add web_search unless purely casual
        if not self.is_purely_casual(query):
            calls.append(ToolCall("web_search", query=query))

        if query_class == QueryClass.MEDICAL:
            # Medical-first: library lookup is BLOCKING, runs before LLM response
            calls = [ToolCall("skill_invoke", skill="medical-library")]
            # web_search runs in parallel AFTER skill dispatch
            calls.append(ToolCall("web_search", query=query))

        elif query_class == QueryClass.ARTIFACT_GENERATION:
            calls.append(ToolCall("skill_invoke", skill=self.detect_artifact_skill(query)))

        # PubMed supplement for medical queries
        if query_class == QueryClass.MEDICAL:
            calls.append(ToolCall("pubmed_search", query=self.extract_mesh_terms(query),
                                  date_range_years=2,
                                  pub_types=["Meta-Analysis", "Systematic Review"]))

        return self.deduplicate_parallelize(calls)

4.4 Skill Dispatcher

# skill_dispatcher.py
SKILL_REGISTRY = {
    "medical-library": "/skills/readonly/medical-library/SKILL.md",
    "pubmed":          "/skills/readonly/pubmed/SKILL.md",
    "docx":            "/home/daytona/skills/docx/SKILL.md",
    "pdf":             "/home/daytona/skills/pdf/SKILL.md",
    "pptx":            "/home/daytona/skills/pptx/SKILL.md",
    "xlsx":            "/home/daytona/skills/xlsx/SKILL.md",
}

class SkillDispatcher:
    def invoke(self, skill_name: str) -> str:
        path = SKILL_REGISTRY[skill_name]
        # read_file with no limit - full SKILL.md always required
        return read_file(path, limit=None)

    def detect_from_query(self, query: str) -> Optional[str]:
        triggers = {
            "docx":  ["word doc", ".docx", "word document", "report", "memo", "letter"],
            "pdf":   [".pdf", "pdf"],
            "pptx":  ["slides", "presentation", ".pptx", "deck", "powerpoint"],
            "xlsx":  ["spreadsheet", ".xlsx", "excel", "csv", "tabular"],
        }
        for skill, keywords in triggers.items():
            if any(k in query.lower() for k in keywords):
                return skill
        return None

4.5 Reflection + Self-Correction Loop

# reflector.py
class ResponseReflector:
    BANNED_PHRASES = [
        "delve", "crucial", "vital", "leverage", "robust", "comprehensive",
        "holistic", "synergy", "innovative", "transformative", "seamless",
        "empower", "groundbreaking", "game-changing", "revolutionize",
        "cutting-edge", "paradigm", "certainly", "absolutely", "of course",
        "it's worth noting", "in conclusion", "as an AI"
    ]
    EM_DASH_PATTERN = re.compile(r'[—–]')

    def validate(self, response: str) -> ValidationResult:
        issues = []
        for phrase in self.BANNED_PHRASES:
            if phrase.lower() in response.lower():
                issues.append(f"Banned phrase detected: '{phrase}'")
        if self.EM_DASH_PATTERN.search(response):
            issues.append("Em/en dash detected - replace with hyphen")
        if "I'll now" in response or "Sure!" in response or "Great question" in response:
            issues.append("Preamble filler detected")
        return ValidationResult(valid=len(issues)==0, issues=issues)

    def correct(self, response: str) -> str:
        response = self.EM_DASH_PATTERN.sub('-', response)
        # Re-route to LLM with correction prompt if banned phrases detected
        if any(p.lower() in response.lower() for p in self.BANNED_PHRASES):
            return self.llm_rewrite(response, instructions="Remove all banned phrases, maintain meaning")
        return response

---

5. RAG + RETRIEVAL SYSTEM

5.1 Medical Textbook Database Architecture

-- Core tables
CREATE TABLE books (
    isbn TEXT PRIMARY KEY,
    title TEXT,
    speciality TEXT,
    total_blocks INTEGER,
    file_path TEXT
);

CREATE TABLE blocks (
    id INTEGER PRIMARY KEY,
    book_isbn TEXT REFERENCES books(isbn),
    block_number INTEGER,
    total_blocks INTEGER,
    file_path TEXT,  -- path to the actual text block file
    size_bytes INTEGER,
    line_count INTEGER,
    heading_count INTEGER
);

CREATE TABLE headings (
    id INTEGER PRIMARY KEY,
    block_id INTEGER REFERENCES blocks(id),
    book_isbn TEXT,
    level INTEGER,
    title TEXT,
    line_number INTEGER
);

CREATE TABLE paragraphs (
    id INTEGER PRIMARY KEY,
    block_id INTEGER REFERENCES blocks(id),
    book_isbn TEXT,
    text TEXT,
    line_start INTEGER,
    line_end INTEGER,
    section_heading TEXT
);

-- FTS5 virtual tables
CREATE VIRTUAL TABLE headings_fts USING fts5(
    title, book_isbn, book_title, speciality,
    block_file_path, line_number, para_line_ranges, para_count,
    content='headings'
);

CREATE VIRTUAL TABLE paragraphs_fts USING fts5(
    text, book_isbn, book_title, speciality,
    block_file_path, line_start, line_end, section_heading,
    content='paragraphs'
);

-- Key views
CREATE VIEW v_book_overview AS
    SELECT isbn, title, speciality, total_blocks, file_path FROM books;

CREATE VIEW v_paragraphs AS
    SELECT p.id, b2.speciality, b2.title as book_title, b2.isbn as book_isbn,
           p.section_heading, p.text, p.line_start, p.line_end,
           bl.block_number, bl.file_path as block_file_path
    FROM paragraphs p
    JOIN blocks bl ON p.block_id = bl.id
    JOIN books b2 ON bl.book_isbn = b2.isbn;

5.2 Retrieval Pipeline

# retrieval_pipeline.py
class MedicalLibraryRetriever:
    def __init__(self, db_path: str, textbook_base: str = "/textbooks/"):
        self.db = sqlite3.connect(db_path)
        self.base = textbook_base

    def search_headings(self, query: str, limit: int = 10) -> List[HeadingResult]:
        """Stage 1: Find relevant sections by heading"""
        sql = """
            SELECT title, book_title, speciality,
                   '/textbooks/' || block_file_path AS file_path,
                   line_number, para_line_ranges, para_count
            FROM headings_fts
            WHERE headings_fts MATCH ?
            ORDER BY rank
            LIMIT ?
        """
        return self.db.execute(sql, [query, limit]).fetchall()

    def search_paragraphs(self, query: str, limit: int = 10) -> List[ParagraphResult]:
        """Stage 2: Deep content search"""
        sql = """
            SELECT book_title, section_heading,
                   '/textbooks/' || block_file_path AS file_path,
                   line_start, line_end,
                   snippet(paragraphs_fts, 0, '>>>', '<<<', '...', 30) as snippet
            FROM paragraphs_fts
            WHERE paragraphs_fts MATCH ?
            ORDER BY rank
            LIMIT ?
        """
        return self.db.execute(sql, [query, limit]).fetchall()

    def read_block(self, file_path: str, offset: int = 0, limit: int = 100) -> str:
        """Read actual text content from block file"""
        with open(file_path, 'r') as f:
            lines = f.readlines()
        return ''.join(lines[offset:offset+limit])

    def retrieve(self, query: str) -> RetrievalResult:
        """Full pipeline: heading search -> paragraph search -> block read"""
        headings = self.search_headings(query)
        paragraphs = self.search_paragraphs(query)
        # Read content from top results
        content_blocks = []
        seen_files = set()
        for h in headings[:3]:
            if h.file_path not in seen_files:
                content = self.read_block(h.file_path, offset=h.line_number, limit=100)
                content_blocks.append(ContentBlock(source=h, content=content))
                seen_files.add(h.file_path)
        return RetrievalResult(headings=headings, paragraphs=paragraphs, blocks=content_blocks)

5.3 FTS5 Query Construction

# fts_query_builder.py
class FTS5QueryBuilder:
    """Handles FTS5 syntax requirements"""

    def build(self, user_query: str) -> str:
        tokens = user_query.strip().split()
        processed = []
        for token in tokens:
            # Hyphenated terms MUST be double-quoted for FTS5
            if '-' in token:
                processed.append(f'"{token}"')
            else:
                processed.append(token)
        # Multi-word: phrase search
        if len(processed) > 1:
            return f'"{" ".join(processed)}"'
        return processed[0] if processed else user_query

    def build_boolean(self, terms: List[str], operator: str = "AND") -> str:
        return f" {operator} ".join(terms)

    def build_prefix(self, term: str) -> str:
        return f"{term}*"

5.4 PubMed Integration

# pubmed_client.py
class PubMedClient:
    BASE_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"

    def search(self, query: str, max_results: int = 10,
               pub_types: List[str] = None,
               date_range_years: int = None) -> List[PubMedResult]:
        params = {
            "db": "pubmed",
            "term": self._build_query(query, pub_types),
            "retmax": max_results,
            "sort": "relevance",
            "usehistory": "y"
        }
        if date_range_years:
            params["datetype"] = "pdat"
            params["reldate"] = date_range_years * 365

        response = requests.get(f"{self.BASE_URL}/esearch.fcgi", params=params)
        pmids = self._parse_pmids(response)
        return self.fetch_details(pmids)

    def _build_query(self, query: str, pub_types: List[str]) -> str:
        q = query
        if pub_types:
            pt_filter = " OR ".join([f'"{pt}"[pt]' for pt in pub_types])
            q += f" AND ({pt_filter})"
        return q

    def rank_by_evidence_tier(self, results: List[PubMedResult]) -> List[PubMedResult]:
        TIER_ORDER = {
            "Systematic Review": 1,
            "Meta-Analysis": 2,
            "Randomized Controlled Trial": 3,
            "Clinical Trial": 4,
            "Review": 5,
            "Practice Guideline": 6,
            "Case Reports": 7
        }
        return sorted(results, key=lambda r: TIER_ORDER.get(r.pub_type, 99))

---

6. MEMORY + CONTEXT SYSTEM

6.1 Session Memory Architecture

# memory_manager.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import redis
import json

@dataclass
class Turn:
    role: str  # "user" | "assistant"
    content: str
    tool_calls: List[dict] = field(default_factory=list)
    tool_results: List[dict] = field(default_factory=list)
    timestamp: float = 0.0

class SessionMemoryManager:
    """
    Orris uses NO persistent cross-session user memory by default.
    Memory is per-conversation only, held in context window.
    Redis handles hot session state for active conversations.
    """

    MAX_CONTEXT_TURNS = 40       # hard cap before summarization
    SUMMARIZE_THRESHOLD = 30     # trigger summarization
    MAX_TOOL_RESULT_TOKENS = 2000  # truncate large tool outputs

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def get_session(self, session_id: str) -> List[Turn]:
        raw = self.redis.get(f"session:{session_id}")
        if not raw:
            return []
        return [Turn(**t) for t in json.loads(raw)]

    def append_turn(self, session_id: str, turn: Turn):
        turns = self.get_session(session_id)
        turns.append(turn)
        if len(turns) >= self.SUMMARIZE_THRESHOLD:
            turns = self._summarize_oldest(turns)
        self.redis.setex(
            f"session:{session_id}",
            3600,  # 1 hour TTL
            json.dumps([t.__dict__ for t in turns])
        )

    def _summarize_oldest(self, turns: List[Turn]) -> List[Turn]:
        """Summarize oldest 20 turns into a single context block"""
        oldest = turns[:20]
        summary_text = self._call_summarizer(oldest)
        summary_turn = Turn(
            role="system",
            content=f"[CONVERSATION SUMMARY - earlier turns compressed]\n{summary_text}"
        )
        return [summary_turn] + turns[20:]

    def truncate_tool_results(self, tool_result: str) -> str:
        """Prevent tool results from blowing context window"""
        tokens = estimate_tokens(tool_result)
        if tokens > self.MAX_TOOL_RESULT_TOKENS:
            # Save to file, return path reference
            path = self._save_to_large_results(tool_result)
            return f"[Result too large - saved to {path}. Use read_file with pagination.]"
        return tool_result

6.2 Budget Tracking System

# budget_tracker.py
@dataclass
class SessionBudget:
    model_calls: int = 0
    model_calls_limit: int = 19
    tool_calls: Dict[str, int] = field(default_factory=dict)
    tool_limits: Dict[str, int] = field(default_factory=lambda: {
        "task": 2,
        "query_library_db": 10,
        "fetch_image": 10,
        "pubmed_search": 10,
        "pubmed_detail": 10,
        "pubmed_fulltext": 5,
        "web_search": 5,
        "web_extract": 3,
    })

class BudgetEnforcer:
    def check(self, budget: SessionBudget, tool_name: str) -> bool:
        used = budget.tool_calls.get(tool_name, 0)
        limit = budget.tool_limits.get(tool_name, 999)
        return used < limit

    def consume(self, budget: SessionBudget, tool_name: str):
        budget.tool_calls[tool_name] = budget.tool_calls.get(tool_name, 0) + 1

    def inject_budget_context(self, budget: SessionBudget) -> str:
        """Inject current budget state into system prompt"""
        lines = [f"Current date: {datetime.now().strftime('%B %d, %Y')}"]
        lines.append(f"Model calls: {budget.model_calls}/{budget.model_calls_limit} used")
        tool_summary = " | ".join([
            f"{tool}: {budget.tool_calls.get(tool,0)}/{limit}"
            for tool, limit in budget.tool_limits.items()
        ])
        lines.append(f"Tool calls: {tool_summary}")
        return "\n".join(lines)

---

7. BACKEND SERVICES + APIs

7.1 Service Architecture

orris-backend/
├── api/
│   ├── chat.py           # POST /chat/stream (SSE)
│   ├── tools.py          # POST /tools/execute
│   ├── sessions.py       # GET/DELETE /sessions/{id}
│   ├── files.py          # POST /files/upload, GET /files/{id}
│   └── health.py         # GET /health, GET /metrics
├── orchestration/
│   ├── graph.py          # LangGraph workflow definition
│   ├── router.py         # Query classification
│   ├── tool_planner.py   # Parallel tool scheduling
│   └── skill_dispatcher.py
├── retrieval/
│   ├── medical_library.py
│   ├── pubmed_client.py
│   ├── web_search.py
│   └── fts_query_builder.py
├── memory/
│   ├── session_manager.py
│   ├── budget_tracker.py
│   └── summarizer.py
├── execution/
│   ├── sandbox.py        # Daytona/Docker sandbox interface
│   ├── file_generators.py # PPTX, DOCX, PDF, XLSX generators
│   └── image_generator.py
├── skills/
│   ├── loader.py         # Reads SKILL.md files
│   └── registry.py       # Skill name -> path mapping
└── models/
    ├── turn.py
    ├── budget.py
    └── retrieval.py

7.2 Core Chat Endpoint

# api/chat.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """
    Main entry point. Returns SSE stream of tokens + tool events.
    """
    session = await session_manager.get_or_create(request.session_id)
    budget = await budget_tracker.get(request.session_id)

    async def event_generator():
        async for event in orchestration_engine.run(
            query=request.message,
            session=session,
            budget=budget,
            attachments=request.attachments
        ):
            if event.type == "token":
                yield {"event": "token", "data": event.content}
            elif event.type == "tool_start":
                yield {"event": "tool_start", "data": json.dumps({
                    "tool": event.tool_name,
                    "args": event.args
                })}
            elif event.type == "tool_result":
                yield {"event": "tool_result", "data": json.dumps({
                    "tool": event.tool_name,
                    "result_preview": event.result[:200]
                })}
            elif event.type == "done":
                yield {"event": "done", "data": json.dumps({
                    "budget_remaining": budget.summary()
                })}

    return EventSourceResponse(event_generator())

7.3 Orchestration Graph (LangGraph)

# orchestration/graph.py
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

def build_orris_graph():
    graph = StateGraph(OrchestratorState)

    # Nodes
    graph.add_node("classify", classify_query_node)
    graph.add_node("check_clarification", clarification_check_node)
    graph.add_node("read_skill", read_skill_node)        # blocking for medical
    graph.add_node("plan_tools", tool_planning_node)
    graph.add_node("execute_tools", ToolNode(ALL_TOOLS)) # parallel execution
    graph.add_node("assemble_response", response_assembly_node)
    graph.add_node("validate_response", reflection_validation_node)
    graph.add_node("stream_output", streaming_output_node)

    # Edges
    graph.set_entry_point("classify")
    graph.add_conditional_edges("classify", route_by_class, {
        "medical": "read_skill",
        "artifact": "read_skill",
        "clarification_needed": "check_clarification",
        "general": "plan_tools"
    })
    graph.add_edge("read_skill", "plan_tools")
    graph.add_edge("check_clarification", "plan_tools")
    graph.add_edge("plan_tools", "execute_tools")
    graph.add_edge("execute_tools", "assemble_response")
    graph.add_edge("assemble_response", "validate_response")
    graph.add_conditional_edges("validate_response", check_validity, {
        "valid": "stream_output",
        "needs_correction": "assemble_response"  # self-correction loop, max 2 iterations
    })
    graph.add_edge("stream_output", END)

    return graph.compile()

---

8. FRONTEND ARCHITECTURE

8.1 Component Tree

app/
├── layout.tsx              # Root layout, auth provider, theme
├── page.tsx                # Landing / redirect to /chat
└── chat/
    ├── page.tsx            # Main chat view
    └── [sessionId]/
        └── page.tsx        # Session-specific chat

components/
├── chat/
│   ├── ChatWindow.tsx      # Message list + scroll management
│   ├── MessageBubble.tsx   # User/assistant message renderer
│   ├── ToolCallCard.tsx    # Tool execution visibility panel
│   ├── CitationInline.tsx  # Inline citation chip [Book, p.X]
│   ├── StreamingCursor.tsx # Animated streaming indicator
│   └── InputBar.tsx        # Textarea + file upload + send
├── evidence/
│   ├── EvidencePanel.tsx   # Slide-in panel: citations, sources
│   ├── PubMedCard.tsx      # Paper card with PMID, evidence tier
│   └── TextbookSnippet.tsx # Textbook passage preview
├── artifacts/
│   ├── FileDownloadCard.tsx # Download link for generated files
│   └── ImagePreview.tsx    # Inline generated images
└── ui/
    ├── BudgetIndicator.tsx  # Remaining tool calls display
    └── SessionList.tsx      # Sidebar: past sessions

8.2 Streaming Logic

// hooks/useOrrisChat.ts
import { useChat } from 'ai/react';

export function useOrrisChat(sessionId: string) {
  const { messages, append, isLoading, data } = useChat({
    api: '/api/chat/stream',
    id: sessionId,
    body: { session_id: sessionId },
    onToolCall: ({ toolCall }) => {
      // Surface tool execution in UI
      addToolEvent(toolCall.toolName, toolCall.args);
    },
  });

  const sendMessage = async (content: string, attachments?: File[]) => {
    const formData = await buildFormData(content, attachments);
    await append({ role: 'user', content }, { body: formData });
  };

  return { messages, sendMessage, isLoading };
}

8.3 Citation Rendering

// components/chat/MessageBubble.tsx
function renderWithCitations(text: string, citations: Citation[]) {
  // Parse inline citations like (Book Title, p. 123)
  const CITATION_REGEX = /\(([^)]+),\s*p\.\s*(\d+)\)/g;
  return text.replace(CITATION_REGEX, (match, book, page) => {
    const citation = citations.find(c => c.book === book && c.page === parseInt(page));
    return citation
      ? `<cite data-id="${citation.id}" class="citation-chip">${match}</cite>`
      : match;
  });
}

---

9. DEPLOYMENT INFRASTRUCTURE

9.1 Docker Compose (Development)

# docker-compose.yml
version: '3.9'
services:
  backend:
    build: ./backend
    ports: ["8000:8000"]
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379
      - POSTGRES_URL=postgresql://orris:${DB_PASS}@postgres:5432/orris
      - TEXTBOOK_DB_PATH=/data/medical_library.db
      - TEXTBOOK_BASE_PATH=/textbooks
    volumes:
      - ./data:/data
      - ./textbooks:/textbooks
      - ./skills:/skills:ro
    depends_on: [redis, postgres]

  frontend:
    build: ./frontend
    ports: ["3000:3000"]
    environment:
      - NEXT_PUBLIC_API_URL=http://backend:8000

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: orris
      POSTGRES_USER: orris
      POSTGRES_PASSWORD: ${DB_PASS}
    volumes:
      - postgres_data:/var/lib/postgresql/data

  sandbox:
    build: ./sandbox  # Isolated execution environment
    privileged: true  # Required for Docker-in-Docker
    volumes:
      - sandbox_workspace:/workspace

  qdrant:
    image: qdrant/qdrant:latest
    ports: ["6333:6333"]
    volumes:
      - qdrant_data:/qdrant/storage

volumes:
  postgres_data:
  qdrant_data:
  sandbox_workspace:

9.2 Kubernetes Production Architecture

# k8s/backend-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: orris-backend
spec:
  replicas: 3
  selector:
    matchLabels:
      app: orris-backend
  template:
    spec:
      containers:
      - name: backend
        image: orris-backend:latest
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        env:
        - name: ANTHROPIC_API_KEY
          valueFrom:
            secretKeyRef:
              name: orris-secrets
              key: anthropic-api-key
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: orris-backend-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: orris-backend
  minReplicas: 2
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

9.3 CI/CD Pipeline

# .github/workflows/deploy.yml
name: Deploy Orris

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run behavioral equivalence tests
        run: pytest tests/behavioral/ -v
      - name: Run retrieval accuracy tests
        run: pytest tests/retrieval/ -v
      - name: Run hallucination tests
        run: pytest tests/hallucination/ -v

  build-push:
    needs: test
    steps:
      - name: Build and push Docker images
        run: |
          docker build -t orris-backend:${{ github.sha }} ./backend
          docker push orris-backend:${{ github.sha }}

  deploy:
    needs: build-push
    steps:
      - name: Deploy to Kubernetes
        run: |
          kubectl set image deployment/orris-backend \
            backend=orris-backend:${{ github.sha }}
          kubectl rollout status deployment/orris-backend

---

10. MONITORING + RELIABILITY

10.1 Observability Stack

# telemetry/setup.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

def setup_telemetry():
    provider = TracerProvider()
    provider.add_span_processor(
        BatchSpanProcessor(OTLPSpanExporter(endpoint="http://tempo:4317"))
    )
    trace.set_tracer_provider(provider)

tracer = trace.get_tracer("orris")

# Usage in orchestration:
with tracer.start_as_current_span("medical_library_retrieval") as span:
    span.set_attribute("query", query)
    span.set_attribute("results_count", len(results))
    result = retriever.retrieve(query)
    span.set_attribute("top_book", result.headings[0].book_title if result.headings else "none")

10.2 Hallucination Mitigation

# reliability/hallucination_guard.py
class HallucinationGuard:
    """
    Orris's primary hallucination defense is SOURCE GROUNDING:
    - Medical answers MUST come from retrieved textbook content
    - PubMed citations MUST include real PMIDs
    - Drug dosages MUST be cited to a source
    - If uncertain: explicit "I don't know" required
    """

    def check_citation_integrity(self, response: str, retrieved_sources: List[Source]) -> bool:
        """Verify all citations in response exist in retrieved_sources"""
        cited = self.extract_citations(response)
        for citation in cited:
            if not any(s.matches(citation) for s in retrieved_sources):
                return False  # Fabricated citation detected
        return True

    def check_uncertainty_markers(self, response: str, query_type: str) -> bool:
        """For medical queries, any uncertain claim must be labeled"""
        UNCERTAINTY_PHRASES = ["I'm not sure", "I don't know", "unclear", "uncertain"]
        # If query is medical and no sources retrieved, response must hedge
        if query_type == "MEDICAL" and not self.has_retrieved_sources(response):
            return any(p in response for p in UNCERTAINTY_PHRASES)
        return True

10.3 Key Metrics (Grafana Dashboards)

# metrics to track:
- orris_response_latency_p50/p95/p99     # End-to-end response time
- orris_tool_calls_per_session           # Tool usage efficiency
- orris_budget_exhaustion_rate           # % sessions hitting limits
- orris_medical_routing_rate             # % medical queries correctly routed
- orris_citation_fabrication_rate        # Hallucination guard failures
- orris_retrieval_hit_rate               # FTS5 search returning results
- orris_skill_invocation_count_by_skill  # Which skills are used most
- orris_sandbox_execution_duration       # Code execution latency
- orris_correction_loop_rate             # How often self-correction triggers

---

11. VALIDATION + TESTING

11.1 Behavioral Equivalence Tests

# tests/behavioral/test_medical_routing.py
class TestMedicalFirstRouting:
    MEDICAL_QUERIES = [
        "What are the symptoms of myocardial infarction?",
        "Explain the mechanism of metformin",
        "What is the normal range for serum creatinine?",
    ]

    def test_medical_query_invokes_skill_before_llm(self, orris_client):
        for query in self.MEDICAL_QUERIES:
            result = orris_client.send(query)
            # First tool call MUST be skill invocation or query_library_db
            assert result.tool_calls[0].tool in ["skill_invoke", "query_library_db"], \
                f"Medical query '{query}' did not route to medical skill first"

    def test_medical_response_includes_citation(self, orris_client):
        result = orris_client.send("What is the treatment for hypertension?")
        assert re.search(r'\([^)]+,\s*p\.\s*\d+\)', result.response), \
            "Medical response missing textbook citation"

# tests/behavioral/test_banned_phrases.py
class TestResponseQuality:
    BANNED = ["delve", "crucial", "leverage", "robust", "comprehensive",
              "certainly", "absolutely", "of course", "as an AI"]

    def test_no_banned_phrases(self, orris_client):
        responses = orris_client.send_batch(SAMPLE_QUERIES_100)
        for r in responses:
            for phrase in self.BANNED:
                assert phrase.lower() not in r.response.lower(), \
                    f"Banned phrase '{phrase}' found in response"

    def test_no_em_dashes(self, orris_client):
        responses = orris_client.send_batch(SAMPLE_QUERIES_100)
        for r in responses:
            assert '—' not in r.response and '–' not in r.response

11.2 Retrieval Accuracy Tests

# tests/retrieval/test_fts5.py
KNOWN_RETRIEVABLE = [
    ("myocardial infarction treatment", "Harrison's Principles"),
    ("metformin mechanism", "Pharmacology textbook"),
    ("ECG interpretation", "Clinical cardiology"),
]

def test_fts5_retrieval_returns_expected_sources(retriever):
    for query, expected_book in KNOWN_RETRIEVABLE:
        results = retriever.search_paragraphs(query)
        assert any(expected_book in r.book_title for r in results), \
            f"Query '{query}' failed to retrieve from {expected_book}"

11.3 Budget Enforcement Tests

def test_budget_hard_limits_enforced(orris_client):
    """System must refuse tool calls that exceed budget"""
    session = orris_client.create_session(limits={"web_search": 2})
    orris_client.send("Search the web for X", session_id=session.id)
    orris_client.send("Search the web for Y", session_id=session.id)
    result = orris_client.send("Search the web for Z", session_id=session.id)
    assert result.tool_calls_attempted["web_search"] == 0, \
        "Budget limit not enforced - web_search executed after limit reached"

---

12. BUILD SEQUENCE

Phase 1 - MVP (Weeks 1-3)

1. SQLite FTS5 medical textbook database (pre-built or ingested)
2. Basic FastAPI backend with single /chat endpoint
3. Claude API integration with system prompt
4. Medical query router (keyword-based)
5. Medical library retrieval pipeline (FTS5 queries)
6. Response reflector (banned phrase + em dash checker)
7. Basic Next.js frontend with SSE streaming
8. Redis session memory (in-memory turns)
9. Docker Compose for local dev

Phase 2 - Production Core (Weeks 4-6)

10. LangGraph orchestration graph
11. Full skill system (SKILL.md loader + dispatcher)
12. Parallel tool execution
13. PubMed API integration + evidence tiering
14. Web search integration
15. Budget tracking + enforcement
16. Sandbox execution environment (Daytona)
17. File generators (PPTX, DOCX, PDF, XLSX)
18. PostgreSQL for session persistence
19. Citation rendering in frontend
20. Evidence panel UI component

Phase 3 - Production Hardening (Weeks 7-9)

21. Kubernetes deployment + HPA
22. OpenTelemetry + Grafana observability
23. Hallucination guard (citation integrity checking)
24. Cohere reranker for retrieval quality
25. Qdrant vector DB (semantic search supplement to FTS5)
26. Subagent spawning (LangGraph subgraphs)
27. Full behavioral equivalence test suite
28. CI/CD pipeline (GitHub Actions)
29. Disaster recovery + backup strategy
30. Rate limiting + authentication

Phase 4 - Scale (Weeks 10+)

31. Multi-region deployment
32. CDN for static assets + file downloads
33. Async ingestion pipeline for new textbooks
34. A/B testing framework for prompt variants
35. Token cost monitoring + optimization
36. Advanced memory: long-term user preferences
37. Adversarial stress testing

---

13. FINAL DNA SUMMARY

1. Medical-first routing - the single most important behavioral guarantee
2. Source grounding before response generation
3. Parallel tool execution by default
4. Budget tracking and enforcement
5. Explicit uncertainty over confident hallucination
6. Skill-file architecture for domain specialization

• SQLite FTS5 textbook database (the knowledge core)
• Skill dispatcher + SKILL.md system (routing intelligence)
• Budget enforcer (safety mechanism)
• Response reflector/validator (quality gate)
• Session memory manager (continuity)

• Bypassing skill invocation for medical queries (hallucinated medical advice)
• Context window overflow without summarization (session degradation)
• Budget exhaustion without graceful degradation (broken responses)
• FTS5 query construction errors (zero retrieval results)