backend/app/main.py

"""CCR Platform - FastAPI application.

Single deployable: serves the JSON API under /api and the prebuilt React
dashboard as static files at /. Local-first by design: corpora, embeddings,
and results never leave this machine (sentence-transformers runs locally),
which keeps sensitive research text IRB-friendly and every run reproducible
against pinned model weights.
"""

from __future__ import annotations

import json
import os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path

from fastapi import Depends, FastAPI, HTTPException, Request, Response, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy.orm import Session

from . import admin as admin_module
from . import auth, auth_google, retention, storage
from . import item_generation
from . import jobs as jobs_module
from . import registry
from .ccr import FAKE_MODEL_NAME
from .construct_files import parse_construct_file
from .construct_lib import sync_library
from .db import (
    DATA_DIR,
    Base,
    SessionLocal,
    auto_migrate_sqlite,
    engine,
    get_db,
    lock_down_public_schema,
)
from .ingest import IngestError, load_corpus, max_rows as corpus_max_rows, suggest_text_column
from .models import (
    AdminAudit,
    Construct,
    Corpus,
    GenerationEvent,
    Invite,
    Job,
    Project,
    RoleAssignment,
    User,
)
from .reproducibility import (
    requirements_filename,
    requirements_text,
    script_filename,
    script_text,
)
from .schemas import (
    ConstructCreate,
    ConstructGeneration,
    ConstructOut,
    CorpusOut,
    GenerateItemsIn,
    GenerateItemsOut,
    JobCreate,
    JobOut,
    LoginIn,
    ProjectCreate,
    ProjectOut,
    ProjectPatch,
    RegisterIn,
)

ALLOWED_SUFFIXES = (".csv", ".xlsx", ".xls")

# Upload ceiling: an OOM/abuse backstop, NOT the research-workflow limit.
# Job cost scales with rows and tokens, not file bytes (20k tweets and 20k
# essays are the same row count and wildly different compute), so the limit
# that actually bounds a run is CCR_MAX_ROWS - see ingest.max_rows(). Keep
# this loose and tune rows per deployment.
MAX_UPLOAD_BYTES_DEFAULT = 50 * 1024 * 1024
UPLOAD_CHUNK_BYTES = 1024 * 1024

# Multi-construct runs: bounds the export width (items × constructs columns)
# and the results-page size; compute is dominated by the single corpus
# embedding pass either way.
MAX_CONSTRUCTS_PER_RUN = 10


def max_upload_bytes() -> int:
    """Global upload ceiling, env-configurable (CCR_MAX_UPLOAD_BYTES)."""
    return int(os.environ.get("CCR_MAX_UPLOAD_BYTES", MAX_UPLOAD_BYTES_DEFAULT))


class _UploadTooLarge(Exception):
    """Raised mid-stream once an upload passes its byte ceiling."""


async def _stream_to_temp(file: UploadFile, dest: Path, ceiling: int) -> int:
    """Copy an upload to `dest` in chunks, aborting past `ceiling` bytes.

    Streaming rather than file.read()-ing the whole payload into one bytes
    object keeps peak memory at one chunk: an oversized upload is rejected
    without ever being fully resident, and the partial temp file is removed
    on any failure (including a client disconnect mid-upload).
    """
    total = 0
    try:
        with dest.open("wb") as out:
            while chunk := await file.read(UPLOAD_CHUNK_BYTES):
                total += len(chunk)
                if total > ceiling:
                    raise _UploadTooLarge()
                out.write(chunk)
    except BaseException:
        dest.unlink(missing_ok=True)
        raise
    return total


# Languages offered in the UI selector; detection may report others (ISO 639-1).
SELECTABLE_LANGUAGES = [
    "en", "es", "fr", "de", "it", "pt", "nl", "ru", "zh", "ja", "ko", "ar", "hi", "tr", "fa",
]

@asynccontextmanager
async def lifespan(_: FastAPI):
    """Create tables and sync the construct library (YAML source of truth) at startup."""
    Base.metadata.create_all(engine)
    auto_migrate_sqlite(engine, Base.metadata)  # additive column adds for existing dev DBs
    lock_down_public_schema(engine, Base.metadata)  # Supabase: RLS on, REST surface closed
    registry.list_models()  # fail fast on an invalid models.yaml
    db = SessionLocal()
    try:
        sync_library(db)
    finally:
        db.close()
    jobs_module.recover_orphaned_jobs()
    retention.start_cleanup()  # anonymous-data TTL purge (no-op if CCR_ANON_TTL_HOURS=0)
    if os.environ.get("CCR_WARM_MODEL") == "1" and os.environ.get("CCR_FAKE_EMBEDDINGS") != "1":
        import threading

        def _warm():
            try:
                from .ccr import get_backend

                get_backend(registry.default_model().id).encode(["warm up"])
            except Exception:
                pass  # first real run will load the model instead

        threading.Thread(target=_warm, daemon=True, name="ccr-warmup").start()
    yield
    retention.stop_cleanup()
    jobs_module.shutdown_executor()


app = FastAPI(title="CCR Platform", version="0.1.0", lifespan=lifespan)
app.include_router(admin_module.router)

app.add_middleware(GZipMiddleware, minimum_size=1024)  # constructs payload + SPA compress ~4-5x
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"],  # Vite dev server
    allow_methods=["*"],
    allow_headers=["*"],
)


