Rewrite leftover HTTP routers except auth.

Project ingest, pipeline enqueue/SSE, reports, admin, contact, survey,
and feedback now resolve from src. Auth router is unchanged.

periscope/src/backend/routers/admin.py

@@ -0,0 +1,499 @@
+"""Admin library, users, usage, sweeper, and report overrides."""
+from __future__ import annotations
+
+import asyncio
+import re
+from datetime import datetime, timezone
+
+import httpx
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from backend.config import settings
+from backend.periscopex.utils import safe_mpn
+from backend.routers.deps import get_storage, resolve_or_404
+from backend.services import admin_settings as settings_svc
+from backend.services import projects as proj_svc
+from backend.services.billing_hook import get_billing
+
+router = APIRouter(prefix="/admin", tags=["admin"])
+_CLERK_USERS = "https://api.clerk.com/v1/users"
+
+
+async def is_admin(request: Request) -> bool:
+    cached = getattr(request.state, "_is_admin", None)
+    if cached is not None:
+        return cached
+
+    user_id: str = request.state.user_id
+    if not settings.use_auth:
+        request.state._is_admin = True
+        return True
+
+    if settings.use_local_auth:
+        from backend.services import local_users
+
+        user = local_users.get_user(user_id)
+        ok = bool(user and user.is_admin)
+        request.state._is_admin = ok
+        return ok
+
+    ok = False
+    try:
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(
+                f"{_CLERK_USERS}/{user_id}",
+                headers={"Authorization": f"Bearer {settings.clerk_secret_key}"},
+            )
+        if resp.status_code == 200:
+            ok = resp.json().get("public_metadata", {}).get("role") == "admin"
+    except Exception:
+        ok = False
+    request.state._is_admin = ok
+    return ok
+
+
+async def _require_admin(request: Request) -> str:
+    if not await is_admin(request):
+        raise HTTPException(403, "Admin access required")
+    return request.state.user_id
+
+
+def _safe_name(name: str) -> str:
+    safe = safe_mpn(name)
+    if ".." in safe or not re.match(r"^[A-Za-z0-9]", safe):
+        raise HTTPException(400, "Invalid component name")
+    return safe
+
+
+def _component_key(storage, component_type: str, safe: str) -> str:
+    if component_type == "ic":
+        return f"library/extracted/{safe}.json"
+    if component_type == "passive":
+        return f"library/patterns/{safe}.json"
+    if component_type == "simple":
+        key = f"library/passives/{safe}.json"
+        if storage.exists(key):
+            return key
+        return f"library/models/{safe}.json"
+    raise HTTPException(400, f"Unknown component type: {component_type}")
+
+
+@router.get("/components")
+async def list_components(request: Request):
+    await _require_admin(request)
+    catalog = proj_svc.list_library_catalog(get_storage(request))
+    return JSONResponse(
+        content={
+            "ics": catalog["ics"],
+            "passives": catalog["passives"],
+            "simple": catalog["simple"],
+        },
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+@router.get("/components/{component_type}/{name:path}")
+async def get_component(component_type: str, name: str, request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    key = _component_key(storage, component_type, _safe_name(name))
+    if not storage.exists(key):
+        raise HTTPException(404, f"Component not found: {name}")
+    return JSONResponse(content=storage.read_json(key))
+
+
+@router.delete("/components/{component_type}/{name:path}")
+async def delete_component(component_type: str, name: str, request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    key = _component_key(storage, component_type, _safe_name(name))
+    if not storage.exists(key):
+        raise HTTPException(404, f"Component not found: {name}")
+    storage.delete_key(key)
+    from backend.services.datasheet_store import delete_datasheet_ref
+
+    deleted_datasheets = 0
+    if delete_datasheet_ref(storage, name):
+        deleted_datasheets += 1
+    ds_key = f"library/datasheets/{_safe_name(name)}.pdf"
+    if storage.exists(ds_key):
+        storage.delete_key(ds_key)
+        deleted_datasheets += 1
+    return {"deleted": key, "deleted_datasheets": deleted_datasheets}
+
+
+def _clerk_profile_fields(clerk: dict) -> dict:
+    first = clerk.get("first_name") or ""
+    last = clerk.get("last_name") or ""
+    emails = clerk.get("email_addresses") or []
+    return {
+        "name": f"{first} {last}".strip() or None,
+        "email": emails[0].get("email_address") if emails else None,
+        "image_url": clerk.get("image_url"),
+    }
+
+
+def _base_admin_user(storage, uid: str) -> dict:
+    try:
+        project_count = len(proj_svc.list_projects(storage, uid))
+    except Exception:
+        project_count = 0
+    try:
+        balance = get_billing().get_balance(storage, uid)
+    except Exception:
+        balance = 0.0
+    return {
+        "user_id": uid,
+        "project_count": project_count,
+        "balance": round(balance, 4),
+        "name": None,
+        "email": None,
+        "image_url": None,
+    }
+
+
+def _user_ids_from_prefix(storage) -> set[str]:
+    ids: set[str] = set()
+    for entry in storage.list_prefix("users/"):
+        parts = entry.split("/")
+        if len(parts) >= 2 and parts[1]:
+            ids.add(parts[1])
+    return ids
+
+
+async def _enrich_clerk_profiles(users: dict[str, dict]) -> None:
+    sem = asyncio.Semaphore(10)
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+
+        async def _one(uid: str) -> None:
+            async with sem:
+                try:
+                    resp = await client.get(
+                        f"{_CLERK_USERS}/{uid}",
+                        headers={"Authorization": f"Bearer {settings.clerk_secret_key}"},
+                    )
+                    if resp.status_code == 200:
+                        users[uid].update(_clerk_profile_fields(resp.json()))
+                except Exception:
+                    return
+
+        await asyncio.gather(*(_one(uid) for uid in users))
+
+
+@router.get("/users")
+async def list_users(request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    user_ids = _user_ids_from_prefix(storage)
+    user_ids.update(get_billing().list_user_ids(storage))
+    users = {uid: _base_admin_user(storage, uid) for uid in user_ids}
+    if settings.use_auth and users:
+        await _enrich_clerk_profiles(users)
+    return list(users.values())
+
+
+@router.get("/users/search")
+async def search_users(request: Request, email: str):
+    await _require_admin(request)
+    storage = get_storage(request)
+    email = email.strip()
+    if not email:
+        return []
+    if not settings.use_auth:
+        raise HTTPException(400, "User search requires authentication to be enabled")
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        try:
+            resp = await client.get(
+                _CLERK_USERS,
+                params={"email_address": [email]},
+                headers={"Authorization": f"Bearer {settings.clerk_secret_key}"},
+            )
+        except Exception as exc:
+            raise HTTPException(502, "Failed to look up user") from exc
+    if resp.status_code != 200:
+        raise HTTPException(502, "Failed to look up user")
+    results: list[dict] = []
+    for clerk in resp.json():
+        uid = clerk.get("id")
+        if not uid:
+            continue
+        entry = _base_admin_user(storage, uid)
+        entry.update(_clerk_profile_fields(clerk))
+        results.append(entry)
+    return results
+
+
+@router.get("/usage")
+async def get_usage(request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    seen: set[str] = set()
+    user_rows: list[dict] = []
+    grand_total = 0.0
+    for uid in _user_ids_from_prefix(storage):
+        if uid in seen:
+            continue
+        seen.add(uid)
+        projects = proj_svc.list_projects(storage, uid)
+        user_cost = 0.0
+        details = []
+        for p in projects:
+            cost = p.total_cost_usd or 0.0
+            user_cost += cost
+            details.append(
+                {
+                    "id": p.id,
+                    "name": p.name,
+                    "status": p.status,
+                    "cost_usd": cost,
+                    "created": p.created,
+                }
+            )
+        user_rows.append(
+            {
+                "user_id": uid,
+                "project_count": len(projects),
+                "total_cost_usd": round(user_cost, 4),
+                "projects": details,
+                "name": None,
+                "email": None,
+            }
+        )
+        grand_total += user_cost
+    if settings.use_auth and user_rows:
+        async with httpx.AsyncClient() as client:
+            for row in user_rows:
+                try:
+                    resp = await client.get(
+                        f"{_CLERK_USERS}/{row['user_id']}",
+                        headers={"Authorization": f"Bearer {settings.clerk_secret_key}"},
+                    )
+                    if resp.status_code == 200:
+                        clerk = resp.json()
+                        first = clerk.get("first_name") or ""
+                        last = clerk.get("last_name") or ""
+                        row["name"] = f"{first} {last}".strip() or None
+                        emails = clerk.get("email_addresses") or []
+                        row["email"] = emails[0].get("email_address") if emails else None
+                except Exception:
+                    continue
+    return {"grand_total_usd": round(grand_total, 4), "users": user_rows}
+
+
+async def _enrich_with_clerk_info(
+    items: list[dict],
+    uid_key: str = "user_id",
+    name_key: str = "owner_name",
+    email_key: str = "owner_email",
+) -> None:
+    if not settings.use_auth or not items:
+        return
+    cache: dict[str, dict] = {}
+    async with httpx.AsyncClient() as client:
+        for item in items:
+            uid = item[uid_key]
+            if uid not in cache:
+                try:
+                    resp = await client.get(
+                        f"{_CLERK_USERS}/{uid}",
+                        headers={"Authorization": f"Bearer {settings.clerk_secret_key}"},
+                    )
+                    if resp.status_code == 200:
+                        clerk = resp.json()
+                        first = clerk.get("first_name") or ""
+                        last = clerk.get("last_name") or ""
+                        emails = clerk.get("email_addresses") or []
+                        cache[uid] = {
+                            name_key: f"{first} {last}".strip() or None,
+                            email_key: emails[0].get("email_address") if emails else None,
+                        }
+                    else:
+                        cache[uid] = {name_key: None, email_key: None}
+                except Exception:
+                    cache[uid] = {name_key: None, email_key: None}
+            item.update(cache[uid])
+
+
+@router.get("/projects")
+async def list_all_projects(request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    all_projects: list[dict] = []
+    for uid in _user_ids_from_prefix(storage):
+        for p in proj_svc.list_projects(storage, uid):
+            all_projects.append(
+                {
+                    "id": p.id,
+                    "name": p.name,
+                    "user_id": p.user_id,
+                    "status": p.status,
+                    "created": p.created,
+                    "updated": p.updated,
+                    "has_bom": p.has_bom,
+                    "has_netlist": p.has_netlist,
+                    "datasheet_count": p.datasheet_count,
+                    "total_cost_usd": p.total_cost_usd,
+                    "pipeline_state": p.pipeline_state,
+                    "summary": p.summary,
+                    "owner_name": None,
+                    "owner_email": None,
+                }
+            )
+    await _enrich_with_clerk_info(all_projects)
+    return all_projects
+
+
+@router.get("/runs")
+async def list_running_pipelines(request: Request):
+    from backend.services import job_runner
+
+    await _require_admin(request)
+    storage = get_storage(request)
+    now = datetime.now(timezone.utc)
+    runs: list[dict] = []
+    for uid in _user_ids_from_prefix(storage):
+        prefix = f"users/{uid}/projects/"
+        for proj_entry in storage.list_prefix(prefix):
+            meta_key = (
+                proj_entry
+                if proj_entry.endswith("/project.json")
+                else f"{proj_entry}/project.json"
+            )
+            if not storage.exists(meta_key):
+                continue
+            try:
+                meta = proj_svc.ProjectMeta.model_validate(storage.read_json(meta_key))
+            except Exception:
+                continue
+            if meta.status not in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING):
+                continue
+            if proj_svc.heal_if_pipeline_finished(storage, uid, meta.id) is not None:
+                continue
+            exec_state = "unknown"
+            if meta.execution_name:
+                exec_state = job_runner.get_execution_state(meta.execution_name)
+            elif not job_runner.use_cloud_run_jobs():
+                exec_state = job_runner.get_execution_state(f"local/projects/{meta.id}")
+            if exec_state in ("succeeded", "failed", "cancelled"):
+                try:
+                    last_update = datetime.fromisoformat(meta.updated)
+                    age = (now - last_update).total_seconds()
+                except Exception:
+                    age = settings.pipeline_sweeper_stale_seconds + 1
+                if age >= settings.pipeline_sweeper_stale_seconds:
+                    proj_svc.mark_stale_running(
+                        storage,
+                        uid,
+                        meta.id,
+                        f"Worker terminated (execution state={exec_state}); please restart.",
+                    )
+                    continue
+            try:
+                started_at = datetime.fromisoformat(meta.updated)
+            except Exception:
+                started_at = now
+            runs.append(
+                {
+                    "project_id": meta.id,
+                    "project_name": meta.name,
+                    "user_id": uid,
+                    "status": meta.status,
+                    "execution_name": meta.execution_name,
+                    "execution_state": exec_state,
+                    "started_at": started_at.isoformat(),
+                    "duration_seconds": int((now - started_at).total_seconds()),
+                    "owner_name": None,
+                    "owner_email": None,
+                }
+            )
+    await _enrich_with_clerk_info(runs)
+    return runs
+
+
+class UpdateMinVersionRequest(BaseModel):
+    min_model_version: str
+
+
+@router.get("/settings")
+async def get_settings(request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    data = settings_svc.get_admin_settings(storage)
+    data["default_model_version"] = settings.get_default_model_version()
+    return data
+
+
+@router.put("/settings/min-model-version")
+async def set_min_model_version(req: UpdateMinVersionRequest, request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    try:
+        settings_svc.set_min_model_version(storage, req.min_model_version)
+    except Exception as e:
+        raise HTTPException(400, f"Invalid version: {e}")
+    return {"min_model_version": req.min_model_version}
+
+
+class TestEmailRequest(BaseModel):
+    to_email: str
+
+
+@router.post("/test-email")
+async def test_email(req: TestEmailRequest, request: Request):
+    await _require_admin(request)
+    from backend.services.email import send_test_email
+
+    return await send_test_email(req.to_email)
+
+
+@router.post("/projects/{project_id}/mark-complete")
+async def mark_project_complete(project_id: str, request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED):
+        raise HTTPException(409, "Cannot mark a running pipeline complete; cancel it first")
+    if meta.status == "complete":
+        return {"status": "complete", "project_id": project_id}
+    proj_svc.update_project(
+        storage,
+        owner,
+        project_id,
+        status="complete",
+        pause_checkpoint=None,
+        pause_reason=None,
+    )
+    return {"status": "complete", "project_id": project_id}
+
+
+@router.delete("/projects/{project_id}/findings/{finding_id}")
+async def delete_finding(project_id: str, finding_id: str, request: Request):
+    await _require_admin(request)
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{proj_svc.project_prefix(owner, project_id)}/report.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Report not found")
+    report = storage.read_json(key)
+    findings = report.get("findings", []) or []
+    remaining = [f for f in findings if f.get("finding_id") != finding_id]
+    if len(remaining) == len(findings):
+        raise HTTPException(404, f"Finding not found: {finding_id}")
+    summary = {"total": len(remaining), "ERROR": 0, "WARNING": 0, "INFO": 0}
+    for f in remaining:
+        status = f.get("status")
+        if status in summary:
+            summary[status] += 1
+    report["findings"] = remaining
+    report["summary"] = summary
+    storage.write_json(key, report)
+    proj_svc.update_project(storage, owner, project_id, summary=summary)
+    return {
+        "deleted": finding_id,
+        "project_id": project_id,
+        "remaining": len(remaining),
+        "summary": summary,
+    }

periscope/src/backend/routers/contact.py

@@ -0,0 +1,118 @@
+"""Unauthenticated contact mailbox. Honeypot and per-IP delay live here."""
+from __future__ import annotations
+
+import html
+import logging
+import time
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+from fastapi import APIRouter, Request
+from pydantic import BaseModel, EmailStr, Field
+
+from backend.config import settings
+from backend.services.email import _send_raw
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+_COOLDOWN_S = 60
+_seen: dict[str, float] = {}
+
+
+class ContactRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=200)
+    email: EmailStr = Field(..., max_length=254)
+    message: str = Field(..., min_length=1, max_length=5000)
+    company: str = Field("", max_length=200)
+    subject: str = Field("", max_length=200)
+    honeypot: str = Field("", alias="_honey")
+
+
+class ContactResponse(BaseModel):
+    success: bool
+    message: str
+
+
+def _client_ip(request: Request) -> str:
+    forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
+    if forwarded:
+        return forwarded
+    return request.client.host if request.client else "unknown"
+
+
+def _compose(data: ContactRequest) -> MIMEMultipart:
+    msg = MIMEMultipart("alternative")
+    msg["From"] = f"Periscope <{settings.email_sender}>"
+    msg["To"] = settings.contact_recipient
+    msg["Reply-To"] = data.email
+    title = data.subject or "New message"
+    msg["Subject"] = f"[Periscope Contact] {title} from {data.name}"
+
+    plain = [f"Name: {data.name}", f"Email: {data.email}"]
+    if data.company:
+        plain.append(f"Company: {data.company}")
+    if data.subject:
+        plain.append(f"Subject: {data.subject}")
+    plain.extend(["", data.message, "", "— Sent from the Periscope contact form"])
+    msg.attach(MIMEText("\n".join(plain), "plain"))
+
+    def cell(label: str, value: str) -> str:
+        return (
+            "<tr>"
+            f'<td style="padding: 8px 12px; border: 1px solid #e5e5e5; font-weight: 600; width: 100px;">{html.escape(label)}</td>'
+            f'<td style="padding: 8px 12px; border: 1px solid #e5e5e5;">{value}</td>'
+            "</tr>"
+        )
+
+    rows = cell("Name", html.escape(data.name))
+    email = html.escape(data.email)
+    rows += (
+        "<tr>"
+        '<td style="padding: 8px 12px; border: 1px solid #e5e5e5; font-weight: 600;">Email</td>'
+        f'<td style="padding: 8px 12px; border: 1px solid #e5e5e5;"><a href="mailto:{email}">{email}</a></td>'
+        "</tr>"
+    )
+    if data.company:
+        rows += cell("Company", html.escape(data.company))
+    if data.subject:
+        rows += cell("Subject", html.escape(data.subject))
+    body = html.escape(data.message)
+    html_body = f"""\
+    <div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 560px; margin: 0 auto; padding: 24px;">
+      <h2 style="font-size: 18px; margin: 0 0 16px;">New contact form submission</h2>
+      <table style="width: 100%; border-collapse: collapse; font-size: 14px;">{rows}</table>
+      <div style="margin-top: 16px; padding: 16px; background: #f9fafb; border-radius: 8px; font-size: 14px; line-height: 1.6; white-space: pre-wrap;">{body}</div>
+      <p style="margin-top: 24px; font-size: 12px; color: #888;">Sent from the Periscope contact form</p>
+    </div>"""
+    msg.attach(MIMEText(html_body, "html"))
+    return msg
+
+
+@router.post("/contact", response_model=ContactResponse)
+async def submit_contact(data: ContactRequest, request: Request):
+    ok = "Message sent! We'll get back to you soon."
+    if data.honeypot:
+        return ContactResponse(success=True, message=ok)
+
+    ip = _client_ip(request)
+    now = time.time()
+    last = _seen.get(ip)
+    if last is not None and now - last < _COOLDOWN_S:
+        return ContactResponse(
+            success=False, message="Please wait a minute before submitting again."
+        )
+    _seen[ip] = now
+    if len(_seen) > 1000:
+        stale = now - _COOLDOWN_S
+        for key in [k for k, ts in _seen.items() if ts < stale]:
+            del _seen[key]
+
+    if not settings.use_email or not settings.contact_recipient:
+        logger.warning("Contact form submitted but email is not configured")
+        return ContactResponse(
+            success=False, message="Email is not configured on this server."
+        )
+
+    await _send_raw(settings.contact_recipient, _compose(data), "Contact form")
+    return ContactResponse(success=True, message=ok)

periscope/src/backend/routers/deps.py

@@ -0,0 +1,33 @@
+"""Request helpers shared by HTTP routers."""
+from __future__ import annotations
+
+from fastapi import HTTPException, Request
+
+from backend.services import projects as proj_svc
+from backend.services.storage import StorageBackend
+
+
+def get_storage(request: Request) -> StorageBackend:
+    return request.app.state.storage
+
+
+def get_user_id(request: Request) -> str:
+    return request.state.user_id
+
+
+async def resolve_or_404(request: Request, project_id: str) -> tuple[str, proj_svc.ProjectMeta]:
+    """Owner, collaborator, or admin; otherwise 404 (never 403 for missing projects)."""
+    storage = get_storage(request)
+    user_id = get_user_id(request)
+    hit = proj_svc.resolve_project_access(storage, user_id, project_id)
+    if hit:
+        return hit
+
+    from backend.routers.admin import is_admin
+
+    if await is_admin(request):
+        found = proj_svc.find_project_any_user(storage, project_id)
+        if found:
+            return found
+
+    raise HTTPException(404, "Project not found")

periscope/src/backend/routers/feedback.py

@@ -0,0 +1,250 @@
+"""Feedback tickets: per-user JSON plus JSONL indexes for listing."""
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Literal
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from backend.routers.deps import get_storage, get_user_id
+from backend.services.storage import StorageBackend
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+_TICKETS = "admin/feedback/tickets/"
+_BY_USER = "admin/feedback/index/by_user/"
+_BY_PROJECT = "admin/feedback/index/by_project/"
+_ALL = "admin/feedback/index/all.jsonl"
+
+FeedbackType = Literal["bug", "rule_feedback", "feature_request"]
+FeedbackStatus = Literal["open", "acknowledged", "resolved"]
+
+
+def _ticket_key(ticket_id: str) -> str:
+    return f"{_TICKETS}{ticket_id}.json"
+
+
+def _user_index_key(user_id: str) -> str:
+    return f"{_BY_USER}{user_id}.jsonl"
+
+
+def _project_index_key(project_id: str) -> str:
+    return f"{_BY_PROJECT}{project_id}.jsonl"
+
+
+class FeedbackTicket(BaseModel):
+    ticket_id: str
+    user_id: str
+    user_name: str | None = None
+    user_email: str | None = None
+    project_id: str | None = None
+    project_name: str | None = None
+    type: FeedbackType
+    status: FeedbackStatus = "open"
+    finding_id: str | None = None
+    finding_text: str | None = None
+    finding_designator: str | None = None
+    finding_mpn: str | None = None
+    finding_status: str | None = None
+    message: str
+    admin_notes: str | None = None
+    created_at: str
+    updated_at: str
+
+
+class CreateFeedbackRequest(BaseModel):
+    type: FeedbackType
+    message: str = Field(..., min_length=1, max_length=5000)
+    project_id: str | None = None
+    project_name: str | None = None
+    user_name: str | None = None
+    user_email: str | None = None
+    finding_id: str | None = None
+    finding_text: str | None = None
+    finding_designator: str | None = None
+    finding_mpn: str | None = None
+    finding_status: str | None = None
+
+
+class UpdateFeedbackRequest(BaseModel):
+    status: FeedbackStatus | None = None
+    admin_notes: str | None = None
+
+
+def _append_index(storage: StorageBackend, key: str, entry: dict) -> None:
+    prior = storage.read_text(key) if storage.exists(key) else ""
+    storage.write_text(key, prior + json.dumps(entry) + "\n")
+
+
+def _read_index(storage: StorageBackend, key: str) -> list[dict]:
+    if not storage.exists(key):
+        return []
+    rows: list[dict] = []
+    for line in storage.read_text(key).splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            rows.append(json.loads(line))
+        except json.JSONDecodeError:
+            logger.warning("Skipping malformed feedback index line in %s", key)
+    return rows
+
+
+def _read_ticket(storage: StorageBackend, ticket_id: str) -> FeedbackTicket | None:
+    key = _ticket_key(ticket_id)
+    if not storage.exists(key):
+        return None
+    try:
+        return FeedbackTicket(**storage.read_json(key))
+    except Exception:
+        logger.warning("Failed to read ticket %s", ticket_id)
+        return None
+
+
+def _tickets_from_index(
+    storage: StorageBackend,
+    index_key: str,
+    *,
+    status: str | None = None,
+    ticket_type: str | None = None,
+    project_id: str | None = None,
+) -> list[FeedbackTicket]:
+    out: list[FeedbackTicket] = []
+    for entry in reversed(_read_index(storage, index_key)):
+        tid = entry.get("ticket_id")
+        if not tid:
+            continue
+        ticket = _read_ticket(storage, tid)
+        if ticket is None:
+            continue
+        if status and ticket.status != status:
+            continue
+        if ticket_type and ticket.type != ticket_type:
+            continue
+        if project_id and ticket.project_id != project_id:
+            continue
+        out.append(ticket)
+    return out
+
+
+@router.post("/feedback", response_model=FeedbackTicket)
+async def create_feedback(body: CreateFeedbackRequest, request: Request):
+    storage = get_storage(request)
+    user_id = get_user_id(request)
+    now = datetime.now(timezone.utc).isoformat()
+    ticket_id = uuid.uuid4().hex[:12]
+    ticket = FeedbackTicket(
+        ticket_id=ticket_id,
+        user_id=user_id,
+        user_name=body.user_name,
+        user_email=body.user_email,
+        project_id=body.project_id,
+        project_name=body.project_name,
+        type=body.type,
+        status="open",
+        finding_id=body.finding_id,
+        finding_text=body.finding_text,
+        finding_designator=body.finding_designator,
+        finding_mpn=body.finding_mpn,
+        finding_status=body.finding_status,
+        message=body.message,
+        admin_notes=None,
+        created_at=now,
+        updated_at=now,
+    )
+    storage.write_json(_ticket_key(ticket_id), ticket.model_dump())
+    index = {"ticket_id": ticket_id, "created_at": now}
+    _append_index(storage, _user_index_key(user_id), index)
+    _append_index(storage, _ALL, index)
+    if body.project_id:
+        _append_index(storage, _project_index_key(body.project_id), index)
+    try:
+        from backend.services.email import send_feedback_received_email
+
+        await send_feedback_received_email(
+            ticket_id=ticket_id,
+            user_id=user_id,
+            feedback_type=body.type,
+            message=body.message,
+            submitter_name=body.user_name,
+            submitter_email=body.user_email,
+            project_name=body.project_name,
+            project_id=body.project_id,
+            finding_designator=body.finding_designator,
+            finding_mpn=body.finding_mpn,
+            finding_status=body.finding_status,
+            finding_text=body.finding_text,
+        )
+    except Exception:
+        logger.exception("Failed to enqueue feedback-received email for %s", ticket_id)
+    return ticket
+
+
+@router.get("/feedback", response_model=list[FeedbackTicket])
+async def list_my_feedback(request: Request, status: str | None = None):
+    return _tickets_from_index(
+        get_storage(request), _user_index_key(get_user_id(request)), status=status
+    )
+
+
+@router.get("/admin/feedback", response_model=list[FeedbackTicket])
+async def list_all_feedback(
+    request: Request,
+    status: str | None = None,
+    type: str | None = None,
+    project_id: str | None = None,
+):
+    from backend.routers.admin import _require_admin
+
+    await _require_admin(request)
+    return _tickets_from_index(
+        get_storage(request),
+        _ALL,
+        status=status,
+        ticket_type=type,
+        project_id=project_id,
+    )
+
+
+@router.put("/admin/feedback/{ticket_id}", response_model=FeedbackTicket)
+async def update_feedback(ticket_id: str, body: UpdateFeedbackRequest, request: Request):
+    from backend.routers.admin import _require_admin
+
+    await _require_admin(request)
+    storage = get_storage(request)
+    ticket = _read_ticket(storage, ticket_id)
+    if ticket is None:
+        raise HTTPException(404, "Ticket not found")
+    previous = (ticket.admin_notes or "").strip()
+    if body.status is not None:
+        ticket.status = body.status
+    if body.admin_notes is not None:
+        ticket.admin_notes = body.admin_notes
+    ticket.updated_at = datetime.now(timezone.utc).isoformat()
+    storage.write_json(_ticket_key(ticket_id), ticket.model_dump())
+    notes = (ticket.admin_notes or "").strip()
+    if notes and notes != previous:
+        try:
+            from backend.services.email import send_feedback_reply_email
+
+            await send_feedback_reply_email(
+                user_id=ticket.user_id,
+                reply_text=notes,
+                original_message=ticket.message,
+                recipient_name=ticket.user_name,
+                recipient_email=ticket.user_email,
+                project_name=ticket.project_name,
+                finding_designator=ticket.finding_designator,
+                finding_mpn=ticket.finding_mpn,
+            )
+        except Exception:
+            logger.exception(
+                "Failed to enqueue feedback-reply email for ticket %s", ticket_id
+            )
+    return ticket

periscope/src/backend/routers/pipeline.py

@@ -0,0 +1,783 @@
+"""HTTP surface for analysis, placement, and PCB jobs.
+
+The API process never runs extract/review itself: it CAS-transitions
+project meta, asks ``job_runner`` to spawn a worker, and SSE-tails the
+storage event log (plus the in-memory ``job_workspace`` broker).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from collections.abc import AsyncIterator, Callable
+from typing import Any, Literal
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+from sse_starlette.sse import EventSourceResponse
+
+from backend.routers.deps import get_storage, get_user_id, resolve_or_404
+from backend.services import event_bridge, job_runner, job_workspace
+from backend.services import projects as proj_svc
+from backend.services.billing_hook import get_billing
+from backend.services.cost_estimator import estimate_pipeline_cost
+from backend.services.pcb_pipeline import (
+    analysis_busy as pcb_analysis_busy,
+    pcb_busy,
+    pcb_sse_terminal_from_status,
+    placement_busy as pcb_placement_busy,
+)
+from backend.services.placement_pipeline import (
+    analysis_busy as placement_analysis_busy,
+    placement_busy,
+)
+
+log = logging.getLogger(__name__)
+
+router = APIRouter(tags=["pipeline"])
+
+_REGEN_STAGES = {"derating"}
+_ANALYSIS_IDLE = frozenset({
+    proj_svc.STATUS_DRAFT,
+    proj_svc.STATUS_COMPLETE,
+    proj_svc.STATUS_ERROR,
+    proj_svc.STATUS_CANCELLED,
+})
+_REPROCESS_IDLE = frozenset({
+    proj_svc.STATUS_COMPLETE,
+    proj_svc.STATUS_ERROR,
+    proj_svc.STATUS_CANCELLED,
+})
+_WORKER_DONE = frozenset({"succeeded", "failed", "cancelled"})
+_ANALYSIS_SSE_DONE = frozenset({
+    "pipeline_complete",
+    "pipeline_error",
+    "pipeline_cancelled",
+    "pipeline_paused",
+})
+_PLACEMENT_IDLE = frozenset({"draft", "complete", "error", "cancelled"})
+_PLACEMENT_SSE_DONE = frozenset({
+    "placement_complete",
+    "placement_error",
+    "placement_cancelled",
+})
+_PCB_IDLE = frozenset({"draft", "complete", "error", "cancelled"})
+_PCB_SSE_DONE = frozenset({
+    "pcb_complete",
+    "pcb_error",
+    "pcb_cancelled",
+})
+_SSE_HEADERS = {
+    "Cache-Control": "no-cache, no-transform",
+    "X-Accel-Buffering": "no",
+    "Connection": "keep-alive",
+}
+_ENQUEUE_FAIL = "Failed to enqueue pipeline worker; please retry"
+
+
+class RegenRequest(BaseModel):
+    stages: list[str]
+
+
+class ReprocessRequest(BaseModel):
+    mode: Literal["failed", "all"] = "failed"
+
+
+def _project_active(meta: proj_svc.ProjectMeta) -> bool:
+    """True while analysis is queued or running (meta only; no Cloud Run poll)."""
+    return meta.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING)
+
+
+def _need_bom_netlist(meta: proj_svc.ProjectMeta, detail: str) -> None:
+    if not meta.has_bom or not meta.has_netlist:
+        raise HTTPException(400, detail)
+
+
+def _reset_event_history(storage, owner: str, project_id: str) -> None:
+    """Drop GCS log objects and any in-process broker history for this project."""
+    try:
+        event_bridge.GCSEventBroker(storage, owner).clear_history(project_id)
+    except Exception:
+        log.exception("event log clear failed for %s", project_id)
+    job_workspace.broker.clear_history(project_id)
+
+
+def _record_exec(storage, owner: str, project_id: str, execution_name: str, **extra: Any) -> None:
+    proj_svc.update_project(
+        storage, owner, project_id, execution_name=execution_name, **extra,
+    )
+
+
+def _enqueue_or_503(
+    storage,
+    owner: str,
+    project_id: str,
+    *,
+    enqueue: Callable[[], str],
+    fail_status: dict[str, Any],
+    http_detail: str,
+) -> str:
+    try:
+        return enqueue()
+    except Exception:
+        log.exception("worker enqueue failed for %s", project_id)
+        proj_svc.update_project(storage, owner, project_id, **fail_status)
+        raise HTTPException(503, http_detail)
+
+
+def _cas_queued(
+    storage,
+    owner: str,
+    project_id: str,
+    *,
+    from_status: frozenset[str],
+    conflict: str,
+    **fields: Any,
+) -> None:
+    try:
+        proj_svc.transition_status(
+            storage, owner, project_id,
+            from_status=from_status,
+            to_status=proj_svc.STATUS_QUEUED,
+            **fields,
+        )
+    except proj_svc.StatusConflict:
+        raise HTTPException(409, conflict)
+
+
+async def _await_terminal(
+    storage, user_id: str, project_id: str, *, timeout_s: float,
+) -> None:
+    """Poll until analysis meta is terminal, or ``timeout_s`` elapses."""
+    step = 0.5
+    spent = 0.0
+    while spent < timeout_s:
+        try:
+            meta = proj_svc.get_project(storage, user_id, project_id)
+        except Exception:
+            meta = None
+        if meta is None or meta.status in proj_svc.TERMINAL_STATUSES:
+            return
+        await asyncio.sleep(step)
+        spent += step
+
+
+async def _interrupt_active_pipeline(storage, user_id: str, project_id: str) -> None:
+    """Soft-cancel, then hard-cancel, then force-cancel leftover running meta."""
+    proj_svc.request_cancel(storage, user_id, project_id)
+    await _await_terminal(storage, user_id, project_id, timeout_s=10.0)
+    meta = proj_svc.get_project(storage, user_id, project_id)
+    if meta is None:
+        return
+    if _project_active(meta) and meta.execution_name:
+        job_runner.cancel_execution(meta.execution_name)
+        await _await_terminal(storage, user_id, project_id, timeout_s=5.0)
+        meta = proj_svc.get_project(storage, user_id, project_id) or meta
+    if _project_active(meta):
+        try:
+            proj_svc.transition_status(
+                storage, user_id, project_id,
+                from_status={proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED},
+                to_status=proj_svc.STATUS_CANCELLED,
+                pipeline_state={"error": "Superseded by reprocess"},
+                cancel_requested=False,
+            )
+        except proj_svc.StatusConflict:
+            pass
+
+
+async def _stop_then_wait(storage, owner: str, project_id: str, meta: proj_svc.ProjectMeta) -> None:
+    if not _project_active(meta):
+        return
+    proj_svc.request_cancel(storage, owner, project_id)
+    await _await_terminal(storage, owner, project_id, timeout_s=10.0)
+    latest = proj_svc.get_project(storage, owner, project_id) or meta
+    if _project_active(latest) and latest.execution_name:
+        job_runner.cancel_execution(latest.execution_name)
+        await _await_terminal(storage, owner, project_id, timeout_s=5.0)
+
+
+def _sse(gen: AsyncIterator[dict]) -> EventSourceResponse:
+    return EventSourceResponse(gen, ping=15, headers=_SSE_HEADERS)
+
+
+def _yield_sse(ev: str, data: dict) -> dict:
+    return {"event": ev, "data": json.dumps(data)}
+
+
+async def _poll_until_crash(
+    storage,
+    owner: str,
+    project_id: str,
+    *,
+    execution_name: str | None,
+    live: Callable[[proj_svc.ProjectMeta], bool],
+    finished: Callable[[proj_svc.ProjectMeta], bool],
+    reason_of: Callable[[proj_svc.ProjectMeta], str],
+    crash: dict[str, str | None],
+    initially_live: bool,
+) -> None:
+    saw = initially_live
+    while True:
+        await asyncio.sleep(2.0)
+        try:
+            cur = proj_svc.get_project(storage, owner, project_id)
+        except Exception:
+            continue
+        if cur is None:
+            continue
+        if live(cur):
+            saw = True
+        elif saw and finished(cur):
+            crash["reason"] = reason_of(cur)
+            return
+        if execution_name and (saw or live(cur)):
+            try:
+                state = job_runner.get_execution_state(execution_name)
+            except Exception:
+                state = "unknown"
+            if state in _WORKER_DONE:
+                crash["reason"] = f"execution state={state}"
+                return
+
+
+# ---------------------------------------------------------------------------
+# Analysis
+# ---------------------------------------------------------------------------
+
+
+@router.post("/pipeline/{project_id}/start", status_code=202)
+async def start(project_id: str, request: Request):
+    from backend._version import PERISCOPE_VERSION
+
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    _need_bom_netlist(meta, "Upload BOM and netlist before starting pipeline")
+    get_billing().ensure_trial_grant(storage, get_user_id(request))
+    _cas_queued(
+        storage, owner, project_id,
+        from_status=_ANALYSIS_IDLE,
+        conflict="Pipeline already running or queued",
+        cancel_requested=False,
+        execution_name=None,
+        periscope_version=PERISCOPE_VERSION,
+    )
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_pipeline(
+            project_id, owner, resume=False, free=False,
+        ),
+        fail_status={
+            "status": proj_svc.STATUS_ERROR,
+            "pipeline_state": {"error": "Failed to enqueue worker"},
+        },
+        http_detail=_ENQUEUE_FAIL,
+    )
+    _record_exec(storage, owner, project_id, name)
+    return {"status": "started", "project_id": project_id}
+
+
+@router.post("/pipeline/{project_id}/cancel")
+async def cancel(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if not _project_active(meta):
+        raise HTTPException(409, f"Pipeline is not running (status={meta.status})")
+    proj_svc.request_cancel(storage, owner, project_id)
+    return {"status": "cancel_requested", "project_id": project_id}
+
+
+@router.post("/pipeline/{project_id}/estimate")
+async def estimate(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if not meta.has_bom:
+        raise HTTPException(400, "Upload a BOM before requesting an estimate")
+    try:
+        est = estimate_pipeline_cost(storage, owner, project_id)
+    except FileNotFoundError as exc:
+        raise HTTPException(400, str(exc)) from exc
+    return est.model_dump()
+
+
+@router.post("/pipeline/{project_id}/resume", status_code=202)
+async def resume(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if meta.status != proj_svc.STATUS_PAUSED:
+        raise HTTPException(
+            400,
+            f"Project is not paused (status={meta.status}); nothing to resume.",
+        )
+    _need_bom_netlist(meta, "Project is missing BOM or netlist")
+    _cas_queued(
+        storage, owner, project_id,
+        from_status=frozenset({proj_svc.STATUS_PAUSED}),
+        conflict="Project state changed; refresh and retry",
+        cancel_requested=False,
+    )
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_pipeline(
+            project_id, owner, resume=True, free=False,
+        ),
+        fail_status={
+            "status": proj_svc.STATUS_ERROR,
+            "pipeline_state": {"error": "Failed to enqueue worker"},
+        },
+        http_detail=_ENQUEUE_FAIL,
+    )
+    _record_exec(storage, owner, project_id, name)
+    return {"status": "resumed", "project_id": project_id}
+
+
+@router.post("/pipeline/{project_id}/reprocess", status_code=202)
+async def reprocess(
+    project_id: str, request: Request, req: ReprocessRequest | None = None,
+):
+    from backend._version import PERISCOPE_VERSION
+
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    _need_bom_netlist(meta, "Upload BOM and netlist before reprocessing")
+    if _project_active(meta):
+        await _interrupt_active_pipeline(storage, owner, project_id)
+        meta = proj_svc.get_project(storage, owner, project_id) or meta
+
+    allowed = _REPROCESS_IDLE | {proj_svc.STATUS_PAUSED} if meta.status == proj_svc.STATUS_PAUSED else _REPROCESS_IDLE
+    if meta.status not in allowed and not _project_active(meta):
+        raise HTTPException(409, f"Cannot reprocess from status={meta.status}.")
+
+    body = req or ReprocessRequest()
+    retry_failed = body.mode == "failed"
+    keep_refs = (
+        proj_svc.completed_review_refs_for_retry(storage, owner, project_id)
+        if retry_failed else []
+    )
+    _cas_queued(
+        storage, owner, project_id,
+        from_status=allowed | {proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING},
+        conflict="Pipeline already running or queued",
+        cancel_requested=False,
+        execution_name=None,
+        pipeline_state=None,
+        pause_checkpoint=None,
+        pause_reason=None,
+        completed_review_refs=keep_refs,
+        periscope_version=PERISCOPE_VERSION,
+    )
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_pipeline(
+            project_id, owner, resume=retry_failed, free=False,
+        ),
+        fail_status={
+            "status": proj_svc.STATUS_ERROR,
+            "pipeline_state": {"error": "Failed to enqueue worker"},
+        },
+        http_detail=_ENQUEUE_FAIL,
+    )
+    _record_exec(storage, owner, project_id, name)
+    return {
+        "status": "reprocess_started",
+        "project_id": project_id,
+        "mode": body.mode,
+        "resume": retry_failed,
+        "kept_review_refs": keep_refs,
+    }
+
+
+@router.post("/pipeline/{project_id}/restart", status_code=202)
+async def restart(project_id: str, request: Request):
+    from backend.routers.admin import _require_admin
+
+    await _require_admin(request)
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    _need_bom_netlist(meta, "Upload BOM and netlist before starting pipeline")
+    await _stop_then_wait(storage, owner, project_id, meta)
+    proj_svc.clear_project_extractions(storage, owner, project_id)
+    _cas_queued(
+        storage, owner, project_id,
+        from_status=_ANALYSIS_IDLE | {proj_svc.STATUS_PAUSED},
+        conflict="Pipeline is busy; cancel first then retry",
+        cancel_requested=False,
+        execution_name=None,
+    )
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_pipeline(
+            project_id, owner, resume=False, free=True,
+        ),
+        fail_status={
+            "status": proj_svc.STATUS_ERROR,
+            "pipeline_state": {"error": "Failed to enqueue worker"},
+        },
+        http_detail=_ENQUEUE_FAIL,
+    )
+    _record_exec(storage, owner, project_id, name)
+    return {"status": "restarted", "project_id": project_id}
+
+
+@router.post("/pipeline/{project_id}/regen", status_code=202)
+async def regen(project_id: str, req: RegenRequest, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    _need_bom_netlist(meta, "Upload BOM and netlist before running regen")
+    unknown = set(req.stages) - _REGEN_STAGES
+    if unknown:
+        raise HTTPException(
+            400,
+            f"Invalid regen stages: {sorted(unknown)}. Valid: {sorted(_REGEN_STAGES)}",
+        )
+    if not req.stages:
+        raise HTTPException(400, "At least one stage is required")
+    await _stop_then_wait(storage, owner, project_id, meta)
+    _cas_queued(
+        storage, owner, project_id,
+        from_status=_ANALYSIS_IDLE | {proj_svc.STATUS_PAUSED},
+        conflict="Pipeline is busy; cancel first then retry",
+        cancel_requested=False,
+        execution_name=None,
+    )
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_pipeline_regen(
+            project_id, owner, stages=req.stages,
+        ),
+        fail_status={
+            "status": proj_svc.STATUS_ERROR,
+            "pipeline_state": {"error": "Failed to enqueue worker"},
+        },
+        http_detail=_ENQUEUE_FAIL,
+    )
+    _record_exec(storage, owner, project_id, name)
+    return {"status": "regen_started", "project_id": project_id, "stages": req.stages}
+
+
+@router.get("/pipeline/{project_id}/events")
+async def events(project_id: str, request: Request):
+    owner, meta = await resolve_or_404(request, project_id)
+    storage = get_storage(request)
+
+    async def gen():
+        crash: dict[str, str | None] = {"reason": None}
+        watcher = asyncio.create_task(_poll_until_crash(
+            storage, owner, project_id,
+            execution_name=meta.execution_name,
+            live=lambda m: m.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING),
+            finished=lambda m: m.status in proj_svc.TERMINAL_STATUSES,
+            reason_of=lambda m: f"project status={m.status} (terminal)",
+            crash=crash,
+            initially_live=meta.status in (
+                proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING,
+            ),
+        ))
+        try:
+            async for msg in event_bridge.tail_events(
+                storage, owner, project_id,
+                terminal_events=_ANALYSIS_SSE_DONE,
+            ):
+                if crash["reason"] is not None:
+                    break
+                ev = msg["event"]
+                if ev.startswith("placement_") or ev.startswith("pcb_"):
+                    continue
+                yield _yield_sse(ev, msg.get("data") or {})
+                if ev in _ANALYSIS_SSE_DONE:
+                    return
+            if crash["reason"] is not None:
+                cur = proj_svc.get_project(storage, owner, project_id)
+                err = (
+                    (cur.pipeline_state or {}).get("error")
+                    if cur and cur.pipeline_state
+                    else crash["reason"]
+                )
+                yield _yield_sse("pipeline_error", {
+                    "error": err or "worker terminated without writing a terminal event",
+                    "synthetic": True,
+                })
+        finally:
+            watcher.cancel()
+            try:
+                await watcher
+            except (asyncio.CancelledError, Exception):
+                pass
+
+    return _sse(gen())
+
+
+@router.get("/pipeline/{project_id}/status")
+async def status(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    healed = proj_svc.heal_if_pipeline_finished(storage, owner, project_id)
+    if healed is not None:
+        meta = healed
+    place = meta.placement_status or "draft"
+    pcb = meta.pcb_status or "draft"
+    return {
+        "status": meta.status,
+        "summary": meta.summary,
+        "pipeline_state": meta.pipeline_state,
+        "running": meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED),
+        "placement_status": meta.placement_status,
+        "placement_state": meta.placement_state,
+        "placement_running": place in ("queued", "running"),
+        "pcb_status": meta.pcb_status,
+        "pcb_state": meta.pcb_state,
+        "pcb_running": pcb in ("queued", "running"),
+        "healed": healed is not None,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Placement (topology; no credits)
+# ---------------------------------------------------------------------------
+
+
+@router.post("/pipeline/{project_id}/placement/start", status_code=202)
+async def start_placement(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    _need_bom_netlist(meta, "Upload BOM and netlist before starting placement")
+    if placement_analysis_busy(meta):
+        raise HTTPException(409, "Analysis pipeline is running; wait or cancel it first")
+    if placement_busy(meta):
+        raise HTTPException(409, "Placement pipeline already running or queued")
+    if pcb_busy(meta):
+        raise HTTPException(409, "PCB review is running; wait or cancel it first")
+    if (meta.placement_status or "draft") not in _PLACEMENT_IDLE:
+        raise HTTPException(
+            409,
+            f"Cannot start placement from placement_status={meta.placement_status}",
+        )
+    proj_svc.update_project(
+        storage, owner, project_id,
+        placement_status="queued",
+        placement_cancel_requested=False,
+        placement_state=None,
+        placement_execution_name=None,
+    )
+    _reset_event_history(storage, owner, project_id)
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_placement_pipeline(project_id, owner),
+        fail_status={
+            "placement_status": "error",
+            "placement_state": {"error": "Failed to enqueue placement worker"},
+        },
+        http_detail="Failed to enqueue placement worker; please retry",
+    )
+    proj_svc.update_project(
+        storage, owner, project_id, placement_execution_name=name,
+    )
+    return {"status": "started", "project_id": project_id}
+
+
+@router.post("/pipeline/{project_id}/placement/cancel")
+async def cancel_placement(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if not placement_busy(meta):
+        raise HTTPException(
+            409,
+            f"Placement is not running (placement_status={meta.placement_status})",
+        )
+    proj_svc.update_project(
+        storage, owner, project_id, placement_cancel_requested=True,
+    )
+    return {"status": "cancel_requested", "project_id": project_id}
+
+
+@router.get("/pipeline/{project_id}/placement/plan")
+async def get_placement_plan(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    prefix = proj_svc.project_prefix(owner, project_id)
+    for name in ("placement_plan.json", "functional_groups.json"):
+        key = f"{prefix}/{name}"
+        if storage.exists(key):
+            return storage.read_json(key)
+    raise HTTPException(404, "Placement plan not found — run placement first")
+
+
+@router.get("/pipeline/{project_id}/placement/pack")
+async def get_placement_pack(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{proj_svc.project_prefix(owner, project_id)}/placement_pack.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Placement pack not found — run placement first")
+    return storage.read_json(key)
+
+
+@router.get("/pipeline/{project_id}/placement/events")
+async def placement_events(project_id: str, request: Request):
+    owner, meta = await resolve_or_404(request, project_id)
+    storage = get_storage(request)
+
+    async def gen():
+        crash: dict[str, str | None] = {"reason": None}
+        watcher = asyncio.create_task(_poll_until_crash(
+            storage, owner, project_id,
+            execution_name=meta.placement_execution_name,
+            live=lambda m: (m.placement_status or "draft") in ("queued", "running"),
+            finished=lambda m: (m.placement_status or "draft") in (
+                "complete", "error", "cancelled",
+            ),
+            reason_of=lambda m: f"placement_status={m.placement_status or 'draft'} (terminal)",
+            crash=crash,
+            initially_live=(meta.placement_status or "draft") in ("queued", "running"),
+        ))
+        try:
+            async for msg in event_bridge.tail_events(
+                storage, owner, project_id,
+                terminal_events=_PLACEMENT_SSE_DONE,
+            ):
+                if crash["reason"] is not None:
+                    break
+                ev = msg["event"]
+                keep = ev.startswith("placement_") or ev == "heartbeat"
+                if not keep or ev.startswith("pcb_"):
+                    continue
+                yield _yield_sse(ev, msg.get("data") or {})
+                if ev in _PLACEMENT_SSE_DONE:
+                    return
+            if crash["reason"] is not None:
+                cur = proj_svc.get_project(storage, owner, project_id)
+                err = cur.placement_state.get("error") if cur and cur.placement_state else None
+                yield _yield_sse("placement_error", {
+                    "error": err or crash["reason"]
+                    or "placement worker terminated without a terminal event",
+                    "synthetic": True,
+                })
+        finally:
+            watcher.cancel()
+            try:
+                await watcher
+            except (asyncio.CancelledError, Exception):
+                pass
+
+    return _sse(gen())
+
+
+# ---------------------------------------------------------------------------
+# PCB review
+# ---------------------------------------------------------------------------
+
+
+@router.post("/pipeline/{project_id}/pcb/start", status_code=202)
+async def start_pcb(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if not meta.has_pcb:
+        raise HTTPException(400, "Upload a .kicad_pcb before starting PCB review")
+    _need_bom_netlist(meta, "Upload BOM and netlist before starting PCB review")
+    if pcb_analysis_busy(meta):
+        raise HTTPException(409, "Analysis pipeline is running; wait or cancel it first")
+    if pcb_placement_busy(meta):
+        raise HTTPException(409, "Placement pipeline is running; wait or cancel it first")
+    if pcb_busy(meta):
+        raise HTTPException(409, "PCB review already running or queued")
+    if (meta.pcb_status or "draft") not in _PCB_IDLE:
+        raise HTTPException(
+            409,
+            f"Cannot start PCB review from pcb_status={meta.pcb_status}",
+        )
+    proj_svc.update_project(
+        storage, owner, project_id,
+        pcb_status="queued",
+        pcb_cancel_requested=False,
+        pcb_state=None,
+        pcb_execution_name=None,
+    )
+    _reset_event_history(storage, owner, project_id)
+    name = _enqueue_or_503(
+        storage, owner, project_id,
+        enqueue=lambda: job_runner.enqueue_pcb_pipeline(project_id, owner),
+        fail_status={
+            "pcb_status": "error",
+            "pcb_state": {"error": "Failed to enqueue PCB worker"},
+        },
+        http_detail="Failed to enqueue PCB worker; please retry",
+    )
+    proj_svc.update_project(
+        storage, owner, project_id, pcb_execution_name=name,
+    )
+    return {"status": "started", "project_id": project_id}
+
+
+@router.post("/pipeline/{project_id}/pcb/cancel")
+async def cancel_pcb(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, meta = await resolve_or_404(request, project_id)
+    if not pcb_busy(meta):
+        raise HTTPException(
+            409,
+            f"PCB review is not running (pcb_status={meta.pcb_status})",
+        )
+    proj_svc.update_project(
+        storage, owner, project_id, pcb_cancel_requested=True,
+    )
+    return {"status": "cancel_requested", "project_id": project_id}
+
+
+@router.get("/pipeline/{project_id}/pcb/inventory")
+async def get_pcb_inventory(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{proj_svc.project_prefix(owner, project_id)}/pcb_inventory.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "PCB inventory not found — run PCB review first")
+    return storage.read_json(key)
+
+
+@router.get("/pipeline/{project_id}/pcb/events")
+async def pcb_events(project_id: str, request: Request):
+    owner, meta = await resolve_or_404(request, project_id)
+    storage = get_storage(request)
+
+    async def gen():
+        crash: dict[str, str | None] = {"reason": None}
+        watcher = asyncio.create_task(_poll_until_crash(
+            storage, owner, project_id,
+            execution_name=meta.pcb_execution_name,
+            live=lambda m: (m.pcb_status or "draft") in ("queued", "running"),
+            finished=lambda m: (m.pcb_status or "draft") in (
+                "complete", "error", "cancelled",
+            ),
+            reason_of=lambda m: f"pcb_status={m.pcb_status or 'draft'} (terminal)",
+            crash=crash,
+            initially_live=(meta.pcb_status or "draft") in ("queued", "running"),
+        ))
+        try:
+            async for msg in event_bridge.tail_events(
+                storage, owner, project_id,
+                terminal_events=_PCB_SSE_DONE,
+            ):
+                if crash["reason"] is not None:
+                    break
+                ev = msg["event"]
+                if not (ev.startswith("pcb_") or ev == "heartbeat"):
+                    continue
+                yield _yield_sse(ev, msg.get("data") or {})
+                if ev in _PCB_SSE_DONE:
+                    return
+            if crash["reason"] is not None:
+                cur = proj_svc.get_project(storage, owner, project_id)
+                ev, payload = pcb_sse_terminal_from_status(
+                    cur.pcb_status if cur else None,
+                    cur.pcb_state if cur else None,
+                    crash["reason"],
+                )
+                yield _yield_sse(ev, payload)
+        finally:
+            watcher.cancel()
+            try:
+                await watcher
+            except (asyncio.CancelledError, Exception):
+                pass
+
+    return _sse(gen())

periscope/src/backend/routers/projects.py

@@ -0,0 +1,990 @@
+"""Project CRUD, library lookup, and file ingestion."""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+import logging
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
+from fastapi.responses import JSONResponse, Response
+from pydantic import BaseModel
+
+from backend.config import settings
+from backend.periscopex.utils import natural_sort_key, safe_mpn
+from backend.routers.deps import get_storage, get_user_id, resolve_or_404
+from backend.services import projects as proj_svc
+
+log = logging.getLogger(__name__)
+
+MAX_UPLOAD_BYTES = 30 * 1024 * 1024
+_MAX_MB = MAX_UPLOAD_BYTES // 1024 // 1024
+
+router = APIRouter(tags=["projects"])
+
+
+def _xlsx_to_csv(raw: bytes) -> bytes:
+    import openpyxl
+
+    wb = openpyxl.load_workbook(io.BytesIO(raw), read_only=True, data_only=True)
+    try:
+        sheet = wb.active
+        buf = io.StringIO()
+        writer = csv.writer(buf)
+        for row in sheet.iter_rows(values_only=True):
+            writer.writerow(["" if cell is None else str(cell) for cell in row])
+        return buf.getvalue().encode("utf-8")
+    finally:
+        wb.close()
+
+
+def _bom_file_to_csv_bytes(path: Path) -> bytes | None:
+    """CSV as-is; xlsx flattened to CSV; anything else ignored (KiCad zip)."""
+    suffix = path.suffix.lower()
+    payload = path.read_bytes()
+    if suffix == ".csv":
+        return payload
+    if suffix == ".xlsx":
+        try:
+            return _xlsx_to_csv(payload)
+        except Exception:
+            return None
+    return None
+
+
+def _reject_oversize(nbytes: int, *, label: str | None = None) -> None:
+    if nbytes <= MAX_UPLOAD_BYTES:
+        return
+    if label:
+        mb = nbytes / 1024 / 1024
+        raise HTTPException(413, f"{label} is {mb:.1f} MB — exceeds {_MAX_MB} MB limit")
+    raise HTTPException(413, f"File too large (max {_MAX_MB} MB)")
+
+
+def _access_or_404(request: Request, project_id: str):
+    hit = proj_svc.resolve_project_access(
+        get_storage(request), get_user_id(request), project_id,
+    )
+    if not hit:
+        raise HTTPException(404, "Project not found")
+    return hit
+
+
+def _parse_bom_csv(
+    data: bytes, *, reference_column: str, mpn_column: str,
+) -> dict:
+    from backend.periscopex.parsers import parse_bom
+
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp:
+        tmp.write(data)
+        tmp_path = tmp.name
+    try:
+        return parse_bom(tmp_path, reference_col=reference_column, mpn_col=mpn_column)
+    finally:
+        os.unlink(tmp_path)
+
+
+def _bucket_mpns(bom: dict) -> tuple[list[str], list[str], list[str]]:
+    from backend.periscopex.taxonomy import SIMPLE_TYPES, type_for_ref
+
+    ic: list[str] = []
+    passive: list[str] = []
+    simple: list[str] = []
+    seen_ic: set[str] = set()
+    seen_passive: set[str] = set()
+    seen_simple: set[str] = set()
+    for ref, info in sorted(bom.items()):
+        mpn = info.get("mpn")
+        if not mpn:
+            continue
+        kind = type_for_ref(ref)
+        if kind == "ic" and mpn not in seen_ic:
+            seen_ic.add(mpn)
+            ic.append(mpn)
+        elif kind == "passive" and mpn not in seen_passive:
+            seen_passive.add(mpn)
+            passive.append(mpn)
+        elif kind and kind in SIMPLE_TYPES and mpn not in seen_simple:
+            seen_simple.add(mpn)
+            simple.append(mpn)
+    return ic, passive, simple
+
+
+def _pins_by_ref(
+    parts: dict[str, str],
+    nets: dict[str, list[tuple[str, str]]],
+) -> list[dict]:
+    by_ref: dict[str, dict[str, str]] = {ref: {} for ref in parts}
+    for net_name, pins in nets.items():
+        for ref, pin in pins:
+            by_ref.setdefault(ref, {}).setdefault(pin, net_name)
+    rows: list[dict] = []
+    for ref in sorted(by_ref, key=natural_sort_key):
+        pin_map = by_ref[ref]
+        rows.append({
+            "ref": ref,
+            "pins": [
+                {"number": num, "net_name": pin_map[num]}
+                for num in sorted(pin_map, key=natural_sort_key)
+            ],
+        })
+    return rows
+
+
+def _netlist_ext(key: str) -> str:
+    leaf = key.rsplit("/", 1)[-1]
+    return leaf.rsplit(".", 1)[-1] if "." in leaf else "asc"
+
+
+async def _gather_netlist_blobs(
+    file: UploadFile | None,
+    files: list[UploadFile] | None,
+    paths: str | None,
+) -> list[tuple[str, bytes]]:
+    uploads = list(files or []) if files else ([file] if file is not None else [])
+    rels: list[str] | None = None
+    if paths:
+        try:
+            parsed = json.loads(paths)
+        except json.JSONDecodeError:
+            parsed = None
+        if isinstance(parsed, list) and all(isinstance(x, str) for x in parsed):
+            rels = parsed
+    blobs: list[tuple[str, bytes]] = []
+    seen: set[tuple[str, int]] = set()
+    for i, uf in enumerate(uploads):
+        data = await uf.read()
+        _reject_oversize(len(data))
+        name = rels[i] if rels is not None and i < len(rels) else (uf.filename or "netlist")
+        mark = (name, len(data))
+        if mark in seen:
+            continue
+        seen.add(mark)
+        blobs.append((name, data))
+    if not blobs:
+        raise HTTPException(400, "No netlist file uploaded")
+    return blobs
+
+
+def _passive_hit_json(mpn: str, safe: str, model_data: dict, *, cached: bool, lcsc_id: str) -> dict:
+    return {
+        "mpn": mpn,
+        "safe_mpn": safe,
+        "model": model_data,
+        "cached": cached,
+        "lcsc_id": lcsc_id,
+    }
+
+
+# --- Library ---
+
+
+class LibraryCheckRequest(BaseModel):
+    ic_mpns: list[str] = []
+    passive_mpns: list[str] = []
+    simple_mpns: list[str] = []
+
+
+@router.post("/library/check")
+async def check_library(body: LibraryCheckRequest, request: Request):
+    storage = get_storage(request)
+    ics = [mpn for mpn in body.ic_mpns if proj_svc.library_has_extraction(storage, mpn)]
+    patterns = proj_svc.load_library_patterns(storage) if body.passive_mpns else []
+    passives: list[str] = []
+    if body.passive_mpns:
+        from backend.periscopex.resolve_passives import resolve_mpn
+
+        passives = [
+            mpn for mpn in body.passive_mpns
+            if resolve_mpn(mpn, patterns) is not None
+            or proj_svc.library_has_passive_model(storage, mpn) is not None
+        ]
+    simples = [mpn for mpn in body.simple_mpns if proj_svc.library_has_model(storage, mpn)]
+    wanted = set(body.ic_mpns + body.passive_mpns + body.simple_mpns)
+    sheets = [
+        mpn for mpn in wanted
+        if proj_svc.library_has_datasheet(storage, mpn, patterns=patterns)
+    ]
+    return {
+        "ic_resolved": ics,
+        "passive_resolved": passives,
+        "simple_resolved": simples,
+        "datasheets_available": sheets,
+    }
+
+
+@router.get("/library")
+async def get_library(request: Request):
+    storage = get_storage(request)
+    return JSONResponse(
+        content=proj_svc.list_library_catalog(storage),
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+@router.get("/library/datasheet/{mpn:path}")
+async def get_library_datasheet(mpn: str, request: Request):
+    storage = get_storage(request)
+    key = proj_svc.library_has_datasheet(storage, mpn)
+    if not key:
+        raise HTTPException(404, f"Datasheet not in library: {mpn}")
+    return Response(
+        content=storage.read_bytes(key),
+        media_type="application/pdf",
+        headers={"Content-Disposition": f'inline; filename="{mpn}.pdf"'},
+    )
+
+
+# --- CRUD ---
+
+
+class CreateProjectRequest(BaseModel):
+    name: str
+
+
+class RenameRequest(BaseModel):
+    name: str
+
+
+@router.post("/projects")
+async def create_project(body: CreateProjectRequest, request: Request):
+    meta = proj_svc.create_project(get_storage(request), get_user_id(request), body.name)
+    return meta.model_dump()
+
+
+@router.get("/projects")
+async def list_projects(request: Request):
+    storage = get_storage(request)
+    uid = get_user_id(request)
+    owned = proj_svc.list_projects(storage, uid)
+    shared = proj_svc.list_shared_projects(storage, uid)
+    return [m.model_dump() for m in owned + shared]
+
+
+@router.get("/projects/{project_id}")
+async def get_project(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner_id, meta = await resolve_or_404(request, project_id)
+    for healer in (
+        proj_svc.heal_if_pipeline_finished,
+        proj_svc.heal_if_placement_stuck,
+        proj_svc.heal_if_pcb_stuck,
+    ):
+        healed = healer(storage, owner_id, project_id)
+        if healed is not None:
+            meta = healed
+    return meta.model_dump()
+
+
+@router.delete("/projects/{project_id}")
+async def delete_project(project_id: str, request: Request):
+    storage = get_storage(request)
+    uid = get_user_id(request)
+    if proj_svc.delete_project(storage, uid, project_id):
+        return {"ok": True}
+    hit = proj_svc.resolve_project_access(storage, uid, project_id)
+    if hit:
+        owner_id, _ = hit
+        proj_svc.remove_collaborator(storage, owner_id, project_id, uid)
+        return {"ok": True, "removed_self": True}
+    raise HTTPException(404, "Project not found")
+
+
+@router.patch("/projects/{project_id}")
+async def rename_project(project_id: str, body: RenameRequest, request: Request):
+    storage = get_storage(request)
+    owner_id, _ = _access_or_404(request, project_id)
+    name = body.name.strip()
+    if not name:
+        raise HTTPException(400, "Name must be non-empty")
+    meta = proj_svc.update_project(storage, owner_id, project_id, name=name)
+    return meta.model_dump()
+
+
+@router.post("/projects/{project_id}/reopen")
+async def reopen_project(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner_id, meta = _access_or_404(request, project_id)
+    if meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED):
+        raise HTTPException(409, "Pipeline is running; cancel it before reopening")
+    meta = proj_svc.reopen_project(storage, owner_id, project_id)
+    return meta.model_dump()
+
+
+# --- Downloads ---
+
+
+@router.get("/projects/{project_id}/files/bom")
+async def download_bom(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner_id, _ = await resolve_or_404(request, project_id)
+    key = proj_svc.get_bom_key(storage, owner_id, project_id)
+    if not key:
+        raise HTTPException(404, "BOM not uploaded")
+    return Response(
+        content=storage.read_bytes(key),
+        media_type="text/csv",
+        headers={"Content-Disposition": 'attachment; filename="bom.csv"'},
+    )
+
+
+@router.get("/projects/{project_id}/files/netlist")
+async def download_netlist(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner_id, _ = await resolve_or_404(request, project_id)
+    key = proj_svc.get_netlist_key(storage, owner_id, project_id)
+    if not key:
+        raise HTTPException(404, "Netlist not uploaded")
+    ext = _netlist_ext(key)
+    return Response(
+        content=storage.read_bytes(key),
+        media_type="text/plain",
+        headers={"Content-Disposition": f'attachment; filename="netlist.{ext}"'},
+    )
+
+
+@router.get("/projects/{project_id}/netlist/subdesigns")
+async def get_netlist_subdesigns(project_id: str, request: Request):
+    from backend.periscopex.parsers_edif import list_edif_subdesigns
+
+    storage = get_storage(request)
+    owner_id, meta = await resolve_or_404(request, project_id)
+    empty = {"sub_designs": [], "selected": None}
+    if meta.netlist_format != "edif":
+        return empty
+    key = proj_svc.get_netlist_key(storage, owner_id, project_id)
+    if not key:
+        return empty
+    data = storage.read_bytes(key)
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".edn") as tmp:
+        tmp.write(data)
+        tmp_path = tmp.name
+    try:
+        subs = list_edif_subdesigns(tmp_path)
+    finally:
+        os.unlink(tmp_path)
+    return {"sub_designs": subs, "selected": meta.netlist_subdesigns}
+
+
+class NetlistSubdesignsUpdate(BaseModel):
+    selected: list[str] | None
+
+
+@router.put("/projects/{project_id}/netlist/subdesigns")
+async def set_netlist_subdesigns(
+    project_id: str, payload: NetlistSubdesignsUpdate, request: Request,
+):
+    storage = get_storage(request)
+    owner_id, _ = _access_or_404(request, project_id)
+    cleaned = [s.strip() for s in (payload.selected or []) if s and s.strip()]
+    meta = proj_svc.update_project(
+        storage, owner_id, project_id,
+        netlist_subdesigns=cleaned if payload.selected is not None else None,
+    )
+    return meta.model_dump()
+
+
+@router.get("/projects/{project_id}/files/datasheets")
+async def list_datasheets(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner_id, _ = await resolve_or_404(request, project_id)
+    stems = proj_svc.list_project_datasheets(storage, owner_id, project_id)
+    return {"stems": sorted(stems)}
+
+
+# --- Uploads ---
+
+
+@router.post("/projects/{project_id}/upload/bom")
+async def upload_bom(
+    project_id: str,
+    file: UploadFile,
+    request: Request,
+    reference_column: str = "Reference",
+    mpn_column: str = "Manufacturer Part Number",
+    column_is_lcsc: bool | None = None,
+):
+    storage = get_storage(request)
+    owner_id, _ = _access_or_404(request, project_id)
+    data = await file.read()
+    _reject_oversize(len(data))
+    filename = file.filename or ""
+    if filename.lower().endswith(".xlsx"):
+        try:
+            data = _xlsx_to_csv(data)
+        except Exception as exc:
+            raise HTTPException(400, f"Invalid Excel file: {exc}") from exc
+
+    try:
+        bom = _parse_bom_csv(
+            data, reference_column=reference_column, mpn_column=mpn_column,
+        )
+    except Exception as exc:
+        raise HTTPException(400, f"Invalid BOM file: {exc}") from exc
+
+    lcsc_resolved = 0
+    lcsc_detected = False
+    lcsc_map: dict[str, str] = {}
+    lcsc_payloads: dict[str, dict] = {}
+    try:
+        from backend.services.purple_parts import (
+            detect_lcsc_column,
+            resolve_lcsc_column_bytes,
+        )
+
+        lcsc_detected = detect_lcsc_column(data, mpn_column)
+        if column_is_lcsc or lcsc_detected:
+            data, lcsc_resolved, lcsc_map, lcsc_payloads = await resolve_lcsc_column_bytes(
+                data, mpn_col=mpn_column,
+            )
+    except Exception:
+        log.warning("purple-parts BOM resolve failed", exc_info=True)
+
+    if lcsc_resolved:
+        try:
+            bom = _parse_bom_csv(
+                data, reference_column=reference_column, mpn_column=mpn_column,
+            )
+        except Exception:
+            log.warning(
+                "Re-parse after LCSC rewrite failed; using pre-rewrite BOM for classification",
+                exc_info=True,
+            )
+
+    ic_mpns, passive_mpns, simple_mpns = _bucket_mpns(bom)
+
+    if (
+        settings.use_purple_parts
+        and not lcsc_detected
+        and not column_is_lcsc
+        and passive_mpns
+    ):
+        try:
+            from backend.services.purple_parts import lookup_mpn_batch
+
+            parts = await lookup_mpn_batch(passive_mpns)
+            for pmpn, part in parts.items():
+                if part and part.get("lcsc") and part.get("description"):
+                    lcsc_map[part["lcsc"]] = pmpn
+                    lcsc_payloads[part["lcsc"]] = dict(part)
+        except Exception:
+            log.warning("purple-parts by-mpn passive enrich failed", exc_info=True)
+
+    key = proj_svc.save_bom(storage, owner_id, project_id, data)
+    fields: dict = {
+        "bom_columns": {"reference": reference_column, "mpn": mpn_column},
+        "component_mpns": {
+            "ic": ic_mpns,
+            "passive": passive_mpns,
+            "simple": simple_mpns,
+        },
+    }
+    if lcsc_map:
+        fields["lcsc_to_mpn"] = lcsc_map
+    if lcsc_payloads:
+        fields["lcsc_payloads"] = lcsc_payloads
+    proj_svc.update_project(storage, owner_id, project_id, **fields)
+    return {
+        "path": key,
+        "components": len(bom),
+        "lcsc_resolved": lcsc_resolved,
+        "lcsc_detected": lcsc_detected,
+        "lcsc_to_mpn": lcsc_map,
+    }
+
+
+@router.post("/projects/{project_id}/upload/netlist")
+async def upload_netlist(
+    project_id: str,
+    request: Request,
+    file: UploadFile | None = File(default=None),
+    files: list[UploadFile] | None = File(default=None),
+    paths: str | None = Form(default=None),
+):
+    from backend.periscopex.netlist_bundle import materialize_netlist_upload
+    from backend.periscopex.parsers import parse_netlist_any, validate_netlist
+    from backend.periscopex.parsers_edif import list_edif_subdesigns
+
+    storage = get_storage(request)
+    owner_id, _ = _access_or_404(request, project_id)
+    blobs = await _gather_netlist_blobs(file, files, paths)
+
+    sub_designs: list[dict] = []
+    bom_saved = False
+    pcb_saved = False
+    sheets = 1
+    try:
+        with tempfile.TemporaryDirectory() as tmp:
+            parsed = materialize_netlist_upload(blobs, Path(tmp) / "work")
+            parts, nets, fmt = parse_netlist_any(parsed.root)
+            if fmt == "edif":
+                sub_designs = list_edif_subdesigns(parsed.root)
+            issues = validate_netlist(parts, nets)
+            if issues:
+                raise ValueError("; ".join(issues))
+            root_bytes = parsed.root.read_bytes()
+            key = proj_svc.save_netlist(storage, owner_id, project_id, root_bytes, fmt=fmt)
+            if fmt == "kicad_sch":
+                proj_svc.save_companion_sheets(
+                    storage, owner_id, project_id, parsed.root, parsed.extra_sch,
+                )
+                sheets = 1 + len(parsed.extra_sch)
+            else:
+                proj_svc.clear_companion_sheets(storage, owner_id, project_id)
+            if parsed.pcb is not None:
+                proj_svc.save_pcb(storage, owner_id, project_id, parsed.pcb.read_bytes())
+                pcb_saved = True
+            if parsed.bom is not None:
+                bom_bytes = _bom_file_to_csv_bytes(parsed.bom)
+                if bom_bytes:
+                    proj_svc.save_bom(storage, owner_id, project_id, bom_bytes)
+                    proj_svc.update_project(
+                        storage, owner_id, project_id,
+                        bom_columns={
+                            "reference": "Reference",
+                            "mpn": "Manufacturer Part Number",
+                        },
+                    )
+                    bom_saved = True
+    except HTTPException:
+        raise
+    except Exception as exc:
+        raise HTTPException(400, f"Netlist failed sanity check: {exc}") from exc
+
+    designator_pins: list[dict] = []
+    if fmt != "pads":
+        designator_pins = _pins_by_ref(parts, nets)
+    return {
+        "path": key,
+        "parts": len(parts),
+        "nets": len(nets),
+        "format": fmt,
+        "sub_designs": sub_designs,
+        "designator_pins": designator_pins,
+        "pcb_saved": pcb_saved,
+        "bom_saved": bom_saved,
+        "sheets": sheets,
+    }
+
+
+@router.post("/projects/{project_id}/upload/pcb")
+async def upload_pcb(project_id: str, file: UploadFile, request: Request):
+    from backend.periscopex.parsers_kicad_pcb import parse_kicad_pcb
+
+    storage = get_storage(request)
+    owner_id, _ = _access_or_404(request, project_id)
+    data = await file.read()
+    _reject_oversize(len(data))
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".kicad_pcb") as tmp:
+        tmp.write(data)
+        tmp_path = tmp.name
+    try:
+        layout = parse_kicad_pcb(tmp_path)
+    except Exception as exc:
+        raise HTTPException(400, f"Invalid KiCad PCB: {exc}") from exc
+    finally:
+        os.unlink(tmp_path)
+    key = proj_svc.save_pcb(storage, owner_id, project_id, data)
+    return {
+        "path": key,
+        "footprints": len(layout.footprints),
+        "nets": len(layout.nets),
+        "segments": len(layout.segments),
+    }
+
+
+@router.post("/projects/{project_id}/upload/datasheets")
+async def upload_datasheets(
+    project_id: str, file: UploadFile, mpn: str,
+    request: Request, also_for: str | None = None,
+):
+    storage = get_storage(request)
+    owner_id, _ = _access_or_404(request, project_id)
+    if not file.filename or not file.filename.lower().endswith(".pdf"):
+        raise HTTPException(400, "File must be a PDF")
+    data = await file.read()
+    _reject_oversize(len(data), label=file.filename or mpn)
+    key = proj_svc.save_datasheet(storage, owner_id, project_id, mpn, data)
+    extra_mpns: list[str] = []
+    if also_for:
+        for extra in also_for.split(","):
+            extra = extra.strip()
+            if extra:
+                proj_svc.save_datasheet(storage, owner_id, project_id, extra, data)
+                extra_mpns.append(extra)
+    return {"path": key, "mpn": mpn, "also_for": extra_mpns}
+
+
+# --- Collaborators ---
+
+
+class AddCollaboratorRequest(BaseModel):
+    email: str
+
+
+def _local_member(uid: str, owner_id: str) -> dict:
+    return {
+        "user_id": uid,
+        "name": None,
+        "email": None,
+        "image_url": None,
+        "role": "owner" if uid == owner_id else "collaborator",
+    }
+
+
+@router.get("/projects/{project_id}/collaborators")
+async def list_collaborators(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner_id, meta = _access_or_404(request, project_id)
+    ids = [owner_id] + [c for c in meta.collaborators if c != owner_id]
+    if settings.use_auth:
+        from backend.services.user_directory import get_user_profile
+
+        members = []
+        for uid in ids:
+            profile = await get_user_profile(uid)
+            members.append({
+                "user_id": uid,
+                "name": profile.get("name"),
+                "email": profile.get("email"),
+                "image_url": profile.get("image_url"),
+                "role": "owner" if uid == owner_id else "collaborator",
+            })
+    else:
+        members = [_local_member(uid, owner_id) for uid in ids]
+    return {"owner_user_id": owner_id, "collaborators": members}
+
+
+@router.post("/projects/{project_id}/collaborators")
+async def add_collaborator(project_id: str, body: AddCollaboratorRequest, request: Request):
+    storage = get_storage(request)
+    uid = get_user_id(request)
+    meta = proj_svc.get_project(storage, uid, project_id)
+    if not meta:
+        raise HTTPException(404, "Project not found")
+    if not settings.use_auth:
+        raise HTTPException(400, "Collaboration requires authentication to be enabled")
+
+    from backend.services.user_directory import find_user_id_by_email, get_user_profile
+
+    collab_id = await find_user_id_by_email(body.email)
+    if not collab_id:
+        raise HTTPException(404, "No user found with that email")
+    if collab_id == uid:
+        raise HTTPException(400, "Cannot add yourself as a collaborator")
+    if collab_id in meta.collaborators:
+        raise HTTPException(409, "User is already a collaborator")
+    proj_svc.add_collaborator(storage, uid, project_id, collab_id)
+    profile = await get_user_profile(collab_id)
+    return {
+        "user_id": collab_id,
+        "name": profile.get("name"),
+        "email": profile.get("email"),
+        "image_url": profile.get("image_url"),
+    }
+
+
+@router.delete("/projects/{project_id}/collaborators/{collaborator_user_id}")
+async def remove_collaborator(project_id: str, collaborator_user_id: str, request: Request):
+    from backend.routers.admin import is_admin
+
+    storage = get_storage(request)
+    uid = get_user_id(request)
+    meta = proj_svc.get_project(storage, uid, project_id)
+    owner_id = uid
+    if not meta:
+        if not await is_admin(request):
+            raise HTTPException(404, "Project not found")
+        found = proj_svc.find_project_any_user(storage, project_id)
+        if not found:
+            raise HTTPException(404, "Project not found")
+        owner_id, meta = found
+    if collaborator_user_id not in meta.collaborators:
+        raise HTTPException(404, "User is not a collaborator")
+    proj_svc.remove_collaborator(storage, owner_id, project_id, collaborator_user_id)
+    return {"ok": True}
+
+
+@router.post("/projects/{project_id}/collaborators/{collaborator_user_id}/make-owner")
+async def make_collaborator_owner(
+    project_id: str, collaborator_user_id: str, request: Request,
+):
+    from backend.routers.admin import is_admin
+
+    if not await is_admin(request):
+        raise HTTPException(403, "Admin access required")
+    storage = get_storage(request)
+    found = proj_svc.find_project_any_user(storage, project_id)
+    if not found:
+        raise HTTPException(404, "Project not found")
+    current_owner, _ = found
+    try:
+        proj_svc.transfer_ownership(
+            storage, current_owner, project_id, collaborator_user_id,
+        )
+    except ValueError as exc:
+        raise HTTPException(400, str(exc)) from exc
+    return {"ok": True, "owner_user_id": collaborator_user_id}
+
+
+# --- Datasheet fetch + DigiKey ---
+
+
+@router.get("/digikey/datasheet")
+@router.get("/datasheets/fetch")
+async def fetch_auto_datasheet(mpn: str, request: Request, lcsc: str | None = None):
+    from backend.services.datasheet_finder import find_datasheet
+
+    result = await find_datasheet(mpn, lcsc_id=lcsc)
+    if not result.ok:
+        return JSONResponse(
+            status_code=404,
+            content={
+                "detail": result.error or "Failed to fetch datasheet",
+                "url": result.url,
+                "urls": result.suggested_urls or ([result.url] if result.url else []),
+                "source": result.source,
+            },
+        )
+    headers = {"Content-Disposition": f'attachment; filename="{mpn}.pdf"'}
+    if result.url:
+        headers["X-Datasheet-Url"] = result.url
+    if result.source:
+        headers["X-Datasheet-Source"] = result.source
+    try:
+        proj_svc.remember_datasheet(
+            get_storage(request), mpn, result.pdf_bytes, extra_mpns=result.alias_mpns,
+        )
+    except Exception:
+        pass
+    return Response(content=result.pdf_bytes, media_type="application/pdf", headers=headers)
+
+
+class AutoResolveItem(BaseModel):
+    mpn: str
+    component_type: str
+
+
+class AutoResolveRequest(BaseModel):
+    items: list[AutoResolveItem]
+
+
+@router.post("/digikey/auto-resolve")
+async def auto_resolve(body: AutoResolveRequest, request: Request):
+    import asyncio
+
+    from backend.services.datasheet_extract import CatalogResolveMiss, auto_resolve_specs
+    from backend.services.digikey import fetch_params
+
+    if not settings.use_digikey:
+        raise HTTPException(400, "DigiKey API not configured")
+
+    storage = get_storage(request)
+    sem = asyncio.Semaphore(10)
+
+    async def resolve_one(item: AutoResolveItem) -> dict:
+        async with sem:
+            try:
+                safe = safe_mpn(item.mpn)
+                if item.component_type == "passive":
+                    lib_key = f"library/passives/{safe}.json"
+                    if not storage.exists(lib_key):
+                        legacy = f"library/models/{safe}.json"
+                        if storage.exists(legacy):
+                            return {"mpn": item.mpn, "status": "resolved"}
+                else:
+                    lib_key = f"library/models/{safe}.json"
+                if storage.exists(lib_key):
+                    return {"mpn": item.mpn, "status": "resolved"}
+
+                result = await fetch_params(item.mpn)
+                if not result.ok or not result.params:
+                    return {
+                        "mpn": item.mpn,
+                        "status": "failed",
+                        "error": result.error or "No parameters",
+                    }
+                try:
+                    model = await auto_resolve_specs(
+                        mpn=item.mpn,
+                        digikey_params=result.params.parameters,
+                        digikey_category=result.params.category,
+                        digikey_description=result.params.description,
+                        component_type=item.component_type,
+                        use_llm=settings.has_llm_credentials(),
+                    )
+                except CatalogResolveMiss as exc:
+                    return {"mpn": item.mpn, "status": "failed", "error": str(exc)}
+                storage.write_json(lib_key, model.model_dump())
+                return {"mpn": item.mpn, "status": "resolved"}
+            except Exception as exc:
+                log.warning("Auto-resolve failed for %s: %s", item.mpn, exc, exc_info=True)
+                return {
+                    "mpn": item.mpn,
+                    "status": "failed",
+                    "error": str(exc) or type(exc).__name__,
+                }
+
+    results = await asyncio.gather(*(resolve_one(item) for item in body.items))
+    return {"results": results}
+
+
+# --- LCSC wizard resolve ---
+
+
+class LcscResolvePassiveRequest(BaseModel):
+    lcsc_id: str
+
+
+def _seed_taxonomy_dir(storage, tax_dir: Path) -> None:
+    for key in storage.list_prefix("taxonomy/"):
+        if key.endswith(".json"):
+            storage.download_to_local(key, tax_dir / key.rsplit("/", 1)[-1])
+    if any(tax_dir.glob("*.json")):
+        return
+    repo_tax = settings.taxonomy_dir
+    if repo_tax.is_dir():
+        for path in repo_tax.glob("*.json"):
+            shutil.copy2(path, tax_dir / path.name)
+
+
+@router.post("/projects/{project_id}/lcsc/resolve-passive")
+async def lcsc_resolve_passive(
+    project_id: str, body: LcscResolvePassiveRequest, request: Request,
+):
+    from backend.services.api_logs import ApiLogger
+    from backend.services.billing_hook import InsufficientCredits, get_billing
+    from backend.services.datasheet_extract import auto_resolve_specs
+    from backend.services.passive_from_distributor import specs_from_lcsc_payload
+
+    storage = get_storage(request)
+    owner_id, meta = _access_or_404(request, project_id)
+    payload = (meta.lcsc_payloads or {}).get(body.lcsc_id)
+    if not payload:
+        raise HTTPException(404, f"No cached payload for LCSC id {body.lcsc_id!r}")
+    mpn = (payload.get("mpn") or "").strip()
+    if not mpn:
+        raise HTTPException(404, f"Cached payload for {body.lcsc_id!r} has no MPN")
+
+    safe = safe_mpn(mpn)
+    project_model_key = (
+        f"{proj_svc.project_prefix(owner_id, project_id)}/models/{safe}.json"
+    )
+    if storage.exists(project_model_key):
+        return _passive_hit_json(
+            mpn, safe, storage.read_json(project_model_key),
+            cached=True, lcsc_id=body.lcsc_id,
+        )
+    lib_key = proj_svc.library_has_passive_model(storage, mpn)
+    if lib_key:
+        storage.copy_object(lib_key, project_model_key)
+        return _passive_hit_json(
+            mpn, safe, storage.read_json(project_model_key),
+            cached=True, lcsc_id=body.lcsc_id,
+        )
+
+    synth_category = " / ".join(
+        p for p in (payload.get("category"), payload.get("subcategory")) if p
+    ) or None
+    synth_params: list[dict[str, str]] = []
+    if payload.get("package"):
+        synth_params.append({"name": "Package / Case", "value": payload["package"]})
+    if payload.get("manufacturer"):
+        synth_params.append({"name": "Manufacturer", "value": payload["manufacturer"]})
+    description = payload.get("description") or ""
+    if not description:
+        raise HTTPException(
+            502,
+            f"Cached LCSC payload for {body.lcsc_id!r} has no description — "
+            "cannot auto-resolve",
+        )
+
+    catalog_model = specs_from_lcsc_payload(mpn, payload)
+    if catalog_model is not None:
+        storage.write_json(project_model_key, catalog_model.model_dump())
+        proj_svc.save_to_library(
+            storage, project_model_key, "passives", f"{safe}.json",
+        )
+        return _passive_hit_json(
+            mpn, safe, catalog_model.model_dump(),
+            cached=False, lcsc_id=body.lcsc_id,
+        )
+
+    api_logger = ApiLogger()
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tax_dir = Path(tmpdir) / "taxonomy"
+        tax_dir.mkdir()
+        _seed_taxonomy_dir(storage, tax_dir)
+        try:
+            model = await auto_resolve_specs(
+                mpn=mpn,
+                digikey_params=synth_params,
+                digikey_category=synth_category or "",
+                digikey_description=description,
+                component_type="passive",
+                taxonomy_dir=tax_dir,
+                api_logger=api_logger,
+            )
+        except Exception as exc:
+            log.warning(
+                "lcsc_resolve_passive: auto_resolve_specs failed for %s (lcsc=%s)",
+                mpn, body.lcsc_id, exc_info=True,
+            )
+            raise HTTPException(502, f"Auto-resolve failed: {exc}") from exc
+
+    total_credits = sum(float(e.get("credits_charged") or 0) for e in api_logger.entries)
+    if total_credits > 0:
+        try:
+            get_billing().charge(
+                storage, owner_id, round(total_credits, 4),
+                reason="pipeline_charge",
+                run_id=project_id,
+                unit_id=f"lcsc_resolve_passive:{mpn}",
+                allow_overdraft=False,
+            )
+        except InsufficientCredits as short:
+            raise HTTPException(
+                402,
+                detail={
+                    "reason": "insufficient_credits",
+                    "required": short.required,
+                    "available": short.available,
+                },
+            ) from short
+
+    storage.write_json(project_model_key, model.model_dump())
+    proj_svc.save_to_library(
+        storage, project_model_key, "passives", f"{safe}.json",
+    )
+    try:
+        logs_key = f"{proj_svc.project_prefix(owner_id, project_id)}/api_logs.jsonl"
+        existing = storage.read_text(logs_key) if storage.exists(logs_key) else ""
+        appended = existing + api_logger.to_jsonl()
+        if appended:
+            storage.write_text(logs_key, appended)
+    except Exception:
+        log.warning("lcsc_resolve_passive: failed to append api_logs.jsonl", exc_info=True)
+
+    try:
+        from backend.services.api_logs import total_cost as _total_cost
+
+        added_cost = _total_cost(api_logger.entries)
+        if added_cost > 0:
+            current_total = float(meta.total_cost_usd or 0)
+            proj_svc.update_project(
+                storage, owner_id, project_id,
+                total_cost_usd=round(current_total + added_cost, 6),
+                credits_spent=round(float(meta.credits_spent or 0) + total_credits, 4),
+            )
+    except Exception:
+        log.warning("lcsc_resolve_passive: failed to update total_cost_usd", exc_info=True)
+
+    return _passive_hit_json(
+        mpn, safe, model.model_dump(),
+        cached=False, lcsc_id=body.lcsc_id,
+    )

periscope/src/backend/routers/reports.py

@@ -0,0 +1,378 @@
+"""Serve reports, graphs, datasheets, and finding review mutations."""
+from __future__ import annotations
+
+import json
+import logging
+import re
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import JSONResponse, Response
+from pydantic import BaseModel
+
+from backend.periscopex.finding_engine import (
+    apply_decisions,
+    complete_findings,
+    decision_from_review,
+    sort_findings,
+    upsert_decision,
+)
+from backend.periscopex.models import Finding
+from backend.periscopex.pcb_checks import merge_schema_pcb_reports
+from backend.periscopex.review_workflow import (
+    ReviewError,
+    apply_review_state,
+    build_eco,
+    eco_csv,
+    sign_report,
+)
+from backend.periscopex.utils import safe_mpn
+from backend.routers.deps import get_storage, get_user_id, resolve_or_404
+from backend.services import projects as proj_svc
+
+log = logging.getLogger(__name__)
+router = APIRouter(tags=["reports"])
+
+_MPN_OK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_\-\.:/ +#,()]*$")
+
+
+def _require_mpn(mpn: str) -> None:
+    if not _MPN_OK.match(mpn) or ".." in mpn:
+        raise HTTPException(400, "Invalid MPN format")
+
+
+def _prefix(owner: str, project_id: str) -> str:
+    return proj_svc.project_prefix(owner, project_id)
+
+
+def _load_report(storage, owner: str, project_id: str) -> tuple[str, dict]:
+    key = f"{_prefix(owner, project_id)}/report.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Report not found")
+    return key, storage.read_json(key)
+
+
+def _findings_from_report(report_data: dict) -> list[Finding]:
+    out: list[Finding] = []
+    for raw in report_data.get("findings") or []:
+        try:
+            out.append(Finding.model_validate(raw))
+        except Exception:
+            log.warning("Skipping malformed finding in report", exc_info=True)
+    return out
+
+
+@router.get("/report/{project_id}")
+async def get_report(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    prefix = _prefix(owner, project_id)
+    schema_key = f"{prefix}/report.json"
+    pcb_key = f"{prefix}/pcb_report.json"
+    schema = storage.read_json(schema_key) if storage.exists(schema_key) else None
+    pcb = storage.read_json(pcb_key) if storage.exists(pcb_key) else None
+    merged = merge_schema_pcb_reports(schema, pcb)
+    if merged is None:
+        raise HTTPException(404, "Report not found — run the pipeline first")
+    findings = _findings_from_report(merged)
+    try:
+        complete_findings(findings)
+    except Exception:
+        log.exception("complete_findings failed while serving report %s", project_id)
+        sort_findings(findings)
+    dec_key = f"{prefix}/decisions.json"
+    if storage.exists(dec_key):
+        try:
+            apply_decisions(findings, storage.read_json(dec_key) or [])
+        except Exception:
+            log.exception("apply_decisions failed for report %s", project_id)
+    merged["findings"] = [json.loads(f.model_dump_json()) for f in findings]
+    summary = {"ERROR": 0, "WARNING": 0, "INFO": 0, "total": len(findings)}
+    for finding in findings:
+        if finding.status in summary:
+            summary[finding.status] += 1
+    merged["summary"] = summary
+    return JSONResponse(merged)
+
+
+@router.get("/report/{project_id}/cad-bridge")
+async def get_cad_bridge(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{_prefix(owner, project_id)}/periscope-findings.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "CAD bridge not found — run the pipeline first")
+    return JSONResponse(storage.read_json(key))
+
+
+class AddCommentBody(BaseModel):
+    finding_id: str
+    text: str
+    user_name: str
+    mentions: list[str] = []
+
+
+@router.post("/report/{project_id}/comments")
+async def add_comment(project_id: str, body: AddCommentBody, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    user_id = get_user_id(request)
+    key = f"{_prefix(owner, project_id)}/report.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Report not found")
+    report = storage.read_json(key)
+    comment = {
+        "comment_id": str(uuid.uuid4()),
+        "finding_id": body.finding_id,
+        "user_id": user_id,
+        "user_name": body.user_name,
+        "text": body.text,
+        "mentions": body.mentions,
+        "created_at": datetime.now(timezone.utc).isoformat(),
+    }
+    bucket = report.setdefault("comments", {})
+    bucket.setdefault(body.finding_id, []).append(comment)
+    storage.write_json(key, report)
+    return JSONResponse(comment, status_code=201)
+
+
+@router.delete("/report/{project_id}/comments/{comment_id}")
+async def delete_comment(project_id: str, comment_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    user_id = get_user_id(request)
+    key = f"{_prefix(owner, project_id)}/report.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Report not found")
+    report = storage.read_json(key)
+    comments = report.get("comments", {})
+    found = False
+    for finding_id, rows in list(comments.items()):
+        for i, row in enumerate(rows):
+            if row.get("comment_id") != comment_id:
+                continue
+            found = True
+            if row.get("user_id") != user_id and user_id != owner:
+                raise HTTPException(403, "Cannot delete another user's comment")
+            rows.pop(i)
+            if not rows:
+                del comments[finding_id]
+            storage.write_json(key, report)
+            return JSONResponse({"ok": True})
+    if not found:
+        raise HTTPException(404, "Comment not found")
+    return JSONResponse({"ok": True})
+
+
+class ReviewBody(BaseModel):
+    state: str
+    reason: str = ""
+    user_name: str = ""
+
+
+@router.put("/report/{project_id}/findings/{finding_id}/review")
+async def put_finding_review(project_id: str, finding_id: str, body: ReviewBody, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    user_id = get_user_id(request)
+    key, report = _load_report(storage, owner, project_id)
+    prefix = _prefix(owner, project_id)
+    findings = _findings_from_report(report)
+    ids = {f.finding_id for f in findings if f.finding_id}
+    if finding_id not in ids:
+        pcb_key = f"{prefix}/pcb_report.json"
+        if storage.exists(pcb_key):
+            pcb = storage.read_json(pcb_key)
+            pcb_findings = _findings_from_report(pcb)
+            if finding_id in {f.finding_id for f in pcb_findings if f.finding_id}:
+                key, report, findings = pcb_key, pcb, pcb_findings
+                ids = {f.finding_id for f in findings if f.finding_id}
+    if finding_id not in ids:
+        raise HTTPException(404, "Finding not found")
+    try:
+        states = apply_review_state(
+            report.get("review_states") or {},
+            finding_id,
+            state=body.state,
+            reason=body.reason,
+            user_id=user_id,
+            user_name=body.user_name,
+        )
+    except ReviewError as exc:
+        raise HTTPException(400, str(exc)) from exc
+    report["review_states"] = states
+    storage.write_json(key, report)
+    if body.state in {"wontfix", "false_positive"}:
+        found = next(
+            (f for f in _findings_from_report(report) if f.finding_id == finding_id),
+            None,
+        )
+        if found is not None:
+            dec = decision_from_review(
+                found, state=body.state, reason=body.reason, user_id=user_id
+            )
+            if dec is not None:
+                dkey = f"{prefix}/decisions.json"
+                existing = storage.read_json(dkey) if storage.exists(dkey) else []
+                if not isinstance(existing, list):
+                    existing = []
+                storage.write_json(dkey, upsert_decision(existing, dec))
+    return JSONResponse(states.get(finding_id) or {"state": "open", "reason": ""})
+
+
+@router.get("/report/{project_id}/eco.json")
+async def get_eco_json(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    _, report = _load_report(storage, owner, project_id)
+    return JSONResponse(
+        {"items": build_eco(_findings_from_report(report), report.get("review_states") or {})}
+    )
+
+
+@router.get("/report/{project_id}/eco.csv")
+async def get_eco_csv(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    _, report = _load_report(storage, owner, project_id)
+    items = build_eco(_findings_from_report(report), report.get("review_states") or {})
+    return Response(eco_csv(items), media_type="text/csv")
+
+
+@router.post("/report/{project_id}/sign")
+async def post_sign_report(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    user_id = get_user_id(request)
+    key, report = _load_report(storage, owner, project_id)
+    release = sign_report(report, user_id=user_id)
+    report["release"] = release
+    storage.write_json(key, report)
+    return JSONResponse(release)
+
+
+@router.get("/bom/{project_id}")
+async def get_bom_summary(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{_prefix(owner, project_id)}/bom_summary.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "BOM summary not found — run the pipeline first")
+    return JSONResponse(storage.read_json(key))
+
+
+@router.get("/derating/{project_id}")
+async def get_derating(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{_prefix(owner, project_id)}/derating.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Derating data not found — run the pipeline first")
+    return JSONResponse(storage.read_json(key))
+
+
+@router.get("/graph/{project_id}")
+async def get_graph(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{_prefix(owner, project_id)}/design_graph.json"
+    if not storage.exists(key):
+        raise HTTPException(404, "Design graph not found — run the pipeline first")
+    return JSONResponse(storage.read_json(key))
+
+
+@router.get("/projects/{project_id}/logs")
+async def get_project_logs(project_id: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    key = f"{_prefix(owner, project_id)}/api_logs.jsonl"
+    if not storage.exists(key):
+        return JSONResponse([])
+    text = storage.read_text(key)
+    entries = [json.loads(line) for line in text.strip().split("\n") if line.strip()]
+    from backend.services.llm.pricing import cost_for_entry
+
+    for entry in entries:
+        if any(
+            entry.get(k)
+            for k in (
+                "input_tokens",
+                "output_tokens",
+                "cache_read_input_tokens",
+                "cache_creation_input_tokens",
+            )
+        ):
+            entry["cost_usd"] = round(cost_for_entry(entry), 6)
+    return JSONResponse(entries)
+
+
+def _find_datasheet_key(storage, owner: str, project_id: str, safe: str, mpn: str | None = None) -> str | None:
+    from backend.services.datasheet_store import resolve_datasheet
+
+    uploaded = f"{_prefix(owner, project_id)}/uploads/datasheets/{safe}.pdf"
+    if storage.exists(uploaded):
+        return uploaded
+    resolved = resolve_datasheet(storage, safe)
+    if resolved:
+        return resolved
+    legacy = f"library/datasheets/{safe}.pdf"
+    if storage.exists(legacy):
+        return legacy
+    if mpn:
+        return proj_svc.library_has_datasheet(storage, mpn)
+    return None
+
+
+@router.get("/projects/{project_id}/datasheet-url/{mpn:path}")
+async def get_datasheet_url(project_id: str, mpn: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    _require_mpn(mpn)
+    key = _find_datasheet_key(storage, owner, project_id, safe_mpn(mpn), mpn=mpn)
+    if key is None:
+        raise HTTPException(404, f"Datasheet not found for MPN: {mpn}")
+    proxy = f"/api/projects/{project_id}/datasheet/{mpn}"
+    return {"url": f"{str(request.base_url).rstrip('/')}{proxy}"}
+
+
+@router.get("/projects/{project_id}/datasheet/{mpn:path}")
+async def get_datasheet_proxy(project_id: str, mpn: str, request: Request):
+    storage = get_storage(request)
+    owner, _ = await resolve_or_404(request, project_id)
+    _require_mpn(mpn)
+    safe = safe_mpn(mpn)
+    key = _find_datasheet_key(storage, owner, project_id, safe, mpn=mpn)
+    if key is None:
+        raise HTTPException(404, f"Datasheet not found for MPN: {mpn}")
+    return Response(
+        content=storage.read_bytes(key),
+        media_type="application/pdf",
+        headers={"Content-Disposition": f'inline; filename="{safe}.pdf"'},
+    )
+
+
+@router.get("/datasheets/{mpn}")
+async def get_datasheet(mpn: str, request: Request):
+    from fastapi.responses import FileResponse
+
+    from backend.services.storage import LocalStorageBackend
+
+    storage = get_storage(request)
+    user_id = get_user_id(request)
+    _require_mpn(mpn)
+    safe = safe_mpn(mpn)
+    if not isinstance(storage, LocalStorageBackend):
+        raise HTTPException(
+            400,
+            "Use GET /projects/{project_id}/datasheet-url/{mpn} for cloud storage",
+        )
+    for entry in storage.list_prefix(f"users/{user_id}/projects/"):
+        pdf_key = f"{entry}/uploads/datasheets/{safe}.pdf"
+        if storage.exists(pdf_key):
+            return FileResponse(
+                storage._path(pdf_key),
+                media_type="application/pdf",
+                filename=f"{safe}.pdf",
+            )
+    raise HTTPException(404, f"Datasheet not found for MPN: {mpn}")

periscope/src/backend/routers/survey.py

@@ -0,0 +1,69 @@
+"""Onboarding survey: status flag plus one-shot Google Sheet append."""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Request
+from pydantic import BaseModel
+
+from backend.config import settings
+from backend.routers.deps import get_storage, get_user_id
+from backend.services import survey as survey_svc
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/survey", tags=["survey"])
+
+
+class SurveySubmission(BaseModel):
+    referral_source: str
+    user_profile: str
+
+
+@router.get("/status")
+async def survey_status(request: Request):
+    return {
+        "completed": survey_svc.is_completed(get_storage(request), get_user_id(request))
+    }
+
+
+async def _identity(user_id: str) -> tuple[str, str]:
+    email, name = "unknown", "unknown"
+    if not settings.use_auth:
+        return email, name
+    try:
+        from backend.services.email import _resolve_clerk_user
+
+        clerk_user = await _resolve_clerk_user(user_id)
+    except Exception:
+        logger.warning("Failed to resolve Clerk user %s for survey", user_id)
+        return email, name
+    if not clerk_user:
+        return email, name
+    emails = clerk_user.get("email_addresses") or []
+    if emails:
+        email = emails[0].get("email_address") or "unknown"
+    first = clerk_user.get("first_name") or ""
+    last = clerk_user.get("last_name") or ""
+    name = f"{first} {last}".strip() or "unknown"
+    return email, name
+
+
+@router.post("")
+async def submit_survey(request: Request, body: SurveySubmission):
+    storage = get_storage(request)
+    user_id = get_user_id(request)
+    if survey_svc.is_completed(storage, user_id):
+        return {"ok": True, "detail": "already_submitted"}
+
+    email, name = await _identity(user_id)
+    sheet_ok = await survey_svc.append_to_sheet(
+        user_id=user_id,
+        email=email,
+        name=name,
+        referral_source=body.referral_source,
+        user_profile=body.user_profile,
+    )
+    if sheet_ok or not settings.survey_sheet_id:
+        survey_svc._mark_completed(storage, user_id)
+        return {"ok": True}
+    return {"ok": False, "detail": "sheet_write_failed"}

tests/test_periscope_routers_rewrite.py

@@ -0,0 +1,49 @@
+"""HTTP routers resolve from periscope/src (auth stays leftover)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import backend.routers.admin as admin
+import backend.routers.contact as contact
+import backend.routers.deps as deps
+import backend.routers.feedback as feedback
+import backend.routers.pipeline as pipeline
+import backend.routers.projects as projects
+import backend.routers.reports as reports
+import backend.routers.survey as survey
+
+
+def _src(mod, name: str) -> None:
+    path = Path(mod.__file__).resolve()
+    assert path.name == name
+    assert "src" in path.parts
+    assert "Native Periscope overlay" not in path.read_text(encoding="utf-8")[:400]
+
+
+def test_inherited_routers_are_src():
+    for mod, name in (
+        (admin, "admin.py"),
+        (contact, "contact.py"),
+        (deps, "deps.py"),
+        (feedback, "feedback.py"),
+        (pipeline, "pipeline.py"),
+        (projects, "projects.py"),
+        (reports, "reports.py"),
+        (survey, "survey.py"),
+    ):
+        _src(mod, name)
+
+
+def test_auth_router_is_not_rewritten_this_slice():
+    import backend.routers.auth as auth
+
+    path = Path(auth.__file__).resolve()
+    assert path.name == "auth.py"
+    assert "src" in path.parts
+
+
+def test_reprocess_helper_names():
+    assert callable(pipeline._await_terminal)
+    assert callable(pipeline._project_active)
+    assert callable(projects._bom_file_to_csv_bytes)