diff --git a/periscope/src/backend/routers/admin.py b/periscope/src/backend/routers/admin.py
new file mode 100644
index 0000000..b51790a
--- /dev/null
+++ b/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,
+ }
diff --git a/periscope/src/backend/routers/contact.py b/periscope/src/backend/routers/contact.py
new file mode 100644
index 0000000..a461634
--- /dev/null
+++ b/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 (
+ "
"
+ f'| {html.escape(label)} | '
+ f'{value} | '
+ "
"
+ )
+
+ rows = cell("Name", html.escape(data.name))
+ email = html.escape(data.email)
+ rows += (
+ ""
+ '| Email | '
+ f'{email} | '
+ "
"
+ )
+ 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"""\
+
+
New contact form submission
+
+
{body}
+
Sent from the Periscope contact form
+
"""
+ 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)
diff --git a/periscope/src/backend/routers/deps.py b/periscope/src/backend/routers/deps.py
new file mode 100644
index 0000000..074831f
--- /dev/null
+++ b/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")
diff --git a/periscope/src/backend/routers/feedback.py b/periscope/src/backend/routers/feedback.py
new file mode 100644
index 0000000..faf0465
--- /dev/null
+++ b/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
diff --git a/periscope/src/backend/routers/pipeline.py b/periscope/src/backend/routers/pipeline.py
new file mode 100644
index 0000000..10899f1
--- /dev/null
+++ b/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())
diff --git a/periscope/src/backend/routers/projects.py b/periscope/src/backend/routers/projects.py
new file mode 100644
index 0000000..426ba56
--- /dev/null
+++ b/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,
+ )
diff --git a/periscope/src/backend/routers/reports.py b/periscope/src/backend/routers/reports.py
new file mode 100644
index 0000000..c95a4b7
--- /dev/null
+++ b/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}")
diff --git a/periscope/src/backend/routers/survey.py b/periscope/src/backend/routers/survey.py
new file mode 100644
index 0000000..b1378ed
--- /dev/null
+++ b/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"}
diff --git a/tests/test_periscope_routers_rewrite.py b/tests/test_periscope_routers_rewrite.py
new file mode 100644
index 0000000..907bd38
--- /dev/null
+++ b/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)