Cortex-Inara/cortex/routers/files.py

"""
Read/write Inara identity markdown files, and search past session logs.
Only whitelisted filenames are accessible — no path traversal possible.
"""
import re
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from persona import persona_path, set_context, validate as validate_persona
from config import settings as _settings

router = APIRouter()

ALLOWED = {
    "SOUL.md",
    "IDENTITY.md",
    "USER.md",
    "PROTOCOLS.md",
    "CONTEXT_TIERS.md",
    "MEMORY.md",          # legacy — kept for reference
    "MEMORY_LONG.md",
    "MEMORY_MID.md",
    "MEMORY_SHORT.md",
    "MEMORY_LONG.bak1.md",
    "MEMORY_LONG.bak2.md",
    "MEMORY_MID.bak1.md",
    "MEMORY_MID.bak2.md",
    "MEMORY_SHORT.bak1.md",
    "MEMORY_SHORT.bak2.md",
    "HELP.md",
    # Agent private notes — backups only; AGENT_NOTES.md itself is agent-only
    "AGENT_NOTES.bak1.md",
    "AGENT_NOTES.bak2.md",
    "AGENT_NOTES.bak3.md",
}

# Files that can be read via the panel but not written by users
READ_ONLY = {
    "AGENT_NOTES.bak1.md",
    "AGENT_NOTES.bak2.md",
    "AGENT_NOTES.bak3.md",
}

# Files served from home/{user}/ instead of persona path
USER_FILES = {"email_allowlist.json", "usage.json"}


def _resolve(user: str, persona: str) -> None:
    """Validate and set context from query params. Raises HTTPException on bad input."""
    try:
        u, p = validate_persona(user, persona)
        set_context(u, p)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))


def _path(filename: str, user: str = ""):
    if filename in USER_FILES:
        if not user:
            raise HTTPException(status_code=400, detail="user param required for this file")
        return _settings.home_root() / user / filename
    if filename not in ALLOWED:
        raise HTTPException(status_code=404, detail=f"File not found: {filename}")
    return persona_path() / filename


@router.get("/files")
async def list_files(
    user: str = Query("scott"),
    persona: str = Query("inara"),
) -> dict:
    _resolve(user, persona)
    persona_dir = persona_path()
    files = []
    for name in sorted(ALLOWED):
        p = persona_dir / name
        st = p.stat() if p.exists() else None
        files.append({
            "name": name,
            "exists": p.exists(),
            "size": st.st_size if st else 0,
            "modified": st.st_mtime if st else None,
        })
    for name in sorted(USER_FILES):
        p = _settings.home_root() / user / name
        st = p.stat() if p.exists() else None
        files.append({
            "name": name,
            "exists": p.exists(),
            "size": st.st_size if st else 0,
            "modified": st.st_mtime if st else None,
            "scope": "user",
        })
    return {"files": files}


@router.get("/files/{filename}")
async def get_file(
    filename: str,
    user: str = Query("scott"),
    persona: str = Query("inara"),
) -> dict:
    _resolve(user, persona)
    p = _path(filename, user=user)
    if not p.exists():
        raise HTTPException(status_code=404, detail=f"{filename} does not exist")
    return {
        "name": filename,
        "content": p.read_text(),
        "readonly": filename in READ_ONLY,
    }


class FileWrite(BaseModel):
    content: str


@router.put("/files/{filename}")
async def save_file(
    filename: str,
    req: FileWrite,
    user: str = Query("scott"),
    persona: str = Query("inara"),
) -> dict:
    if filename in READ_ONLY:
        raise HTTPException(status_code=403, detail=f"{filename} is read-only.")
    _resolve(user, persona)
    p = _path(filename, user=user)
    p.write_text(req.content)
    return {"ok": True, "name": filename, "size": len(req.content)}


# ── Session search ────────────────────────────────────────────────────────────

_CONTEXT_CHARS = 120  # chars of context to include around each match


@router.get("/sessions/search")
async def search_sessions(
    q: str = Query(..., min_length=2),
    user: str = Query("scott"),
    persona: str = Query("inara"),
    limit: int = Query(20, ge=1, le=100),
) -> dict:
    """Full-text search across past session logs.

    Returns up to `limit` matches, newest sessions first.
    Each match includes a short excerpt (120 chars before/after) for context.
    """
    _resolve(user, persona)
    sessions_dir = persona_path() / "sessions"
    if not sessions_dir.exists():
        return {"query": q, "matches": [], "total_files_searched": 0}

    pattern = re.compile(re.escape(q), re.IGNORECASE)
    session_files = sorted(sessions_dir.glob("*.md"), reverse=True)  # newest first

    matches = []
    for sf in session_files:
        if len(matches) >= limit:
            break
        try:
            text = sf.read_text()
        except OSError:
            continue
        for m in pattern.finditer(text):
            if len(matches) >= limit:
                break
            start = max(0, m.start() - _CONTEXT_CHARS)
            end   = min(len(text), m.end() + _CONTEXT_CHARS)
            excerpt = text[start:end].strip()
            # Prefix with ellipsis if we truncated the left side
            if start > 0:
                excerpt = "…" + excerpt
            if end < len(text):
                excerpt = excerpt + "…"
            matches.append({
                "date":    sf.stem,          # YYYY-MM-DD
                "excerpt": excerpt,
            })

    return {
        "query":               q,
        "matches":             matches,
        "total_files_searched": len(session_files),
    }