resolutionflow/docs/plans/archive/2026-03-20-search-recall-evidence-impl.md

Search & Recall + Evidence-Rich Sessions — Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Add file upload with clipboard paste for evidence capture during FlowPilot sessions, and build three layers of session search (structured filters, full-text, semantic similarity).

Architecture: Evidence uses Railway Object Storage (S3-compatible) via boto3, with a file_uploads table tracking metadata. Search uses PostgreSQL generated tsvector columns with GIN indexes for full-text, and Voyage AI embeddings with pgvector for semantic similarity (reusing existing RAG infrastructure). A RichTextInput component handles clipboard paste-to-upload UX.

Tech Stack: FastAPI, SQLAlchemy 2.0 (async), boto3 (S3), pgvector, Voyage AI embeddings, React 19, TypeScript, Tailwind CSS v4

Design doc: docs/plans/2026-03-20-search-recall-evidence-design.md

---

Part A: Evidence-Rich Sessions

Task 1: S3 storage service + file_uploads model

Files:

• Create: backend/app/services/storage_service.py
• Create: backend/app/models/file_upload.py
• Modify: backend/app/models/__init__.py
• Modify: backend/app/core/config.py
• Create: migration

Step 1: Add storage config

In backend/app/core/config.py, add to the Settings class:

# Object Storage (Railway S3-compatible)
STORAGE_ENDPOINT: str | None = None
STORAGE_ACCESS_KEY: str | None = None
STORAGE_SECRET_KEY: str | None = None
STORAGE_BUCKET_NAME: str = "resolutionflow-uploads"
STORAGE_REGION: str = "us-east-1"

Step 2: Create file_uploads model

backend/app/models/file_upload.py:

"""File upload metadata — tracks files stored in S3-compatible object storage."""
import uuid
from datetime import datetime, timezone
from typing import Optional

from sqlalchemy import String, Integer, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.dialects.postgresql import UUID

from app.core.database import Base


class FileUpload(Base):
    __tablename__ = "file_uploads"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    account_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False, index=True
    )
    uploaded_by: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=False
    )
    session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
        UUID(as_uuid=True), ForeignKey("ai_sessions.id", ondelete="SET NULL"), nullable=True, index=True
    )
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    content_type: Mapped[str] = mapped_column(String(100), nullable=False)
    size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
    storage_key: Mapped[str] = mapped_column(String(500), nullable=False, unique=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
    )

Register in backend/app/models/__init__.py.

Step 3: Create storage service

backend/app/services/storage_service.py:

"""S3-compatible object storage service for file uploads."""
import logging
import uuid
from io import BytesIO

import boto3
from botocore.config import Config as BotoConfig
from botocore.exceptions import ClientError

from app.core.config import settings

logger = logging.getLogger(__name__)

ALLOWED_IMAGE_TYPES = {"image/png", "image/jpeg", "image/gif", "image/webp"}
ALLOWED_TEXT_TYPES = {"text/plain", "text/csv", "application/octet-stream"}
ALLOWED_TYPES = ALLOWED_IMAGE_TYPES | ALLOWED_TEXT_TYPES

MAX_IMAGE_SIZE = 5 * 1024 * 1024  # 5MB
MAX_TEXT_SIZE = 1 * 1024 * 1024   # 1MB
MAX_FILES_PER_SESSION = 20
MAX_BYTES_PER_SESSION = 50 * 1024 * 1024  # 50MB

PRESIGNED_URL_EXPIRY = 3600  # 1 hour


def _get_client():
    """Get S3 client configured for Railway Object Storage."""
    if not settings.STORAGE_ENDPOINT:
        raise RuntimeError("Object storage not configured (STORAGE_ENDPOINT missing)")
    return boto3.client(
        "s3",
        endpoint_url=settings.STORAGE_ENDPOINT,
        aws_access_key_id=settings.STORAGE_ACCESS_KEY,
        aws_secret_access_key=settings.STORAGE_SECRET_KEY,
        region_name=settings.STORAGE_REGION,
        config=BotoConfig(signature_version="s3v4"),
    )


def validate_upload(content_type: str, size_bytes: int) -> str | None:
    """Validate file type and size. Returns error message or None."""
    if content_type not in ALLOWED_TYPES:
        return f"File type {content_type} not allowed"
    max_size = MAX_IMAGE_SIZE if content_type in ALLOWED_IMAGE_TYPES else MAX_TEXT_SIZE
    if size_bytes > max_size:
        return f"File too large ({size_bytes} bytes, max {max_size})"
    return None