# ---------------------------------------------------------------- helpers
def _construct_out(c: Construct) -> ConstructOut:
    items = json.loads(c.items_json)
    flags = json.loads(c.reverse_flags_json or "[]") or [False] * len(items)
    return ConstructOut(
        id=c.id,
        name=c.name,
        description=c.description,
        reference=c.reference,
        items=items,
        reverse_scored=flags,
        is_seed=c.is_seed,
        version=c.version or 1,
        verification_status=c.verification_status or "draft",
        language=c.language or "en",
        category=c.category or "",
        item_hash=(c.item_hash or "")[:16],
        ai_generated=bool(getattr(c, "generation_json", "") or ""),
    )


def _job_out(db: Session, j: Job) -> JobOut:
    construct_ids = jobs_module.job_construct_ids(j)
    constructs = [db.get(Construct, cid) for cid in construct_ids]
    names = [c.name if c else "?" for c in constructs]
    corpus = db.get(Corpus, j.corpus_id)
    opp_id = j.opposite_construct_id or ""
    opp = db.get(Construct, opp_id) if opp_id else None
    return JobOut(
        id=j.id,
        project_id=j.project_id,
        corpus_id=j.corpus_id,
        construct_id=j.construct_id,
        construct_ids=construct_ids,
        construct_name=names[0] if names else "",
        construct_names=names,
        opposite_construct_id=opp_id,
        opposite_construct_name=(opp.name if opp else ""),
        similarity_metric=j.similarity_metric or "",
        anchored=bool(opp_id),
        corpus_filename=corpus.filename if corpus else "",
        text_column=j.text_column,
        model_name=j.model_name,
        language=j.language or "en",
        status=j.status,
        progress=j.progress,
        error=j.error,
        created_at=j.created_at,
        started_at=j.started_at,
        finished_at=j.finished_at,
    )


def _get_or_404(db: Session, model, obj_id: str):
    obj = db.get(model, obj_id)
    if obj is None:
        raise HTTPException(404, f"{model.__name__} not found")
    return obj


# ------------------------------------------------------------------- meta
@app.get("/api/health")
def health():
    try:
        import sentence_transformers  # noqa: F401

        st = True
    except ImportError:
        st = False
    return {"status": "ok", "sentence_transformers_available": st}


@app.get("/api/models")
def list_models():
    """Model options from the registry (spec 0003) - never hardcoded."""
    return [
        {
            "id": m.id,
            "label": m.display_name,
            "default": m.default,
            "languages": (m.language_set_name or ", ".join(sorted(m.supported_languages)) or "unspecified"),
            "speed_tier": m.speed_tier,
            "quality_tier": m.quality_tier,
            "warnings": list(m.user_warnings),
        }
        for m in registry.list_models()
    ]


@app.get("/api/languages")
def list_languages():
    return SELECTABLE_LANGUAGES


# ------------------------------------------------------------------ accounts
# Local email+password accounts (auth.py) - the free interim provider. The
# managed swap (Supabase: Google + email/password) replaces token issuance
# only; every other endpoint just depends on auth.get_current_user.
def _saved_runs_used(db: Session, user_id: str) -> int:
    return (
        db.query(Job)
        .join(Project, Job.project_id == Project.id)
        .filter(Project.owner_user_id == user_id, Job.status.in_(("queued", "running", "completed")))
        .count()
    )


def _generations_used_today(db: Session, user_id: str) -> int:
    """created_at is an ISO-8601 string, so a date-prefix match is the day
    filter (UTC, same clock as the run counter)."""
    today = datetime.now(timezone.utc).date().isoformat()
    return (
        db.query(GenerationEvent)
        .filter(GenerationEvent.user_id == user_id, GenerationEvent.created_at.like(f"{today}%"))
        .count()
    )


def _set_session_cookie(response: Response, user: User) -> None:
    response.set_cookie(
        auth.COOKIE_NAME,
        auth.create_session_token(user.id, user.email, user.name),
        httponly=True,
        samesite="lax",
        secure=auth.cookies_secure(),
        max_age=30 * 24 * 3600,
    )