async def upload_file(
    file_data: bytes,
    filename: str,
    content_type: str,
    account_id: str,
) -> str:
    """Upload file to S3, returns the storage key."""
    ext = filename.rsplit(".", 1)[-1] if "." in filename else "bin"
    storage_key = f"uploads/{account_id}/{uuid.uuid4()}.{ext}"

    client = _get_client()
    client.upload_fileobj(
        BytesIO(file_data),
        settings.STORAGE_BUCKET_NAME,
        storage_key,
        ExtraArgs={"ContentType": content_type},
    )
    return storage_key


def get_presigned_url(storage_key: str) -> str:
    """Generate a time-limited presigned URL for downloading a file."""
    client = _get_client()
    return client.generate_presigned_url(
        "get_object",
        Params={"Bucket": settings.STORAGE_BUCKET_NAME, "Key": storage_key},
        ExpiresIn=PRESIGNED_URL_EXPIRY,
    )


async def delete_file(storage_key: str) -> None:
    """Delete a file from S3."""
    try:
        client = _get_client()
        client.delete_object(Bucket=settings.STORAGE_BUCKET_NAME, Key=storage_key)
    except ClientError:
        logger.warning(f"Failed to delete S3 object: {storage_key}")

Step 4: Generate migration, add boto3 to requirements

pip install boto3
echo "boto3>=1.34.0" >> backend/requirements.txt
cd backend && alembic revision --autogenerate -m "add file_uploads table"
alembic upgrade head

Step 5: Commit

git commit -m "feat(evidence): add S3 storage service and file_uploads model"

---

Task 2: Upload API endpoints

Files:

• Create: backend/app/api/endpoints/uploads.py
• Create: backend/app/schemas/upload.py
• Modify: backend/app/api/router.py
• Create: backend/tests/test_uploads.py

Schemas (backend/app/schemas/upload.py):

from datetime import datetime
from uuid import UUID
from pydantic import BaseModel

class FileUploadResponse(BaseModel):
    id: UUID
    filename: str
    content_type: str
    size_bytes: int
    url: str  # presigned URL
    created_at: datetime
    model_config = {"from_attributes": True}

Endpoints (backend/app/api/endpoints/uploads.py):

POST   /uploads                          — Multipart upload, returns FileUploadResponse
GET    /uploads/{id}/url                 — Presigned download URL
GET    /uploads?session_id={id}          — List uploads for a session
DELETE /uploads/{id}                     — Delete upload + S3 object

Key details:

• POST /uploads accepts UploadFile from FastAPI + session_id form field
• Validates content_type and size via storage_service.validate_upload()
• Checks per-session limits (count + total bytes) before uploading
• Rate limit: @limiter.limit("10/minute")
• All endpoints require auth via get_current_active_user
• Delete: verify ownership (uploaded_by == current_user.id OR user is admin)

Tests: Upload happy path, type rejection, size rejection, per-session limit, presigned URL, delete.

Commit:

git commit -m "feat(evidence): add file upload/download API endpoints"

---

Task 3: Frontend — RichTextInput component

Files:

• Create: frontend/src/components/common/RichTextInput.tsx
• Create: frontend/src/api/uploads.ts
• Create: frontend/src/types/upload.ts

Types (frontend/src/types/upload.ts):

export interface FileUploadResponse {
  id: string
  filename: string
  content_type: string
  size_bytes: number
  url: string
  created_at: string
}

export interface PendingUpload {
  id: string          // temp client ID
  file: File
  preview: string     // object URL for thumbnail
  status: 'uploading' | 'done' | 'error'
  result?: FileUploadResponse
  error?: string
}

API (frontend/src/api/uploads.ts):

import apiClient from './client'
import type { FileUploadResponse } from '@/types/upload'

export const uploadsApi = {
  async upload(file: File, sessionId?: string): Promise<FileUploadResponse> {
    const formData = new FormData()
    formData.append('file', file)
    if (sessionId) formData.append('session_id', sessionId)
    const response = await apiClient.post<FileUploadResponse>('/uploads', formData, {
      headers: { 'Content-Type': 'multipart/form-data' },
    })
    return response.data
  },
  async getUrl(id: string): Promise<string> {
    const response = await apiClient.get<{ url: string }>(`/uploads/${id}/url`)
    return response.data.url
  },
  async list(sessionId: string): Promise<FileUploadResponse[]> {
    const response = await apiClient.get<FileUploadResponse[]>('/uploads', { params: { session_id: sessionId } })
    return response.data
  },
  async remove(id: string): Promise<void> {
    await apiClient.delete(`/uploads/${id}`)
  },
}

RichTextInput component (frontend/src/components/common/RichTextInput.tsx):

Props:

interface RichTextInputProps {
  value: string
  onChange: (value: string) => void
  onFilesChange?: (uploads: FileUploadResponse[]) => void
  sessionId?: string
  placeholder?: string
  rows?: number
  className?: string
  disabled?: boolean
}

Behavior:

• Renders a <textarea> with standard styling
• Listens for paste event — if clipboardData.items contains image/* blob, extracts it
• On image paste: creates a PendingUpload, shows thumbnail strip below textarea, calls uploadsApi.upload() in background
• Thumbnail strip: horizontal row of image previews (64x64 rounded), each with:
  • Uploading: pulse animation overlay
  • Done: image thumbnail, X button to remove
  • Error: red border, retry button
• Completed uploads reported to parent via onFilesChange
• Also supports drag-and-drop (same handler)
• Design: thumbnails in a flex gap-2 flex-wrap row below the textarea, .glass-card styling on each thumbnail

Build and commit:

cd frontend && npm run build
git commit -m "feat(evidence): add RichTextInput with clipboard paste upload"

---

Task 4: Wire RichTextInput into FlowPilot

Files:

• Modify: frontend/src/components/flowpilot/FlowPilotIntake.tsx
• Modify: frontend/src/components/flowpilot/FlowPilotOptions.tsx (free-text input)
• Modify: frontend/src/components/flowpilot/EscalateModal.tsx

Replace plain <textarea> elements in these components with <RichTextInput>. Pass the current session ID so uploads are linked.

For intake: the session doesn't exist yet at intake time. Upload without session_id, then link uploads to the session after creation (via a PATCH or by passing upload IDs to the session creation endpoint).

Alternative simpler approach: upload to a temporary session_id (null), then after session creation, call PATCH /uploads/{id} to set session_id. Or include upload_ids in the session creation payload.

Build and commit:

git commit -m "feat(evidence): wire clipboard paste into FlowPilot intake, free-text, and escalation"

---

Task 5: Evidence in exports

Files:

• Modify: backend/app/services/export_service.py

Extend each format generator to include file upload references:

• Query file_uploads for the session
• Markdown: ![{filename}]({presigned_url}) for images, [{filename}]({presigned_url}) for text files
• HTML/PDF: <img src="{presigned_url}" alt="{filename}" style="max-width: 100%"> for images
• PSA: List as Evidence: {filename} — {presigned_url} (CW notes don't support inline images)
• Text: [Attachment: {filename}] with URL on next line

Generate presigned URLs at export time via storage_service.get_presigned_url().

Commit:

git commit -m "feat(evidence): include file uploads in session exports"

---

Part B: Search & Recall

Task 6: Structured filters on AI sessions

Files:

• Modify: backend/app/api/endpoints/ai_sessions.py
• Modify: frontend/src/pages/SessionHistoryPage.tsx

Backend: Extend list_sessions endpoint with new query params:

async def list_sessions(
    ...,
    problem_domain: Optional[str] = Query(None),
    matched_flow_id: Optional[UUID] = Query(None),
    confidence_tier: Optional[str] = Query(None, pattern="^(guided|exploring|discovery)$"),
    ticket_id: Optional[str] = Query(None),
    date_from: Optional[datetime] = Query(None),
    date_to: Optional[datetime] = Query(None),
    q: Optional[str] = Query(None, min_length=2, max_length=200),
):

Add .where() clauses for each non-None filter.

Frontend: Add a filter bar to the AI Sessions tab on SessionHistoryPage:

• Search input (for q param)
• Problem domain dropdown (populated from distinct domains in sessions)
• Confidence tier pills (All / Guided / Exploring / Discovery)
• Date range inputs
• Match existing Flow Sessions filter bar styling

Commit:

git commit -m "feat(search): add structured filters to AI session list"

---

Task 7: Full-text search (PostgreSQL FTS)

Files:

• Create: migration for tsvector column + GIN index
• Modify: backend/app/api/endpoints/ai_sessions.py

Migration:

# Generated tsvector column on ai_sessions
op.execute("""
    ALTER TABLE ai_sessions ADD COLUMN IF NOT EXISTS search_vector tsvector
    GENERATED ALWAYS AS (
        to_tsvector('english',
            coalesce(intake_summary, '') || ' ' ||
            coalesce(resolution_summary, '') || ' ' ||
            coalesce(escalation_reason, '') || ' ' ||
            coalesce(problem_domain, ''))
    ) STORED
""")
op.execute("""
    CREATE INDEX IF NOT EXISTS idx_ai_sessions_search
    ON ai_sessions USING gin(search_vector)
""")

Wire into list endpoint: When q is provided, add:

from sqlalchemy import text as sa_text
query = query.where(
    sa_text("ai_sessions.search_vector @@ plainto_tsquery('english', :q)")
).params(q=q)

Extend Command Palette: In frontend/src/components/layout/CommandPalette.tsx, add AI session search alongside flows. Call the AI sessions list endpoint with q param. Show results in a new "AI Sessions" group.

Commit:

git commit -m "feat(search): add PostgreSQL FTS on AI sessions with Command Palette integration"

---

Task 8: Similar session matching (semantic)

Files:

• Create: backend/app/models/ai_session_embedding.py
• Create: migration
• Modify: backend/app/api/endpoints/ai_sessions.py
• Modify: backend/app/services/flowpilot_engine.py

Model (backend/app/models/ai_session_embedding.py):

Same pattern as tree_embedding.py:

class AISessionEmbedding(Base):
    __tablename__ = "ai_session_embeddings"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    session_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("ai_sessions.id", ondelete="CASCADE"),
        nullable=False, unique=True, index=True
    )
    account_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("accounts.id", ondelete="CASCADE"),
        nullable=False, index=True
    )
    chunk_text: Mapped[str] = mapped_column(Text, nullable=False)
    embedding_model: Mapped[str] = mapped_column(String(50), nullable=False, default="voyage-3.5")
    # embedding column created in migration as vector(1024)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
    )

Migration adds the table + embedding vector(1024) column via raw SQL (same as tree_embeddings migration).

Generate embedding: In flowpilot_engine.py, after session intake is processed, generate an embedding of the intake_summary using the existing rag_service / Voyage AI client. Store in ai_session_embeddings. Update embedding on resolution (add resolution_summary to the text).

Endpoint: GET /ai-sessions/{id}/similar?limit=5

@router.get("/{session_id}/similar")
async def get_similar_sessions(session_id, current_user, db, limit=5):
    # Get this session's embedding
    # Cosine similarity query against all session embeddings for the account
    # Return top N with similarity score + session summary

Use the same cosine similarity query pattern as rag_service.search().

Commit:

git commit -m "feat(search): add semantic similar session matching via Voyage AI embeddings"

---

Task 9: Similar sessions UI

Files:

• Create: frontend/src/components/flowpilot/SimilarSessions.tsx
• Modify: frontend/src/components/flowpilot/FlowPilotSession.tsx
• Modify: frontend/src/api/flowpilotAnalytics.ts (or ai-sessions API)

Component — small card list showing 3-5 similar sessions:

interface SimilarSession {
  id: string
  intake_summary: string
  status: string
  resolution_summary: string | null
  problem_domain: string | null
  similarity: number
  created_at: string
}

Each card: .glass-card with:

• Problem summary (1-2 lines, truncated)
• Status badge (resolved/escalated)
• Resolution summary if resolved (1 line)
• Similarity percentage
• Click → navigate to session detail

Where it renders:

• FlowPilot session sidebar (desktop) or expandable section (mobile): "Similar Past Sessions" below the session info
• Only fetched if the session has been through intake (has intake_summary)

Build and commit:

cd frontend && npm run build
git commit -m "feat(search): add similar sessions UI in FlowPilot sidebar"

---

Task 10: Final verification and docs

Step 1: Run backend tests

cd backend && python -m pytest --override-ini="addopts="

Step 2: Run frontend build

cd frontend && npm run build

Step 3: Update CURRENT-STATE.md

Add evidence-rich sessions and search & recall to completed items.

Step 4: Update stack priorities doc — mark items 1 and 2 as complete.

Step 5: Commit

git commit -m "docs: update CURRENT-STATE.md — Search & Recall and Evidence-Rich Sessions complete"