@app.get("/api/auth/me")
def auth_me(
    request: Request,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    if user:
        row = db.get(User, user["id"])
        role = auth.normalize_role(row.role if row else None)
        return {
            "signed_in": True,
            "name": user["name"],
            "email": user["email"],
            "role": role,
            # effective admin: env allowlist OR staff role (pi/maintainer) -
            # drives the Admin link in the header
            "is_admin": auth.is_admin(user["email"]) or auth.role_is_staff(role),
            # max_rows is NOT unlimited for signed-in users: CCR_MAX_ROWS is a
            # global ingest ceiling and is usually the limit that actually
            # binds (deployed instances set it well below the byte ceiling).
            # Reporting None here told users there was no row limit until they
            # hit one mid-upload.
            "limits": {"max_bytes": max_upload_bytes(), "max_rows": corpus_max_rows()},
            "usage": {
                "saved_runs": _saved_runs_used(db, user["id"]),
                # lab members and above: unlimited saved runs (admin-granted)
                "max_saved_runs": None if auth.role_unlimited(role) else auth.user_max_saved_runs(),
                "generations_used_today": _generations_used_today(db, user["id"]),
                "max_generations_per_day": item_generation.user_max_generations_per_day(),
            },
            # signed-in-only feature (PI decision 2026-08-05); False when the
            # instance has no ANTHROPIC_API_KEY - the UI hides the button
            "generation_available": item_generation.configured(),
            # live model identity (name, provider, prompt version, item caps)
            # so the construct form's info tooltip never hardcodes the model
            "generation": item_generation.public_info(),
        }
    return {
        "signed_in": False,
        "name": None,
        "email": None,
        "google_available": auth_google.configured(),
        "limits": {"max_bytes": auth.anon_max_bytes(), "max_rows": auth.anon_max_rows()},
        "usage": {
            "runs_used_today": auth.runs_used_today(request),
            "max_runs_per_day": auth.anon_max_runs_per_day(),
        },
        # true configured state: the UI shows anonymous visitors a "sign in to
        # use AI drafting" nudge only when the feature actually exists here
        # (the endpoint itself still requires sign-in regardless)
        "generation_available": item_generation.configured(),
        "generation": item_generation.public_info(),
    }


def _live_invite(db: Session, invite_token: str | None) -> Invite | None:
    """The Invite row for a token that is well-signed, unexpired, and still
    live (exists, not revoked). Tokens from before invites became stateful
    have no row and are therefore dead. While invites are on hold
    (auth.invites_enabled), every token is dead."""
    if not auth.invites_enabled():
        return None
    verified = auth.verify_invite_token(invite_token)
    if not verified:
        return None
    invite = db.get(Invite, verified["jti"])
    if invite is None or invite.revoked_at:
        return None
    return invite


def _initial_role(db: Session, email: str, invite_token: str | None = None) -> str:
    """Tier a brand-new account lands at. Precedence: an email-bound
    pre-assignment (may carry staff roles - the 'credentials before first
    sign-in' case) beats an invite link (bearer token, external/lab only)
    beats the external default. Claiming is recorded in the audit trail,
    and invite redemptions also land on the invite row itself."""
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    pre = db.query(RoleAssignment).filter_by(email=email).first()
    if pre is not None and not pre.claimed_at:
        pre.claimed_at = now
        role = auth.normalize_role(pre.role)
        db.add(AdminAudit(actor_email=email, action="role_claimed", target=email,
                          detail=f"pre-assigned {role} by {pre.assigned_by}"))
        return role
    invite = _live_invite(db, invite_token)
    if invite is not None:
        role = auth.normalize_role(invite.role)
        redemptions = json.loads(invite.redemptions_json or "[]")
        redemptions.append({"email": email, "at": now})
        invite.redemptions_json = json.dumps(redemptions)
        db.add(AdminAudit(actor_email=email, action="invite_redeemed",
                          target=email, detail=f"{role} link by {invite.created_by}"))
        return role
    return "external"


@app.post("/api/auth/register", status_code=201)
def register(body: RegisterIn, response: Response, db: Session = Depends(get_db)):
    email = body.email.strip().lower()
    if not auth.valid_email(email):
        raise HTTPException(400, "Please enter a valid email address.")
    if len(body.password) < auth.MIN_PASSWORD_LEN:
        raise HTTPException(400, f"Password must be at least {auth.MIN_PASSWORD_LEN} characters.")
    if db.query(User).filter_by(email=email).first():
        raise HTTPException(409, "An account with this email already exists. Sign in instead.")
    # A dead invite link (bad signature, expired, or revoked) should say so,
    # not silently demote to external - unless a pre-assignment covers the
    # email anyway.
    has_preassignment = db.query(RoleAssignment).filter_by(email=email).first() is not None
    if body.invite_token and _live_invite(db, body.invite_token) is None and not has_preassignment:
        if not auth.invites_enabled():
            raise HTTPException(
                400,
                "Invite links are currently on hold. Ask an admin to grant your "
                "email access instead - or register normally as an external user.",
            )
        raise HTTPException(400, "This invite link is invalid, expired, or revoked. Ask for a new one.")
    user = User(
        email=email, name=body.name.strip(),
        password_hash=auth.hash_password(body.password),
        role=_initial_role(db, email, body.invite_token),
    )
    db.add(user)
    db.commit()
    _set_session_cookie(response, user)
    return {"signed_in": True, "name": user.name, "email": user.email}


@app.post("/api/auth/login")
def login(body: LoginIn, response: Response, db: Session = Depends(get_db)):
    email = body.email.strip().lower()
    user = db.query(User).filter_by(email=email).first()
    if user is not None and not user.password_hash:
        raise HTTPException(401, "This account uses Google sign-in - use the Google button.")
    if user is None or not auth.verify_password(body.password, user.password_hash):
        raise HTTPException(401, "Incorrect email or password.")
    _set_session_cookie(response, user)
    return {"signed_in": True, "name": user.name, "email": user.email}


@app.get("/api/auth/google/login")
def google_login():
    """Start the Google sign-in flow (Supabase PKCE). Plain redirect - the
    frontend links here directly, no SDK involved."""
    if not auth_google.configured():
        raise HTTPException(503, "Google sign-in is not configured on this instance.")
    from fastapi.responses import RedirectResponse

    url, verifier = auth_google.begin()
    resp = RedirectResponse(url, status_code=307)
    resp.set_cookie(
        auth_google.VERIFIER_COOKIE,
        auth.sign_payload({"v": verifier}),
        httponly=True,
        samesite="lax",
        secure=auth.cookies_secure(),
        max_age=auth_google.VERIFIER_TTL_SECONDS,
    )
    return resp


@app.get("/api/auth/google/callback")
def google_callback(request: Request, code: str = "", db: Session = Depends(get_db)):
    from fastapi.responses import RedirectResponse

    def fail(msg: str):
        return RedirectResponse(f"/?auth_error={msg}", status_code=307)

    if not auth_google.configured():
        return fail("google-not-configured")
    payload = auth.verify_payload(request.cookies.get(auth_google.VERIFIER_COOKIE))
    if not code or not payload or "v" not in payload:
        return fail("sign-in-expired-try-again")
    try:
        info = auth_google.exchange(code, payload["v"])
    except ValueError:
        return fail("google-exchange-failed")

    user = db.query(User).filter_by(email=info["email"]).first()
    if user is None:
        # Google-verified account: no local password (password login is refused
        # with a pointer to the Google button). Pre-assigned roles apply here
        # too - the "credentials before first sign-in" path works either way.
        user = User(email=info["email"], name=info["name"], password_hash="",
                    role=_initial_role(db, info["email"].strip().lower()))
        db.add(user)
        db.commit()

    resp = RedirectResponse("/", status_code=307)
    resp.delete_cookie(auth_google.VERIFIER_COOKIE)
    _set_session_cookie(resp, user)
    return resp


@app.post("/api/auth/logout")
def logout(response: Response):
    response.delete_cookie(auth.COOKIE_NAME)
    return {"signed_in": False}


# --------------------------------------------------------------- projects
def _visible_owners(user: dict | None) -> tuple[str, ...]:
    """Who a viewer may see projects for. Signed-in users see ONLY their own
    projects; anonymous viewers see the shared anonymous bucket
    (owner_user_id == ""). A signed-in user must never see other people's work,
    including the anonymous demo projects. (Signed-in users used to also get the
    anonymous bucket, which surfaced everyone's anonymous projects to any
    signed-in account - the privacy leak this fixes.)"""
    return (user["id"],) if user else ("",)


def _require_project_access(project: Project, user: dict | None) -> None:
    # Owned projects are gated to their owner. Anonymous projects (owner "")
    # stay open by design: they have no identity to gate by, and a visitor who
    # starts anonymously then signs in mid-session must keep working on the
    # project they just created. They simply never appear in a signed-in user's
    # project LIST (see _visible_owners) - that listing leak is what was fixed.
    if project.owner_user_id and (user is None or project.owner_user_id != user["id"]):
        raise HTTPException(403, "This project belongs to another account.")


@app.get("/api/projects", response_model=list[ProjectOut])
def list_projects(db: Session = Depends(get_db), user: dict | None = Depends(auth.get_current_user)):
    """Projects ordered by last activity (latest run, else creation) - the
    project a researcher wants is almost always the one they last worked on."""
    from sqlalchemy import func

    activity = {
        pid: (last, count)
        for pid, last, count in db.query(
            Job.project_id, func.max(Job.created_at), func.count(Job.id)
        )
        .group_by(Job.project_id)
        .all()
    }
    rows = db.query(Project).filter(Project.owner_user_id.in_(_visible_owners(user))).all()
    out = []
    for p in rows:
        last, count = activity.get(p.id, (None, 0))
        out.append(
            ProjectOut(
                id=p.id,
                name=p.name,
                description=p.description,
                created_at=p.created_at,
                last_activity_at=last or p.created_at,
                n_runs=count,
                archived=bool(p.archived),
            )
        )
    out.sort(key=lambda x: x.last_activity_at, reverse=True)
    return out


@app.post("/api/projects", response_model=ProjectOut, status_code=201)
def create_project(
    body: ProjectCreate,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    project = Project(
        name=body.name.strip(),
        description=body.description.strip(),
        owner_user_id=user["id"] if user else "",  # "" = anonymous (TTL purge applies)
    )
    db.add(project)
    db.commit()
    return ProjectOut(
        id=project.id,
        name=project.name,
        description=project.description,
        created_at=project.created_at,
        last_activity_at=project.created_at,
        n_runs=0,
        archived=False,
    )


@app.patch("/api/projects/{project_id}", response_model=ProjectOut)
def patch_project(
    project_id: str,
    body: ProjectPatch,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    """Archive/unarchive - reversible, no data loss. Archived projects collapse
    into the sidebar's Archived section and keep all datasets and runs."""
    project = _get_or_404(db, Project, project_id)
    _require_project_access(project, user)
    if body.archived is not None:
        project.archived = bool(body.archived)
    db.commit()
    return ProjectOut(
        id=project.id,
        name=project.name,
        description=project.description,
        created_at=project.created_at,
        last_activity_at=project.created_at,
        n_runs=0,
        archived=bool(project.archived),
    )


@app.delete("/api/projects/{project_id}", status_code=204)
def delete_project(
    project_id: str,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    """Permanent delete: removes the project, its datasets, runs, uploaded
    files, result files, and cached embeddings. Logged without retaining any
    uploaded text (design doc §9)."""
    import logging

    project = _get_or_404(db, Project, project_id)
    _require_project_access(project, user)
    counts = retention.delete_project_cascade(db, project)
    logging.getLogger("ccr.projects").info(
        "project deleted: id=%s name=%r corpora=%d runs=%d",
        project_id, project.name, counts["corpora"], counts["runs"],
    )
    return Response(status_code=204)


# ----------------------------------------------------------------- corpora
@app.get("/api/projects/{project_id}/corpora", response_model=list[CorpusOut])
def list_corpora(project_id: str, db: Session = Depends(get_db)):
    _get_or_404(db, Project, project_id)
    rows = (
        db.query(Corpus)
        .filter_by(project_id=project_id)
        .order_by(Corpus.created_at.desc())
        .all()
    )
    return [
        CorpusOut(
            id=c.id,
            project_id=c.project_id,
            filename=c.filename,
            n_rows=c.n_rows,
            columns=json.loads(c.columns_json),
            suggested_text_column=c.suggested_text_column or None,
            parse_info=json.loads(c.parse_info_json or "{}"),
            file_available=bool(c.path) and storage.exists(c.path),
            created_at=c.created_at,
        )
        for c in rows
    ]


@app.post("/api/projects/{project_id}/corpora", response_model=CorpusOut, status_code=201)
async def upload_corpus(
    project_id: str,
    file: UploadFile,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    project = _get_or_404(db, Project, project_id)
    _require_project_access(project, user)

    suffix = Path(file.filename or "upload.csv").suffix.lower()
    if suffix not in ALLOWED_SUFFIXES:
        raise HTTPException(400, f"Unsupported file type '{suffix}'. Use CSV or XLSX.")

    # Tier gate (design §5.1): anonymous users get strict caps; signing in
    # lifts them to the global ceiling.
    is_anon = user is None
    ceiling = min(max_upload_bytes(), auth.anon_max_bytes()) if is_anon else max_upload_bytes()

    # Assign the id NOW: the model's default fires at INSERT flush, so reading
    # corpus.id before commit yields None - every upload would then share one
    # "None.csv" on disk, each new upload silently clobbering the previous one.
    corpus = Corpus(
        id=uuid.uuid4().hex,
        project_id=project_id, filename=file.filename, path="", n_rows=0, columns_json="[]"
    )
    # Stream to a local temp file, parse it, then hand it to the storage backend
    # (local disk by default; S3/R2 when CCR_STORAGE=s3 in production).
    tmp_dir = DATA_DIR / "tmp"
    tmp_dir.mkdir(exist_ok=True)
    tmp = tmp_dir / f"{corpus.id}{suffix}"
    try:
        await _stream_to_temp(file, tmp, ceiling)
    except _UploadTooLarge:
        mb = ceiling // (1024 * 1024)
        if is_anon:
            raise HTTPException(
                413,
                f"Anonymous uploads are limited to {mb} MB. Sign in (top right) "
                "to upload larger files.",
            ) from None
        raise HTTPException(413, f"File exceeds the {mb} MB upload limit.") from None

    try:
        df, parse_info = load_corpus(str(tmp))
    except IngestError as exc:
        tmp.unlink(missing_ok=True)
        raise HTTPException(400, str(exc)) from exc

    if user is None and len(df) > auth.anon_max_rows():
        tmp.unlink(missing_ok=True)
        raise HTTPException(
            400,
            f"Anonymous uploads are limited to {auth.anon_max_rows():,} rows "
            f"(this file has {len(df):,}). Sign in (top right) to upload larger corpora.",
        )

    corpus.path = storage.move_local_into_storage("corpora", f"{corpus.id}{suffix}", tmp)

    corpus.n_rows = int(len(df))
    corpus.columns_json = json.dumps(list(df.columns))
    corpus.parse_info_json = json.dumps(parse_info)
    corpus.suggested_text_column = suggest_text_column(df) or ""
    db.add(corpus)
    db.commit()

    preview = json.loads(df.head(5).to_json(orient="records", force_ascii=False))
    return CorpusOut(
        id=corpus.id,
        project_id=project_id,
        filename=corpus.filename,
        n_rows=corpus.n_rows,
        columns=json.loads(corpus.columns_json),
        suggested_text_column=corpus.suggested_text_column or None,
        parse_info=parse_info,
        preview=preview,
        created_at=corpus.created_at,
    )


# -------------------------------------------------------------- constructs
def _latest_seed_versions(rows: list[Construct]) -> list[Construct]:
    """Collapse library constructs to the newest version of each slug.

    Versions are append-only, so a corrected construct ships as a new version
    alongside the old one (spec 0007). Only the newest belongs in the picker;
    superseded rows stay in the database and stay reachable by id, so runs,
    results, and reproduction scripts that used them still resolve. Custom
    constructs are not versioned and pass through untouched.
    """
    newest: dict[str, Construct] = {}
    for c in rows:
        if not (c.is_seed and c.construct_slug):
            continue
        best = newest.get(c.construct_slug)
        if best is None or (c.version or 1) > (best.version or 1):
            newest[c.construct_slug] = c
    keep = {id(c) for c in newest.values()}
    return [c for c in rows if not (c.is_seed and c.construct_slug) or id(c) in keep]


@app.get("/api/constructs", response_model=list[ConstructOut])
def list_constructs(db: Session = Depends(get_db)):
    rows = db.query(Construct).order_by(Construct.is_seed.desc(), Construct.name).all()
    return [_construct_out(c) for c in _latest_seed_versions(rows)]


@app.post("/api/constructs", response_model=ConstructOut, status_code=201)
def create_construct(body: ConstructCreate, db: Session = Depends(get_db)):
    items = [i.strip() for i in body.items if i.strip()]
    if not items:
        raise HTTPException(400, "Construct needs at least one non-empty item.")
    flags = body.reverse_scored or [False] * len(items)
    if len(flags) != len(items):
        raise HTTPException(400, "reverse_scored must have one flag per item.")
    construct = Construct(
        name=body.name.strip(),
        description=body.description.strip(),
        reference=body.reference.strip(),
        items_json=json.dumps(items),
        reverse_flags_json=json.dumps([bool(f) for f in flags]),
        is_seed=False,
        verification_status="draft",  # user-defined research tools, not validated scales
        language=(body.language or "en").lower(),
        # AI-generated drafts carry provenance (model, prompt version, date);
        # the label persists even after the researcher edits items.
        generation_json=json.dumps(body.generation.model_dump()) if body.generation else "",
    )
    db.add(construct)
    db.commit()
    return _construct_out(construct)


@app.post("/api/constructs/generate-items", response_model=GenerateItemsOut)
def generate_construct_items(
    body: GenerateItemsIn,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    """Draft candidate items for a construct with an LLM - a PREVIEW only
    (nothing saved; the researcher reviews/edits, then saves via POST
    /api/constructs). Signed-in only with a daily cap: generation spends real
    API money (PI decisions 2026-08-05; design note ITEM_GENERATION.md)."""
    if user is None:
        raise HTTPException(
            401, "Sign in (top right) to draft items with AI - accounts are free."
        )
    if not item_generation.configured():
        raise HTTPException(503, "Item generation is not configured on this instance.")

    cap = item_generation.user_max_generations_per_day()
    used = _generations_used_today(db, user["id"])
    if used >= cap:
        raise HTTPException(
            429,
            f"Daily limit reached ({cap} generations/day). It resets at midnight UTC - "
            "you can still add or edit items by hand.",
        )

    try:
        draft = item_generation.generate_items(
            name=body.name.strip(),
            description=body.description.strip(),
            n_items=body.n_items,
            language=(body.language or "en").strip() or "en",
        )
    except item_generation.GenerationError as exc:
        raise HTTPException(exc.status_code, str(exc)) from exc

    stamp = item_generation.generation_stamp()
    # Count only successful generations (failures cost ~nothing and shouldn't
    # burn quota); recorded before returning so the cap can't be raced past
    # its budget by much.
    db.add(GenerationEvent(user_id=user["id"], model=stamp["model"]))
    db.commit()

    return GenerateItemsOut(
        items=draft.items,
        notes=draft.notes,
        # items ride inside the stamp so the client echoes the full audit
        # record on save: what the AI drafted, distinct from what the
        # researcher edits and saves.
        generation=ConstructGeneration(**stamp, items=draft.items),
        generations_used_today=used + 1,
        max_generations_per_day=cap,
    )


@app.post("/api/constructs/parse-file")
async def parse_construct_upload(file: UploadFile):
    """Parse a CSV/XLSX of scale items into a PREVIEW (nothing is saved).
    The researcher reviews/edits, then saves via POST /api/constructs."""
    suffix = Path(file.filename or "items.csv").suffix.lower()
    if suffix not in ALLOWED_SUFFIXES:
        raise HTTPException(400, f"Unsupported file type '{suffix}'. Use CSV or XLSX.")
    payload = await file.read()
    if len(payload) > 1024 * 1024:
        raise HTTPException(413, "Item files are capped at 1 MB (a scale is a short list).")

    tmp_dir = DATA_DIR / "tmp"
    tmp_dir.mkdir(exist_ok=True)
    tmp = tmp_dir / f"construct_upload_{os.urandom(6).hex()}{suffix}"
    tmp.write_bytes(payload)
    try:
        parsed = parse_construct_file(str(tmp))
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    finally:
        tmp.unlink(missing_ok=True)  # item files are never retained

    stem = Path(file.filename or "").stem.replace("_", " ").replace("-", " ").strip()
    parsed["suggested_name"] = stem.title() if stem else ""
    return parsed


# -------------------------------------------------------------------- jobs
@app.post("/api/jobs", response_model=JobOut, status_code=201)
def create_job(
    body: JobCreate,
    request: Request,
    response: Response,
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    project = _get_or_404(db, Project, body.project_id)
    _require_project_access(project, user)
    corpus = _get_or_404(db, Corpus, body.corpus_id)

    # One run may score several constructs against the same corpus (the corpus
    # is embedded once, so N constructs cost barely more than one). A run
    # counts once toward the anonymous/saved-run limits regardless of N;
    # the per-run construct cap bounds output size, not compute.
    construct_ids = body.construct_ids if body.construct_ids else (
        [body.construct_id] if body.construct_id else []
    )
    if not construct_ids:
        raise HTTPException(400, "Provide construct_id or a non-empty construct_ids list.")
    if len(construct_ids) > MAX_CONSTRUCTS_PER_RUN:
        raise HTTPException(
            400, f"At most {MAX_CONSTRUCTS_PER_RUN} constructs per run (got {len(construct_ids)})."
        )
    if len(set(construct_ids)) != len(construct_ids):
        raise HTTPException(400, "Duplicate constructs in one run are not allowed.")
    for cid in construct_ids:
        _get_or_404(db, Construct, cid)

    # Anchor-vector (bipolar) run (spec 0006): a single target construct scored
    # against one opposite pole. Mutually exclusive with multi-construct.
    opposite_id = (body.opposite_construct_id or "").strip()
    metric = (body.similarity_metric or "cosine").strip().lower()
    if opposite_id:
        if len(construct_ids) != 1:
            raise HTTPException(
                400,
                "An anchored (bipolar) run scores exactly one target construct "
                "against one opposite pole. Provide a single target construct.",
            )
        if opposite_id == construct_ids[0]:
            raise HTTPException(400, "The contrasting construct must differ from the target.")
        if metric not in ("cosine", "dot"):
            raise HTTPException(
                400, f"similarity_metric must be 'cosine' or 'dot' (got '{body.similarity_metric}')."
            )
        _get_or_404(db, Construct, opposite_id)
    else:
        metric = ""  # only meaningful for anchored runs

    # Anonymous tier: N runs per day, then sign-in (PI decision 2026-07-10).
    # Cookie counter = a nudge, not a security boundary (recorded in DECISIONS.md).
    if user is None:
        used = auth.runs_used_today(request)
        if used >= auth.anon_max_runs_per_day():
            raise HTTPException(
                429,
                f"Anonymous limit reached ({auth.anon_max_runs_per_day()} runs/day). "
                "Sign in (top right) to keep running - accounts are free.",
            )
    else:
        # Signed-in tier: saved-run cap instead of deletion (their data, their
        # call). Lab members and above (admin-granted roles) are uncapped.
        row = db.get(User, user["id"])
        if not auth.role_unlimited(row.role if row else None) and (
            _saved_runs_used(db, user["id"]) >= auth.user_max_saved_runs()
        ):
            raise HTTPException(
                409,
                f"You have {auth.user_max_saved_runs()} saved runs (the maximum). "
                "Delete a project or old runs to start a new analysis.",
            )

    if body.text_column not in json.loads(corpus.columns_json):
        raise HTTPException(400, f"Column '{body.text_column}' not in corpus columns.")
    allowed = registry.known_ids() | {FAKE_MODEL_NAME}
    if body.model_name not in allowed:
        raise HTTPException(400, f"Unknown model '{body.model_name}'.")
    language = (body.language or "en").strip().lower()
    if not (2 <= len(language) <= 8 and language.replace("-", "").isalpha()):
        raise HTTPException(400, f"Invalid language code '{body.language}'.")

    # The raw file may be gone for two reasons: an anonymous upload was deleted
    # right after its analysis, or an ephemeral-disk restart wiped it. Either
    # way, past results are intact (DB) but a NEW run needs the file re-supplied.
    if not corpus.path or not storage.exists(corpus.path):
        if user is None:
            msg = ("This dataset's file was removed after analysis (anonymous uploads "
                   "are not kept). Upload the file again, or sign in to keep datasets.")
        else:
            msg = ("This dataset's file is no longer on the server, so it can't be "
                   "re-run. Your past results for it are safe. Upload the file again "
                   "to run a new analysis.")
        raise HTTPException(410, msg)

    job = Job(
        project_id=body.project_id,
        corpus_id=body.corpus_id,
        construct_id=construct_ids[0],
        construct_ids_json=json.dumps(construct_ids),
        opposite_construct_id=opposite_id,
        similarity_metric=metric,
        text_column=body.text_column,
        model_name=body.model_name,
        language=language,
    )
    db.add(job)
    db.commit()
    jobs_module.submit_job(job.id)

    if user is None:  # advance the daily counter only after the job is accepted
        response.set_cookie(
            auth.RUNS_COOKIE_NAME,
            auth.run_counter_token(auth.runs_used_today(request) + 1),
            httponly=True,
            samesite="lax",
            secure=auth.cookies_secure(),
            max_age=24 * 3600,
        )
    return _job_out(db, job)


@app.get("/api/jobs", response_model=list[JobOut])
def list_jobs(project_id: str, db: Session = Depends(get_db)):
    rows = (
        db.query(Job).filter_by(project_id=project_id).order_by(Job.created_at.desc()).all()
    )
    return [_job_out(db, j) for j in rows]


@app.get("/api/jobs/{job_id}", response_model=JobOut)
def get_job(job_id: str, db: Session = Depends(get_db)):
    return _job_out(db, _get_or_404(db, Job, job_id))


@app.get("/api/jobs/{job_id}/results")
def job_results(job_id: str, db: Session = Depends(get_db)):
    job = _get_or_404(db, Job, job_id)
    if job.status != "completed":
        raise HTTPException(409, f"Job status is '{job.status}', not completed.")
    return {
        "summary": json.loads(job.summary_json),
        "metadata": json.loads(job.metadata_json),
    }


@app.get("/api/jobs/{job_id}/export")
def export_results(job_id: str, db: Session = Depends(get_db)):
    job = _get_or_404(db, Job, job_id)
    if job.status != "completed" or not job.result_path:
        raise HTTPException(409, "Results not available.")
    if not storage.exists(job.result_path):
        raise HTTPException(
            410,
            "The results CSV file is no longer on the server. The summary and "
            "per-item loadings on the results page are still available; re-run the "
            "analysis to regenerate the downloadable CSV.",
        )
    filename = f"ccr_results_{job_id[:8]}.csv"
    if storage.is_s3(job.result_path):
        from fastapi.responses import StreamingResponse

        return StreamingResponse(
            storage.open_stream(job.result_path),
            media_type="text/csv",
            headers={"Content-Disposition": f'attachment; filename="{filename}"'},
        )
    return FileResponse(job.result_path, media_type="text/csv", filename=filename)


@app.get("/api/jobs/{job_id}/metadata")
def export_metadata(job_id: str, db: Session = Depends(get_db)):
    job = _get_or_404(db, Job, job_id)
    if job.status != "completed":
        raise HTTPException(409, "Metadata not available.")
    return JSONResponse(
        json.loads(job.metadata_json),
        headers={
            "Content-Disposition": f'attachment; filename="ccr_run_{job_id[:8]}.json"'
        },
    )


@app.get("/api/jobs/{job_id}/script")
def export_script(job_id: str, db: Session = Depends(get_db)):
    """Offline-runnable reproduction script generated from run metadata (spec 0002)."""
    job = _get_or_404(db, Job, job_id)
    if job.status != "completed":
        raise HTTPException(409, "Script not available until the run completes.")
    return PlainTextResponse(
        script_text(json.loads(job.metadata_json)),
        media_type="text/x-python",
        headers={
            "Content-Disposition": f'attachment; filename="{script_filename(job_id)}"'
        },
    )


@app.get("/api/jobs/{job_id}/script-requirements")
def export_script_requirements(job_id: str, db: Session = Depends(get_db)):
    job = _get_or_404(db, Job, job_id)
    if job.status != "completed":
        raise HTTPException(409, "Requirements not available until the run completes.")
    return PlainTextResponse(
        requirements_text(json.loads(job.metadata_json)),
        media_type="text/plain",
        headers={
            "Content-Disposition": f'attachment; filename="{requirements_filename(job_id)}"'
        },
    )


# ------------------------------------------------------------ guides + samples
# Two guides (split 2026-08-12):
#   /guide   - public how-to-use guide (public_guide.html): what CCR is, the
#              upload -> construct -> run -> export flow, the AI-drafting model,
#              and the reproducibility record. Anyone may read it.
#   /testing - the click-through TESTING guide (guide.html) for the PI/students,
#              with the synthetic demo corpora it references. Lab-only, same gate
#              as /product (PI: keep it, but don't show testers' scenarios to the
#              public).
# Both HTMLs live in app/ (not static/, which `npm run build` wipes); sample_data/
# sits at the repo root, same resolution as packages/ (= / in the container).
PUBLIC_GUIDE_HTML = Path(__file__).resolve().parent / "public_guide.html"
GUIDE_HTML = Path(__file__).resolve().parent / "guide.html"
SAMPLES_DIR = Path(__file__).resolve().parents[2] / "sample_data"

# Shared "this page is internal to the lab" body for the gated docs (/testing,
# /product). {what} names the specific doc so the message reads naturally.
LAB_FORBIDDEN_HTML = """<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>CCR Platform - Internal page</title>
<style>
  body {{ margin: 0; background: #f7f7f8; color: #1d2129;
         font: 16px/1.6 -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; }}
  main {{ max-width: 520px; margin: 14vh auto 0; padding: 0 1.25rem; text-align: center; }}
  h1 {{ font-size: 1.25rem; }}
  p {{ color: #667085; }}
  a {{ color: #26736f; font-weight: 600; }}
</style></head><body><main>
<h1>This page is internal to the lab</h1>
<p>{what} is limited to Culture &amp; Morality Lab
members. If you are in the lab, sign in on the
<a href="/">dashboard</a> with your lab account and come back; accounts are
granted lab access by the admins.</p>
<p>Looking for how to use the platform? The
<a href="/guide">public guide</a> is open to everyone.</p>
<p><a href="/">← Back to the CCR Platform</a></p>
</main></body></html>"""


@app.get("/guide", include_in_schema=False)
def public_guide():
    """Public how-to guide (no auth). Falls back to the testing guide only if
    the public one is somehow absent, so /guide is never a dead link."""
    if PUBLIC_GUIDE_HTML.exists():
        return FileResponse(PUBLIC_GUIDE_HTML, media_type="text/html")
    if GUIDE_HTML.exists():
        return FileResponse(GUIDE_HTML, media_type="text/html")
    raise HTTPException(404, "Guide not available on this instance.")


@app.get("/testing", include_in_schema=False)
def testing_guide(
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    """Lab-only testing guide (same gate as /product)."""
    row = db.get(User, user["id"]) if user else None
    if row is None or not auth.role_lab_or_above(row.role):
        return HTMLResponse(
            LAB_FORBIDDEN_HTML.format(what="The click-through testing guide"),
            status_code=403,
        )
    if not GUIDE_HTML.exists():
        raise HTTPException(404, "Testing guide not available on this instance.")
    return FileResponse(GUIDE_HTML, media_type="text/html")


# /product: the under-the-hood companion to the guides - access model (tiers,
# invites, pre-assignments, audit), architecture, data flow, retention.
# Internal-only since 2026-07-31 (PI request): lab members, maintainers, PI.
PRODUCT_HTML = Path(__file__).resolve().parent / "product.html"


@app.get("/product", include_in_schema=False)
def product_page(
    db: Session = Depends(get_db),
    user: dict | None = Depends(auth.get_current_user),
):
    row = db.get(User, user["id"]) if user else None
    if row is None or not auth.role_lab_or_above(row.role):
        return HTMLResponse(
            LAB_FORBIDDEN_HTML.format(what="The architecture and access-model docs"),
            status_code=403,
        )
    if not PRODUCT_HTML.exists():
        raise HTTPException(404, "Product page not available on this instance.")
    return FileResponse(PRODUCT_HTML, media_type="text/html")


@app.get("/admin", include_in_schema=False)
def admin_page():
    """Serve the SPA at /admin; the frontend renders the admin view there
    (and shows access-denied unless /api/auth/me says is_admin)."""
    index = Path(__file__).resolve().parent.parent / "static" / "index.html"
    if not index.exists():
        raise HTTPException(404, "UI not built.")
    return FileResponse(index, media_type="text/html")


@app.get("/welcome", include_in_schema=False)
def welcome_page():
    """Serve the SPA at /welcome; the frontend renders the landing page there
    (what CCR is, who runs the platform, links to /guide and /product).
    First-time visitors to / are shown it automatically."""
    index = Path(__file__).resolve().parent.parent / "static" / "index.html"
    if not index.exists():
        raise HTTPException(404, "UI not built.")
    return FileResponse(index, media_type="text/html")


if SAMPLES_DIR.exists():
    app.mount("/samples", StaticFiles(directory=SAMPLES_DIR), name="samples")

# ------------------------------------------------------------ static (SPA)
STATIC_DIR = Path(__file__).resolve().parent.parent / "static"
if STATIC_DIR.exists():
    app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")