Overlay leftover FastAPI entry, routers, and jobs into periscope/src (2.45.0).

Copy main, config, middleware/auth, leftover routers, pipeline_worker,
job_runner, event_bridge, projects, and admin_settings without deleting
dependency copies. Native auth and impedance routers stay as-is.
This commit is contained in:
2026-09-20 17:39:15 +02:00
parent 49a27d5951
commit 6d6e70cec0
20 changed files with 6614 additions and 2 deletions
@@ -2,6 +2,13 @@
What's new in Periscope.
## 2.45.0 — 2026-09-20 — Overlay leftover app entry, routers, and jobs
`main.py`, `config.py`, `middleware/auth.py`, leftover `routers/*` (not native `auth`/`impedance`), `pipeline_worker.py`, `job_runner`, `event_bridge`, `projects`, and `admin_settings` resolve from `periscope/src`. Inherited copies stay on disk. Auth users and `AUTH_JWT_SECRET` unchanged.
- [New] src overlay of the leftover FastAPI entrypoint, settings, JWT middleware, and HTTP routers the live app still imported from `dependency/`.
- [New] src overlay of `pipeline_worker` and job/project settings services.
## 2.44.0 — 2026-09-20 — Overlay leftover services src still imported
Pipeline helpers still pulled from `dependency/` (`api_logs`, `normalize_findings`, `dedupe_findings`, `billing_hook`, `datasheet_store`, `cost_estimator`, `storage`, `purple_parts`, `email`, `digikey`) plus LLM `factory` / `types` / `base` now resolve from `periscope/src`. Inherited copies stay on disk.
+335
View File
@@ -0,0 +1,335 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Backend configuration via environment variables."""
import importlib.util
import json
from pathlib import Path
from pydantic import Field
from pydantic_settings import BaseSettings
from backend.repo_paths import data_dir as _data_dir
from backend.repo_paths import env_file as _env_file
from backend.repo_paths import skills_dir as _skills_dir
from backend.repo_paths import taxonomy_dir as _taxonomy_dir
# Resolve paths relative to the project root (one level up from backend/)
_BACKEND_DIR = Path(__file__).resolve().parent
_PROJECT_ROOT = _BACKEND_DIR.parent
# Load skills manifest once at import time
_MANIFEST_PATH = _BACKEND_DIR / "skills_manifest.json"
_SKILLS_MANIFEST: dict = (
json.loads(_MANIFEST_PATH.read_text()) if _MANIFEST_PATH.exists() else {}
)
class Settings(BaseSettings):
# DeepSeek (default provider — OpenAI-compatible Chat Completions)
deepseek_api_key: str = ""
deepseek_base_url: str = "https://api.deepseek.com"
deepseek_model: str = "deepseek-flash"
deepseek_vision_model: str = "deepseek-flash"
# "enabled" (default) or "disabled". DeepSeek V4 thinks by default;
# disable to cut cost on simple mapping calls.
deepseek_thinking: str = "enabled"
# Official values: low | high | max. Review sessions with
# max_tokens >= 16000 still bump to "high" in the provider.
deepseek_reasoning_effort: str = "high"
# PDF ingest: DeepSeek does not accept native PDFs. Text is always
# extracted; page images are attached only when the stage model is a
# vision model (see model_*_deepseek defaults below).
deepseek_pdf_max_chars: int = 500_000
deepseek_pdf_image_pages: int = 32
# Per-stage DeepSeek model overrides (fall back to deepseek_model)
model_pintable_deepseek: str = "deepseek-flash"
model_pattern_deepseek: str = "deepseek-flash"
model_specs_deepseek: str = "deepseek-flash"
model_validation_deepseek: str = "deepseek-flash"
model_auto_resolve_deepseek: str = "deepseek-flash"
model_normalize_deepseek: str = "deepseek-flash"
# Anthropic (optional fallback)
anthropic_api_key: str = ""
anthropic_model: str = "claude-sonnet-4-6"
# Per-stage model overrides (fall back to anthropic_model if empty)
model_pintable: str = ""
model_pattern: str = ""
model_specs: str = ""
model_validation: str = "claude-sonnet-4-6"
model_auto_resolve: str = "claude-haiku-4-5-20251001"
model_normalize: str = "claude-sonnet-4-6"
# Gemini (set GEMINI_API_KEY to enable)
gemini_api_key: str = ""
gemini_model: str = "gemini-3.1-pro-preview"
# Per-stage Gemini model overrides (fall back to gemini_model if empty)
model_validation_gemini: str = ""
model_pintable_gemini: str = ""
model_pattern_gemini: str = ""
model_specs_gemini: str = ""
model_auto_resolve_gemini: str = ""
model_normalize_gemini: str = ""
# Provider routing — provider_default is the global default; per-stage
# overrides win when non-empty. Valid values: deepseek | anthropic | gemini.
provider_default: str = "deepseek"
provider_pintable: str = ""
provider_pattern: str = ""
provider_specs: str = ""
provider_validation: str = ""
provider_auto_resolve: str = ""
provider_normalize: str = ""
# Per-stage fallback provider/model — used if the primary stage call
# raises (e.g. DeepSeek 503). Leave empty to disable fallback
# for that stage. If fallback_provider_<stage> is set but
# fallback_model_<stage> is empty, the fallback uses that provider's
# default model (deepseek_model, anthropic_model, or gemini_model).
fallback_provider_pintable: str = ""
fallback_provider_pattern: str = ""
fallback_provider_specs: str = ""
fallback_provider_validation: str = ""
fallback_provider_auto_resolve: str = ""
fallback_provider_normalize: str = ""
fallback_model_pintable: str = ""
fallback_model_pattern: str = ""
fallback_model_specs: str = ""
fallback_model_validation: str = ""
fallback_model_auto_resolve: str = ""
fallback_model_normalize: str = ""
# Max parallel IC agents — the single knob controlling concurrency for
# BOTH the IC pintable extraction stage and the direct datasheet review
# stage. Change this one number (or the IC_CONCURRENCY env var) to scale
# how many ICs are processed in parallel.
ic_concurrency: int = 6
# Per-IC normalize pass — dedup findings sharing a root cause and
# re-grade severity against a fixed rubric. Runs after submit_review.
normalize_findings_enabled: bool = True
# Cross-IC dedup pass — collapse one physical interface defect reported
# from both ICs (e.g. a 5V-into-3V3 net flagged once per endpoint) into a
# single finding. Runs once after all per-IC reviews complete.
cross_ic_dedup_enabled: bool = True
# Paths (git split: data at repo root; taxonomy/skills under periscope/dependency)
data_dir: Path = _data_dir()
taxonomy_dir: Path = _taxonomy_dir()
skills_dir: Path = _skills_dir()
# GCS (if set, use GCSStorageBackend; otherwise LocalStorageBackend)
gcs_bucket: str = ""
# Clerk authentication (cloud). When set, takes priority over local auth.
clerk_secret_key: str = ""
clerk_publishable_key: str = ""
clerk_jwks_url: str = ""
# Local Periscope auth (self-host). Set AUTH_JWT_SECRET to enable email/password
# accounts and multi-user project collaborators without Clerk.
auth_jwt_secret: str = ""
# Comma-separated emails that become admin on register (in addition to the
# first account, which is always admin).
auth_admin_emails: str = ""
# DigiKey API (optional — enables auto-fetch datasheets)
digikey_client_id: str = ""
digikey_client_secret: str = ""
digikey_environment: str = "production"
digikey_locale_site: str = "US"
digikey_locale_language: str = "en"
digikey_locale_currency: str = "USD"
# Mouser Search API (optional — fourth datasheet source)
mouser_api_key: str = ""
# Purple Parts API (optional — converts LCSC codes to MPNs before DigiKey)
purple_parts_url: str = ""
purple_parts_api_key: str = ""
# Email notifications (Gmail API via service account)
email_sender: str = ""
email_frontend_url: str = ""
email_admin_notify: str = "" # fixed recipient for pipeline-started alerts
contact_recipient: str = "" # where /api/contact submissions are delivered
# Stripe billing (pay-as-you-go only — no subscription prices needed)
stripe_secret_key: str = ""
stripe_webhook_secret: str = ""
# Open-core: master switch for the credits/Stripe billing system.
# True = credit gating + charges + billing/credits routers.
# False = OSS/self-host mode: pipelines run free, billing routes unmounted.
# Defaults to whether the private billing modules exist in this checkout
# (present in the cloud/gateway repo, absent in the open-source core), so
# a bare core checkout runs free with no configuration. An explicit
# BILLING_ENABLED env var always wins.
billing_enabled: bool = Field(
default_factory=lambda: importlib.util.find_spec(
"backend.services.stripe_billing"
)
is not None
)
# Onboarding survey (Google Sheet)
survey_sheet_id: str = ""
# CORS
cors_origins: list[str] = [
"http://localhost:3000",
"http://127.0.0.1:3000",
"http://localhost:18742",
"http://127.0.0.1:18742",
]
# Cloud Run Job worker (pipeline runner)
pipeline_worker_job_name: str = "periscopex-pipeline-worker"
pipeline_worker_region: str = "us-central1"
pipeline_worker_project: str = "" # GCP project id; defaults to GOOGLE_CLOUD_PROJECT or metadata
pipeline_worker_timeout_seconds: int = 3600
# Sweeper: a "running" project is considered stale if its last update
# timestamp is older than this and the worker execution is in a
# terminal Cloud Run state (or the executor isn't reachable).
pipeline_sweeper_stale_seconds: int = 60
model_config = {
"env_file": str(_env_file()),
"env_file_encoding": "utf-8",
"extra": "ignore",
}
@property
def use_stripe(self) -> bool:
return bool(self.stripe_secret_key)
@property
def use_digikey(self) -> bool:
return bool(self.digikey_client_id and self.digikey_client_secret)
@property
def use_mouser(self) -> bool:
return bool(self.mouser_api_key)
@property
def use_purple_parts(self) -> bool:
return bool(self.purple_parts_url and self.purple_parts_api_key)
@property
def use_gcs(self) -> bool:
return bool(self.gcs_bucket)
@property
def use_clerk(self) -> bool:
return bool(self.clerk_secret_key and self.clerk_jwks_url)
@property
def use_local_auth(self) -> bool:
"""Self-host email/password auth when JWT secret is set and Clerk is not."""
return bool(self.auth_jwt_secret) and not self.use_clerk
@property
def use_auth(self) -> bool:
return self.use_clerk or self.use_local_auth
@property
def use_email(self) -> bool:
return bool(self.email_sender and self.email_frontend_url)
def provider_for_stage(self, stage: str) -> str:
"""Return the LLM provider name for a pipeline stage.
Anthropic is never used: any ``PROVIDER_*=anthropic`` override is
coerced to DeepSeek.
"""
override = getattr(self, f"provider_{stage}", "")
name = override or self.provider_default
if name == "anthropic":
return "deepseek"
return name
def model_for_stage(self, stage: str) -> str:
"""Return the model for a pipeline stage, provider-aware.
For DeepSeek: falls back to model_<stage>_deepseek, then deepseek_model.
For Gemini: falls back to model_<stage>_gemini, then gemini_model.
For Anthropic: falls back to model_<stage>, then anthropic_model.
"""
provider = self.provider_for_stage(stage)
if provider == "gemini":
override = getattr(self, f"model_{stage}_gemini", "")
return override or self.gemini_model
if provider == "deepseek":
override = getattr(self, f"model_{stage}_deepseek", "")
return override or self.deepseek_model
override = getattr(self, f"model_{stage}", "")
return override or self.anthropic_model
def fallback_for_stage(self, stage: str) -> tuple[str, str] | None:
"""Return (provider, model) for the stage's fallback, or None if no
fallback is configured. Used by call_with_fallback() to retry once
when the primary provider raises.
"""
fb_provider = getattr(self, f"fallback_provider_{stage}", "")
if not fb_provider or fb_provider == "anthropic":
return None
fb_model = getattr(self, f"fallback_model_{stage}", "")
if not fb_model:
if fb_provider == "gemini":
fb_model = self.gemini_model
elif fb_provider == "deepseek":
fb_model = self.deepseek_model
else:
fb_model = self.anthropic_model
return (fb_provider, fb_model)
def default_model_for_provider(self, provider: str) -> str:
if provider == "gemini":
return self.gemini_model
if provider == "deepseek":
return self.deepseek_model
return self.anthropic_model
def has_llm_credentials(self) -> bool:
"""True if the configured default provider has an API key."""
name = self.provider_default
if name == "anthropic":
name = "deepseek"
if name == "deepseek":
return bool(self.deepseek_api_key)
if name == "gemini":
return bool(self.gemini_api_key)
return bool(self.deepseek_api_key)
def get_skill_or_none(self, name: str) -> tuple[str | None, str | None]:
"""Return (skill_id, version) or (None, None) if the Anthropic
Console skill is not in the manifest. DeepSeek/Gemini extraction
inlines SKILL.md locally and does not need a skill_id."""
entry = _SKILLS_MANIFEST.get(name)
if not entry:
return None, None
return entry.get("skill_id"), entry.get("latest_version")
def get_default_model_version(self) -> str:
"""Return the default model_version for new extractions from skills_manifest.json."""
return _SKILLS_MANIFEST.get("default_model_version", "1.0.0")
def get_skill(self, name: str) -> tuple[str, str]:
"""Return (skill_id, version) from skills_manifest.json or raise."""
entry = _SKILLS_MANIFEST.get(name)
if not entry:
raise RuntimeError(
f"Skill '{name}' not found in {_MANIFEST_PATH}. "
f"Run scripts/upload_skills.py to create skills."
)
return entry["skill_id"], entry["latest_version"]
settings = Settings()
+177
View File
@@ -0,0 +1,177 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""PeriscopeX backend — FastAPI application."""
import logging
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from backend.config import settings
from backend.routers import admin, auth, contact, feedback, impedance, pipeline, projects, reports, survey
from backend.services.projects import ProjectNotFound
from backend.services.storage import LocalStorageBackend
logger = logging.getLogger(__name__)
# Default user ID for unauthenticated local dev
LOCAL_DEV_USER = "local"
def _create_storage():
"""Create the appropriate storage backend based on config."""
if settings.use_gcs:
from backend.services.storage_gcs import GCSStorageBackend
return GCSStorageBackend(settings.gcs_bucket)
return LocalStorageBackend(settings.data_dir)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Guard: refuse to start in production without authentication
env = os.getenv("ENVIRONMENT", "").lower()
if env == "production" and not settings.use_auth:
raise RuntimeError(
"Production requires authentication: set AUTH_JWT_SECRET "
"(local Periscope accounts) or CLERK_JWKS_URL + CLERK_SECRET_KEY."
)
if not settings.use_auth:
logger.warning(
"Authentication is DISABLED — all users have full access. "
"This is only safe for local development."
)
elif settings.use_local_auth:
logger.info("Local Periscope authentication enabled (AUTH_JWT_SECRET)")
elif settings.use_clerk:
logger.info("Clerk authentication enabled")
if not settings.billing_enabled:
logger.warning(
"Billing is DISABLED — pipelines run free and the billing/credits "
"routes are not mounted."
)
app.state.storage = _create_storage()
# For local backend, ensure base directories exist
if isinstance(app.state.storage, LocalStorageBackend):
base = settings.data_dir
(base / "users").mkdir(parents=True, exist_ok=True)
(base / "auth" / "users").mkdir(parents=True, exist_ok=True)
(base / "auth" / "by_email").mkdir(parents=True, exist_ok=True)
(base / "library" / "extracted").mkdir(parents=True, exist_ok=True)
(base / "library" / "patterns").mkdir(parents=True, exist_ok=True)
(base / "library" / "models").mkdir(parents=True, exist_ok=True)
(base / "library" / "passives").mkdir(parents=True, exist_ok=True)
(base / "library" / "datasheets" / "refs").mkdir(parents=True, exist_ok=True)
(base / "library" / "datasheets" / "blobs").mkdir(parents=True, exist_ok=True)
yield
# Pipelines run in a separate Cloud Run Job worker (or local
# subprocess in dev), so the API process has nothing to clean up
# on shutdown.
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
"""Add standard security headers to all responses."""
async def dispatch(self, request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
if settings.use_auth:
# Only set HSTS when running behind TLS in production
response.headers["Strict-Transport-Security"] = (
"max-age=31536000; includeSubDomains"
)
return response
class AuthMiddleware(BaseHTTPMiddleware):
"""Extract user_id from JWT (Clerk or local) or default to local dev user."""
async def dispatch(self, request: Request, call_next):
# Let CORS preflight through — browsers send OPTIONS without credentials
if request.method == "OPTIONS":
return await call_next(request)
# Public endpoints that don't require authentication
if request.url.path in {
"/api/contact",
"/api/auth/mode",
"/api/auth/register",
"/api/auth/login",
}:
request.state.user_id = LOCAL_DEV_USER
return await call_next(request)
if settings.use_auth:
from backend.middleware.auth import verify_request_user
user_id = await verify_request_user(request)
if user_id is None:
is_production = os.getenv("ENVIRONMENT", "").lower() == "production"
# Local auth (and production) require a valid token for API routes.
if is_production or settings.use_local_auth:
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=401,
content={"detail": "Authentication required"},
)
# Non-production Clerk: fall back so missing token doesn't block
# local development when Clerk is configured but unused.
user_id = LOCAL_DEV_USER
request.state.user_id = user_id
else:
request.state.user_id = LOCAL_DEV_USER
response = await call_next(request)
return response
app = FastAPI(
title="PeriscopeX",
description="Agentic schematic validation API",
lifespan=lifespan,
)
# Middleware order matters: Starlette applies in LIFO order (last added =
# outermost). CORSMiddleware MUST be outermost so that CORS headers are
# present on every response — including 401s from AuthMiddleware.
app.add_middleware(AuthMiddleware)
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers=["content-type", "authorization"],
expose_headers=["X-Datasheet-Url", "X-Datasheet-Source"],
)
@app.exception_handler(ProjectNotFound)
async def _project_not_found_handler(request: Request, exc: ProjectNotFound):
# A mutation raced a project deletion (or hit never-fully-created metadata).
# Return a clean 404 — CORSMiddleware is outermost, so headers still land.
return JSONResponse(status_code=404, content={"detail": str(exc)})
app.include_router(projects.router, prefix="/api")
app.include_router(pipeline.router, prefix="/api")
app.include_router(reports.router, prefix="/api")
app.include_router(impedance.router, prefix="/api")
app.include_router(admin.router, prefix="/api")
app.include_router(auth.router, prefix="/api")
if settings.billing_enabled:
# Import guarded too: with billing disabled the core never loads the
# billing/credits routers (or, transitively, the Stripe SDK).
from backend.routers import billing, credits
app.include_router(billing.router, prefix="/api")
app.include_router(credits.router, prefix="/api")
app.include_router(contact.router, prefix="/api")
app.include_router(feedback.router, prefix="/api")
app.include_router(survey.router, prefix="/api")
@@ -0,0 +1,2 @@
# Native Periscope overlay: middleware package.
# PinScope original remains in dependency/.
+118
View File
@@ -0,0 +1,118 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""JWT verification for FastAPI (Clerk JWKS or local Periscope HS256)."""
from __future__ import annotations
from typing import Any
import jwt
from fastapi import Request
from backend.config import settings
# JWKS cache
_jwks_client: jwt.PyJWKClient | None = None
_SKIP_PATHS = {
"/docs",
"/openapi.json",
"/redoc",
"/health",
"/api/billing/webhook",
"/api/auth/mode",
"/api/auth/register",
"/api/auth/login",
"/api/contact",
}
def _get_jwks_client() -> jwt.PyJWKClient:
global _jwks_client
if _jwks_client is None:
jwks_url = settings.clerk_jwks_url
if not jwks_url:
raise RuntimeError(
"CLERK_JWKS_URL must be set for Clerk authentication. "
"Find it in your Clerk dashboard under API Keys."
)
_jwks_client = jwt.PyJWKClient(jwks_url, cache_keys=True)
return _jwks_client
def _bearer_or_query_token(request: Request) -> str | None:
auth_header = request.headers.get("authorization", "")
if auth_header.startswith("Bearer "):
return auth_header[7:]
# EventSource/SSE can't send headers
return request.query_params.get("token")
async def verify_clerk_token(request: Request) -> str | None:
"""Verify Clerk JWT and return user_id, or None if invalid."""
if request.url.path in _SKIP_PATHS:
return "anonymous"
token = _bearer_or_query_token(request)
if not token:
return None
try:
client = _get_jwks_client()
signing_key = client.get_signing_key_from_jwt(token)
payload: dict[str, Any] = jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
options={
"verify_exp": True,
"verify_aud": False,
"verify_iss": True,
},
issuer=(
settings.clerk_jwks_url.replace("/.well-known/jwks.json", "")
if settings.clerk_jwks_url
else None
),
leeway=10,
)
user_id = payload.get("sub")
if not user_id:
return None
return user_id
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
except Exception:
return None
async def verify_local_token(request: Request) -> str | None:
"""Verify Periscope local JWT and return user_id, or None if invalid."""
if request.url.path in _SKIP_PATHS:
return "anonymous"
token = _bearer_or_query_token(request)
if not token:
return None
from backend.services.local_jwt import decode_token
payload = decode_token(token)
if not payload:
return None
user_id = payload.get("sub")
return str(user_id) if user_id else None
async def verify_request_user(request: Request) -> str | None:
"""Dispatch to Clerk or local JWT verification."""
if settings.use_clerk:
return await verify_clerk_token(request)
if settings.use_local_auth:
return await verify_local_token(request)
return None
+139
View File
@@ -0,0 +1,139 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Pipeline worker entrypoint — runs as a Cloud Run Job execution.
Invoked by Cloud Run Jobs (prod) or as a child subprocess (local dev).
Reads execution parameters from environment variables, swaps the
in-memory event broker for the GCS-backed one, and dispatches to either
``run_pipeline`` (full run) or ``run_regen_pipeline`` (admin regen).
Required env vars:
PROJECT_ID — project to run
USER_ID — owner user id (Clerk sub or "local")
Optional env vars:
RESUME "1"/"0" — resume a paused run from its checkpoint
FREE "1"/"0" — admin-initiated free run (no charge)
MODE "run" (default) | "regen"
REGEN_STAGES comma-separated, e.g. "derating" (regen mode only)
EXECUTION_NAME Cloud Run execution resource name (purely for log
correlation — the API already wrote it onto
``ProjectMeta.execution_name`` at enqueue time)
This module **must not** import :mod:`backend.main` — the FastAPI
lifespan would attempt to wire up shutdown handlers we don't want here.
"""
from __future__ import annotations
import asyncio
import logging
import os
import sys
from backend.config import settings
from backend.services import event_bridge as event_bridge
from backend.services import pipeline as pipeline_svc
from backend.services.storage import LocalStorageBackend, StorageBackend
def _build_storage() -> StorageBackend:
if settings.use_gcs:
from backend.services.storage_gcs import GCSStorageBackend
return GCSStorageBackend(settings.gcs_bucket)
return LocalStorageBackend(settings.data_dir)
def _required_env(name: str) -> str:
val = os.environ.get(name, "").strip()
if not val:
raise SystemExit(f"missing required env var: {name}")
return val
def _bool_env(name: str, default: bool = False) -> bool:
raw = os.environ.get(name, "").strip().lower()
if not raw:
return default
return raw in ("1", "true", "yes", "on")
async def _run() -> None:
project_id = _required_env("PROJECT_ID")
user_id = _required_env("USER_ID")
resume = _bool_env("RESUME")
free = _bool_env("FREE")
mode = os.environ.get("MODE", "run").strip().lower() or "run"
execution_name = os.environ.get("EXECUTION_NAME", "").strip()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [worker %(name)s] %(message)s",
)
log = logging.getLogger("backend.pipeline_worker")
log.info(
"starting worker mode=%s project=%s user=%s resume=%s free=%s execution=%s",
mode, project_id, user_id, resume, free, execution_name or "(none)",
)
storage = _build_storage()
# Swap in the GCS-backed broker so events written from this process
# are visible to any API instance tailing the event log.
pipeline_svc.set_broker(event_bridge.GCSEventBroker(storage, user_id))
# Always wipe the prior event log. Reprocess uses resume=True, and
# the old log still contains ``pipeline_complete``; the SSE tail
# would stop there and the UI would show a finished run with no live
# log while the worker is still reviewing. Pause-resume also hits a
# terminal ``pipeline_paused``. A fresh seq from 0 is the only safe
# option — completed_review_refs still skip paid ICs.
pipeline_svc.broker.clear_history(project_id)
if mode == "run":
await pipeline_svc.run_pipeline(
storage, user_id, project_id, resume=resume, free=free,
)
elif mode == "regen":
stages_raw = os.environ.get("REGEN_STAGES", "").strip()
stages = [s for s in (s.strip() for s in stages_raw.split(",")) if s]
if not stages:
raise SystemExit("REGEN_STAGES must list at least one stage in regen mode")
await pipeline_svc.run_regen_pipeline(
storage, user_id, project_id, stages,
)
elif mode == "placement":
from backend.services import placement_pipeline as placement_svc
await placement_svc.run_placement_pipeline(storage, user_id, project_id)
elif mode == "pcb":
from backend.services import pcb_pipeline as pcb_svc
await pcb_svc.run_pcb_pipeline(storage, user_id, project_id)
else:
raise SystemExit(
f"unknown MODE={mode!r}; expected 'run', 'regen', 'placement', or 'pcb'"
)
def main() -> None:
try:
asyncio.run(_run())
except SystemExit:
raise
except KeyboardInterrupt:
# Local dev convenience — the run_pipeline cancel handler will
# have already transitioned the project on SIGTERM.
sys.exit(130)
except BaseException as exc: # pragma: no cover — last-mile safety
# The pipeline's own ``except Exception`` already logs and writes
# ``status=error`` for the project. This catch only exists so a
# truly unhandled BaseException (e.g. SystemExit during boot
# before run_pipeline starts) still surfaces as a non-zero exit
# code, which Cloud Run records as "Failed" on the execution.
logging.exception("worker crashed before run_pipeline cleanup: %s", exc)
sys.exit(1)
if __name__ == "__main__":
main()
+684
View File
@@ -0,0 +1,684 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Admin endpoints — library components, user management, and limits.
All endpoints require the requesting user to have role: "admin" in their
Clerk public metadata. In local dev (no auth), all requests are treated as admin.
"""
from __future__ import annotations
import asyncio
import re
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
from backend.services import admin_settings as settings_svc
from backend.services.billing_hook import get_billing
from backend.services import projects as proj_svc
router = APIRouter(prefix="/admin", tags=["admin"])
# ---------------------------------------------------------------------------
# Admin verification
# ---------------------------------------------------------------------------
async def is_admin(request: Request) -> bool:
"""Check if the caller is an admin. Result is cached on request.state."""
cached = getattr(request.state, "_is_admin", None)
if cached is not None:
return cached
user_id: str = request.state.user_id
# Local dev — no auth, treat as admin
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)
result = bool(user and user.is_admin)
request.state._is_admin = result
return result
# Fetch user from Clerk Backend API and check public_metadata.role
try:
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.clerk.com/v1/users/{user_id}",
headers={"Authorization": f"Bearer {settings.clerk_secret_key}"},
)
if resp.status_code == 200:
data = resp.json()
role = data.get("public_metadata", {}).get("role")
result = role == "admin"
else:
result = False
except Exception:
result = False
request.state._is_admin = result
return result
async def _require_admin(request: Request) -> str:
"""Return user_id if the caller is an admin, else raise 403."""
if not await is_admin(request):
raise HTTPException(403, "Admin access required")
return request.state.user_id
# ---------------------------------------------------------------------------
# Library components
# ---------------------------------------------------------------------------
@router.get("/components")
async def list_components(request: Request):
"""List all extracted IC components and passive patterns in the library."""
await _require_admin(request)
storage = get_storage(request)
catalog = proj_svc.list_library_catalog(storage)
return JSONResponse(
content={
"ics": catalog["ics"],
"passives": catalog["passives"],
"simple": catalog["simple"],
},
headers={"Cache-Control": "no-store"},
)
def _safe_name(name: str) -> str:
"""Sanitize MPN to safe filename (same logic as pipeline)."""
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
@router.get("/components/{component_type}/{name:path}")
async def get_component(component_type: str, name: str, request: Request):
"""Return the raw JSON for an IC extraction or passive pattern."""
await _require_admin(request)
safe = _safe_name(name)
storage = get_storage(request)
if component_type == "ic":
key = f"library/extracted/{safe}.json"
elif component_type == "passive":
key = f"library/patterns/{safe}.json"
elif component_type == "simple":
# Check library/passives/ first, then library/models/
key = f"library/passives/{safe}.json"
if not storage.exists(key):
key = f"library/models/{safe}.json"
else:
raise HTTPException(400, f"Unknown component type: {component_type}")
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):
"""Delete an IC extraction or passive pattern from the shared library."""
await _require_admin(request)
safe = _safe_name(name)
storage = get_storage(request)
if component_type == "ic":
key = f"library/extracted/{safe}.json"
elif component_type == "passive":
key = f"library/patterns/{safe}.json"
elif component_type == "simple":
# Check library/passives/ first, then library/models/
key = f"library/passives/{safe}.json"
if not storage.exists(key):
key = f"library/models/{safe}.json"
else:
raise HTTPException(400, f"Unknown component type: {component_type}")
if not storage.exists(key):
raise HTTPException(404, f"Component not found: {name}")
storage.delete_key(key)
# Delete datasheet ref (blob preserved for other refs; GC cleans orphans)
from backend.services.datasheet_store import delete_datasheet_ref
deleted_datasheets = 0
old_blob = delete_datasheet_ref(storage, name)
if old_blob:
deleted_datasheets += 1
# Legacy flat file cleanup (remove after migration confirmed)
ds_key = f"library/datasheets/{safe}.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:
"""Pull display name / email / avatar out of a Clerk user object."""
first = clerk.get("first_name") or ""
last = clerk.get("last_name") or ""
emails = clerk.get("email_addresses", [])
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:
"""Build the project-count + balance record for a single user_id."""
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,
}
async def _enrich_clerk_profiles(users: dict[str, dict]) -> None:
"""Fill name/email/avatar for each user via the Clerk Backend API.
Fetches in parallel (bounded) so the list stays fast even with many
users. Failures per-user are swallowed — the row still renders with
the user_id as a fallback label.
"""
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"https://api.clerk.com/v1/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:
pass
await asyncio.gather(*(_one(uid) for uid in users))
@router.get("/users")
async def list_users(request: Request):
"""List every user with a project or any credit activity.
The balance file is written on a user's first ``GET /api/credits``
(trial grant), so this includes everyone who has ever opened the
authenticated app — not only project creators. To find a user who has
never opened the app, use ``GET /api/admin/users/search?email=``.
"""
await _require_admin(request)
storage = get_storage(request)
user_ids: set[str] = set()
# Project creators (users/{user_id}/...)
for entry in storage.list_prefix("users/"):
parts = entry.split("/")
if len(parts) >= 2 and parts[1]:
user_ids.add(parts[1])
# Anyone with credit activity (covers the trial grant on first app open)
user_ids.update(get_billing().list_user_ids(storage))
users: dict[str, dict] = {uid: _base_admin_user(storage, uid) for uid in user_ids}
# Enrich with Clerk user info when auth is enabled
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):
"""Find users by email via Clerk so any account can be topped up (admin).
Resolves even users with no project and no credit activity yet — useful
for granting credits to someone who has just signed up. Requires auth
to be enabled (no Clerk directory exists in local dev).
"""
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(
"https://api.clerk.com/v1/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
# ---------------------------------------------------------------------------
# Usage / cost tracking
# ---------------------------------------------------------------------------
@router.get("/usage")
async def get_usage(request: Request):
"""Aggregate API token usage and cost across all users and projects."""
await _require_admin(request)
storage = get_storage(request)
user_entries = storage.list_prefix("users/")
seen_uids: set[str] = set()
user_rows: list[dict] = []
grand_total = 0.0
for entry in user_entries:
parts = entry.split("/")
if len(parts) >= 2:
uid = parts[1]
if uid in seen_uids:
continue
seen_uids.add(uid)
projects = proj_svc.list_projects(storage, uid)
user_cost = 0.0
project_details = []
for p in projects:
cost = p.total_cost_usd or 0.0
user_cost += cost
project_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": project_details,
"name": None,
"email": None,
})
grand_total += user_cost
# Enrich with Clerk user info
if settings.use_auth and user_rows:
async with httpx.AsyncClient() as client:
for row in user_rows:
try:
resp = await client.get(
f"https://api.clerk.com/v1/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", [])
row["email"] = emails[0].get("email_address") if emails else None
except Exception:
pass
return {
"grand_total_usd": round(grand_total, 4),
"users": user_rows,
}
# ---------------------------------------------------------------------------
# All projects (cross-user)
# ---------------------------------------------------------------------------
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:
"""Enrich a list of dicts with Clerk user info, deduplicating API calls."""
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"https://api.clerk.com/v1/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", [])
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):
"""List all projects across all users with metadata."""
await _require_admin(request)
storage = get_storage(request)
user_entries = storage.list_prefix("users/")
seen_uids: set[str] = set()
all_projects: list[dict] = []
for entry in user_entries:
parts = entry.split("/")
if len(parts) >= 2:
uid = parts[1]
if uid in seen_uids:
continue
seen_uids.add(uid)
projects = proj_svc.list_projects(storage, uid)
for p in projects:
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
# ---------------------------------------------------------------------------
# Running pipelines
# ---------------------------------------------------------------------------
@router.get("/runs")
async def list_running_pipelines(request: Request):
"""List queued and running pipelines, plus drive the stale-running sweeper.
Source of truth is ``project.json`` (``status`` ∈ {queued, running}); we
cross-check with the Cloud Run Job execution. Any project whose
execution is in a terminal Cloud Run state but whose status is still
queued/running is flipped to ``error`` here — this is the sweeper that
keeps zombie projects from showing "running" forever in the UI.
"""
from datetime import datetime, timezone
from backend.services import job_runner
await _require_admin(request)
storage = get_storage(request)
now = datetime.now(timezone.utc)
runs: list[dict] = []
seen_uids: set[str] = set()
for entry in storage.list_prefix("users/"):
parts = entry.split("/")
if len(parts) < 2:
continue
uid = parts[1]
if uid in seen_uids:
continue
seen_uids.add(uid)
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
# Sweeper: if the execution is in a terminal Cloud Run / local
# state, the worker is already gone. Flip status → error so the
# UI stops lying. Also heal projects whose event log already
# ends with pipeline_complete (finished, meta never flipped).
healed = proj_svc.heal_if_pipeline_finished(storage, uid, meta.id)
if healed 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():
# Local zombie: no execution_name but a dead pid file, or
# no live proc — treat as failed after the stale window.
exec_state = job_runner.get_execution_state(
f"local/projects/{meta.id}"
)
if exec_state in ("succeeded", "failed", "cancelled"):
# Allow a short grace period so we don't race the worker
# writing its own terminal status. updated may be stale
# if the worker died before any status write.
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
# ---------------------------------------------------------------------------
# Global settings
# ---------------------------------------------------------------------------
class UpdateMinVersionRequest(BaseModel):
min_model_version: str
@router.get("/settings")
async def get_settings(request: Request):
"""Get global admin settings (model version threshold, etc.)."""
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):
"""Set the minimum model version for library reuse."""
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}
# ---------------------------------------------------------------------------
# Email test
# ---------------------------------------------------------------------------
class TestEmailRequest(BaseModel):
to_email: str
@router.post("/test-email")
async def test_email(req: TestEmailRequest, request: Request):
"""Send a test email to verify Gmail API setup. Admin only."""
await _require_admin(request)
from backend.services.email import send_test_email
result = await send_test_email(req.to_email)
return result
# ---------------------------------------------------------------------------
# Project state overrides
# ---------------------------------------------------------------------------
@router.post("/projects/{project_id}/mark-complete")
async def mark_project_complete(project_id: str, request: Request):
"""Admin-only: force a paused project to ``complete`` status.
Intended for projects stuck at ``paused_insufficient_credits`` that the
admin has decided to finalize rather than resume. Clears the pause
checkpoint/reason; does not touch credits, cost totals, or artifacts.
"""
from backend.routers.deps import resolve_or_404
from backend.services import projects as proj_svc
await _require_admin(request)
storage = get_storage(request)
owner_user_id, 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_user_id,
project_id,
status="complete",
pause_checkpoint=None,
pause_reason=None,
)
return {"status": "complete", "project_id": project_id}
# ---------------------------------------------------------------------------
# Report overrides
# ---------------------------------------------------------------------------
@router.delete("/projects/{project_id}/findings/{finding_id}")
async def delete_finding(project_id: str, finding_id: str, request: Request):
"""Admin-only: delete a single finding (rule violation) from a report.
Rewrites ``report.json`` without the matching finding, recomputes summary
counts, and mirrors the summary onto ``ProjectMeta`` so dashboard totals
stay consistent. Returns 404 if the project, report, or finding is missing.
"""
from backend.routers.deps import resolve_or_404
await _require_admin(request)
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
key = f"{proj_svc.project_prefix(owner_user_id, 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_user_id, project_id, summary=summary)
return {
"deleted": finding_id,
"project_id": project_id,
"remaining": len(remaining),
"summary": summary,
}
+138
View File
@@ -0,0 +1,138 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Public contact form endpoint — no authentication required."""
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()
# Simple in-memory rate limiting (per-instance, resets on deploy)
_recent: dict[str, float] = {}
_RATE_LIMIT_SECONDS = 60
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 _build_contact_message(data: ContactRequest) -> MIMEMultipart:
"""Build the contact form email."""
msg = MIMEMultipart("alternative")
msg["From"] = f"Periscope <{settings.email_sender}>"
msg["To"] = settings.contact_recipient
msg["Reply-To"] = data.email
msg["Subject"] = f"[Periscope Contact] {data.subject or 'New message'} from {data.name}"
# Plain text
lines = [
f"Name: {data.name}",
f"Email: {data.email}",
]
if data.company:
lines.append(f"Company: {data.company}")
if data.subject:
lines.append(f"Subject: {data.subject}")
lines += ["", data.message, "", "— Sent from the Periscope contact form"]
msg.attach(MIMEText("\n".join(lines), "plain"))
# HTML
name = html.escape(data.name)
email = html.escape(data.email)
company = html.escape(data.company)
subject = html.escape(data.subject)
message = html.escape(data.message)
rows = f"""\
<tr>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5; font-weight: 600; width: 100px;">Name</td>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5;">{name}</td>
</tr>
<tr>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5; font-weight: 600;">Email</td>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5;"><a href="mailto:{email}">{email}</a></td>
</tr>"""
if data.company:
rows += f"""\
<tr>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5; font-weight: 600;">Company</td>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5;">{company}</td>
</tr>"""
if data.subject:
rows += f"""\
<tr>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5; font-weight: 600;">Subject</td>
<td style="padding: 8px 12px; border: 1px solid #e5e5e5;">{subject}</td>
</tr>"""
html_body = f"""\
<div style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; max-width: 560px; margin: 0 auto; padding: 24px;">
<h2 style="font-size: 18px; margin: 0 0 16px;">New contact form submission</h2>
<table style="width: 100%; border-collapse: collapse; font-size: 14px;">
{rows}
</table>
<div style="margin-top: 16px; padding: 16px; background: #f9fafb; border-radius: 8px; font-size: 14px; line-height: 1.6; white-space: pre-wrap;">{message}</div>
<p style="margin-top: 24px; font-size: 12px; color: #888;">Sent from the Periscope contact form</p>
</div>"""
msg.attach(MIMEText(html_body, "html"))
return msg
@router.post("/contact", response_model=ContactResponse)
async def submit_contact(data: ContactRequest, request: Request):
# Honeypot check — bots fill hidden fields
if data.honeypot:
return ContactResponse(success=True, message="Message sent! We'll get back to you soon.")
# Rate limiting by IP
ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() or request.client.host
now = time.time()
last = _recent.get(ip)
if last and now - last < _RATE_LIMIT_SECONDS:
return ContactResponse(success=False, message="Please wait a minute before submitting again.")
_recent[ip] = now
# Clean up old entries
if len(_recent) > 1000:
cutoff = now - _RATE_LIMIT_SECONDS
for key in [k for k, v in _recent.items() if v < cutoff]:
del _recent[key]
# Check email is configured
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.",
)
msg = _build_contact_message(data)
await _send_raw(settings.contact_recipient, msg, "Contact form")
return ContactResponse(success=True, message="Message sent! We'll get back to you soon.")
+40
View File
@@ -0,0 +1,40 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Shared dependencies for FastAPI 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]:
"""Resolve project access (owner, collaborator, or admin) or raise 404."""
storage = get_storage(request)
user_id = get_user_id(request)
# 1. Try normal access — cheap, no external API call
result = proj_svc.resolve_project_access(storage, user_id, project_id)
if result:
return result
# 2. Admin fallback — Clerk API call only when normal access fails
from backend.routers.admin import is_admin
if await is_admin(request):
result = proj_svc.find_project_any_user(storage, project_id)
if result:
return result
raise HTTPException(404, "Project not found")
+306
View File
@@ -0,0 +1,306 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""User feedback / ticket system.
Users can submit feedback tickets (bugs, rule reports, feature requests).
Tickets are stored as individual JSON files with JSONL indexes for fast listing.
Storage layout:
admin/feedback/tickets/{ticket_id}.json
admin/feedback/index/by_user/{user_id}.jsonl
admin/feedback/index/by_project/{project_id}.jsonl
admin/feedback/index/all.jsonl
"""
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()
# ---------------------------------------------------------------------------
# Storage key helpers
# ---------------------------------------------------------------------------
_TICKETS_PREFIX = "admin/feedback/tickets/"
_INDEX_BY_USER = "admin/feedback/index/by_user/"
_INDEX_BY_PROJECT = "admin/feedback/index/by_project/"
_INDEX_ALL = "admin/feedback/index/all.jsonl"
def _ticket_key(ticket_id: str) -> str:
return f"{_TICKETS_PREFIX}{ticket_id}.json"
def _user_index_key(user_id: str) -> str:
return f"{_INDEX_BY_USER}{user_id}.jsonl"
def _project_index_key(project_id: str) -> str:
return f"{_INDEX_BY_PROJECT}{project_id}.jsonl"
# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------
FeedbackType = Literal["bug", "rule_feedback", "feature_request"]
FeedbackStatus = Literal["open", "acknowledged", "resolved"]
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
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _append_index(storage: StorageBackend, key: str, entry: dict) -> None:
existing = ""
if storage.exists(key):
existing = storage.read_text(key)
line = json.dumps(entry) + "\n"
storage.write_text(key, existing + line)
def _read_index(storage: StorageBackend, key: str) -> list[dict]:
if not storage.exists(key):
return []
text = storage.read_text(key)
entries: list[dict] = []
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
entries.append(json.loads(line))
except json.JSONDecodeError:
continue
return entries
def _read_ticket(storage: StorageBackend, ticket_id: str) -> FeedbackTicket | None:
key = _ticket_key(ticket_id)
if not storage.exists(key):
return None
try:
data = storage.read_json(key)
return FeedbackTicket(**data)
except Exception:
logger.warning("Failed to read ticket %s", ticket_id)
return None
def _read_tickets_from_index(
storage: StorageBackend,
index_key: str,
*,
status: str | None = None,
ticket_type: str | None = None,
project_id: str | None = None,
) -> list[FeedbackTicket]:
index_entries = _read_index(storage, index_key)
tickets: list[FeedbackTicket] = []
for entry in reversed(index_entries):
tid = entry.get("ticket_id")
if not tid:
continue
ticket = _read_ticket(storage, tid)
if not ticket:
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
tickets.append(ticket)
return tickets
# ---------------------------------------------------------------------------
# User endpoints
# ---------------------------------------------------------------------------
@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_entry = {"ticket_id": ticket_id, "created_at": now}
_append_index(storage, _user_index_key(user_id), index_entry)
_append_index(storage, _INDEX_ALL, index_entry)
if body.project_id:
_append_index(storage, _project_index_key(body.project_id), index_entry)
# Notify the admin inbox (fire-and-forget, identical pattern to
# pipeline-started). Errors are swallowed inside the email service.
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):
storage = get_storage(request)
user_id = get_user_id(request)
return _read_tickets_from_index(
storage, _user_index_key(user_id), status=status,
)
# ---------------------------------------------------------------------------
# Admin endpoints
# ---------------------------------------------------------------------------
@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)
storage = get_storage(request)
return _read_tickets_from_index(
storage, _INDEX_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 not ticket:
raise HTTPException(404, "Ticket not found")
prev_admin_notes = (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())
# If admin_notes changed to a new, non-empty value, notify the submitter.
new_admin_notes = (ticket.admin_notes or "").strip()
if new_admin_notes and new_admin_notes != prev_admin_notes:
try:
from backend.services.email import send_feedback_reply_email
await send_feedback_reply_email(
user_id=ticket.user_id,
reply_text=new_admin_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
+980
View File
@@ -0,0 +1,980 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Pipeline start, SSE events, and status endpoints.
Pipelines run in a Cloud Run Job worker (or, in local dev, a child
subprocess). The API only enqueues, transitions status with
``if-generation-match`` for idempotency, and tails the GCS-backed event
log for SSE.
"""
from __future__ import annotations
import asyncio
import json
import logging
from fastapi import APIRouter, HTTPException, Request
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
from typing import Literal
from backend.routers.deps import get_storage, resolve_or_404
from backend.services import event_bridge, job_runner
from backend.services import projects as proj_svc
logger = logging.getLogger(__name__)
VALID_REGEN_STAGES = {"derating"}
_REPROCESS_OK_FROM = frozenset({
proj_svc.STATUS_COMPLETE,
proj_svc.STATUS_ERROR,
proj_svc.STATUS_CANCELLED,
})
class RegenRequest(BaseModel):
stages: list[str]
class ReprocessRequest(BaseModel):
"""``failed`` retries skipped/errored reviews and ICs whose circuit
neighborhood changed; ``all`` re-reviews every IC."""
mode: Literal["failed", "all"] = "failed"
router = APIRouter(tags=["pipeline"])
# Statuses from which a fresh ``/start`` is allowed to transition into queued.
_START_OK_FROM = frozenset({
proj_svc.STATUS_DRAFT,
proj_svc.STATUS_COMPLETE,
proj_svc.STATUS_ERROR,
proj_svc.STATUS_CANCELLED,
})
def _project_active(meta: proj_svc.ProjectMeta) -> bool:
"""A project is "active" if a worker is or could be running for it.
Used as the running-guard. We trust the meta status as the primary
signal, and only fall back to the Cloud Run execution state when the
status is one we expect a worker to be touching. This deliberately
does NOT call get_execution_state on every request — it's an admin
API call. The stale-running sweeper is responsible for clearing
zombie ``running`` projects.
"""
return meta.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING)
@router.post("/pipeline/{project_id}/start", status_code=202)
async def start(project_id: str, request: Request):
from backend.routers.deps import get_user_id
from backend.services.billing_hook import get_billing
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Upload BOM and netlist before starting pipeline")
# Ensure the caller has at least their trial credits allocated. The
# pipeline itself enforces pause-on-empty — this just makes sure a
# brand-new user isn't blocked before their grant is issued.
get_billing().ensure_trial_grant(storage, get_user_id(request))
# Idempotent enqueue: only one ``draft|complete|error|cancelled`` ->
# ``queued`` transition can win. Concurrent /start clicks => 409.
from backend._version import PERISCOPE_VERSION
try:
proj_svc.transition_status(
storage, owner_user_id, project_id,
from_status=_START_OK_FROM,
to_status=proj_svc.STATUS_QUEUED,
cancel_requested=False,
execution_name=None,
periscope_version=PERISCOPE_VERSION,
)
except proj_svc.StatusConflict:
raise HTTPException(409, "Pipeline already running or queued")
try:
execution_name = job_runner.enqueue_pipeline(
project_id, owner_user_id, resume=False, free=False,
)
except Exception:
logger.exception("enqueue_pipeline failed for %s", project_id)
# Roll the meta back so the user can retry.
proj_svc.update_project(
storage, owner_user_id, project_id,
status=proj_svc.STATUS_ERROR,
pipeline_state={"error": "Failed to enqueue worker"},
)
raise HTTPException(503, "Failed to enqueue pipeline worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id, execution_name=execution_name,
)
return {"status": "started", "project_id": project_id}
@router.post("/pipeline/{project_id}/cancel")
async def cancel(project_id: str, request: Request):
"""Soft-cancel: set ``cancel_requested`` so the worker exits cleanly.
The worker re-reads this flag inside ``_charge_for_logs`` after every
Claude API call (throttled). Cancellation latency is bounded by the
in-flight call's duration, typ 160s.
"""
storage = get_storage(request)
owner_user_id, 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_user_id, project_id)
return {"status": "cancel_requested", "project_id": project_id}
@router.post("/pipeline/{project_id}/estimate")
async def estimate(project_id: str, request: Request):
"""Pre-flight cost estimate — read-only, no side effects."""
from backend.services.cost_estimator import estimate_pipeline_cost
storage = get_storage(request)
owner_user_id, 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_user_id, 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):
"""Resume a pipeline that was paused for insufficient credits."""
storage = get_storage(request)
owner_user_id, 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.",
)
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Project is missing BOM or netlist")
try:
proj_svc.transition_status(
storage, owner_user_id, project_id,
from_status=proj_svc.STATUS_PAUSED,
to_status=proj_svc.STATUS_QUEUED,
cancel_requested=False,
)
except proj_svc.StatusConflict:
raise HTTPException(409, "Project state changed; refresh and retry")
try:
execution_name = job_runner.enqueue_pipeline(
project_id, owner_user_id, resume=True, free=False,
)
except Exception:
logger.exception("enqueue_pipeline (resume) failed for %s", project_id)
proj_svc.update_project(
storage, owner_user_id, project_id,
status=proj_svc.STATUS_ERROR,
pipeline_state={"error": "Failed to enqueue worker"},
)
raise HTTPException(503, "Failed to enqueue pipeline worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id, execution_name=execution_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):
"""Re-run a finished project without the create wizard.
Keeps BOM, netlist, datasheets, and library extractions. ``failed``
(default) skips ICs that already produced a review; ``all`` re-reviews
every IC.
"""
from backend._version import PERISCOPE_VERSION
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Upload BOM and netlist before reprocessing")
if _project_active(meta):
await _interrupt_active_pipeline(storage, owner_user_id, project_id)
meta = proj_svc.get_project(storage, owner_user_id, project_id) or meta
if meta.status == proj_svc.STATUS_PAUSED:
allowed = _REPROCESS_OK_FROM | {proj_svc.STATUS_PAUSED}
else:
allowed = _REPROCESS_OK_FROM
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_user_id, project_id)
if retry_failed else []
)
try:
proj_svc.transition_status(
storage, owner_user_id, project_id,
from_status=allowed | {
proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING,
},
to_status=proj_svc.STATUS_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,
)
except proj_svc.StatusConflict:
raise HTTPException(409, "Pipeline already running or queued")
try:
execution_name = job_runner.enqueue_pipeline(
project_id, owner_user_id, resume=retry_failed, free=False,
)
except Exception:
logger.exception("enqueue_pipeline (reprocess) failed for %s", project_id)
proj_svc.update_project(
storage, owner_user_id, project_id,
status=proj_svc.STATUS_ERROR,
pipeline_state={"error": "Failed to enqueue worker"},
)
raise HTTPException(503, "Failed to enqueue pipeline worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id, execution_name=execution_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):
"""Admin-only: wipe per-project extractions and run the pipeline free."""
from backend.routers.admin import _require_admin
await _require_admin(request)
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Upload BOM and netlist before starting pipeline")
# If something is currently running/queued, request cancel and wait
# briefly for the worker to honour it (or exit on its own). Hard-kill
# the execution as a last resort.
if _project_active(meta):
proj_svc.request_cancel(storage, owner_user_id, project_id)
await _await_terminal(storage, owner_user_id, project_id, timeout_s=10.0)
# If still active, hard-kill via Cloud Run cancel.
meta = proj_svc.get_project(storage, owner_user_id, project_id) or meta
if _project_active(meta) and meta.execution_name:
job_runner.cancel_execution(meta.execution_name)
await _await_terminal(storage, owner_user_id, project_id, timeout_s=5.0)
proj_svc.clear_project_extractions(storage, owner_user_id, project_id)
# After clear_project_extractions the project is left in whatever
# status it was; the transition below enforces queued.
try:
proj_svc.transition_status(
storage, owner_user_id, project_id,
from_status=_START_OK_FROM | {proj_svc.STATUS_PAUSED},
to_status=proj_svc.STATUS_QUEUED,
cancel_requested=False,
execution_name=None,
)
except proj_svc.StatusConflict:
raise HTTPException(409, "Pipeline is busy; cancel first then retry")
try:
execution_name = job_runner.enqueue_pipeline(
project_id, owner_user_id, resume=False, free=True,
)
except Exception:
logger.exception("enqueue_pipeline (restart) failed for %s", project_id)
proj_svc.update_project(
storage, owner_user_id, project_id,
status=proj_svc.STATUS_ERROR,
pipeline_state={"error": "Failed to enqueue worker"},
)
raise HTTPException(503, "Failed to enqueue pipeline worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id, execution_name=execution_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):
"""Rebuild graph and regenerate only the requested stages."""
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Upload BOM and netlist before running regen")
invalid = set(req.stages) - VALID_REGEN_STAGES
if invalid:
raise HTTPException(400, f"Invalid regen stages: {sorted(invalid)}. Valid: {sorted(VALID_REGEN_STAGES)}")
if not req.stages:
raise HTTPException(400, "At least one stage is required")
if _project_active(meta):
proj_svc.request_cancel(storage, owner_user_id, project_id)
await _await_terminal(storage, owner_user_id, project_id, timeout_s=10.0)
meta = proj_svc.get_project(storage, owner_user_id, project_id) or meta
if _project_active(meta) and meta.execution_name:
job_runner.cancel_execution(meta.execution_name)
await _await_terminal(storage, owner_user_id, project_id, timeout_s=5.0)
try:
proj_svc.transition_status(
storage, owner_user_id, project_id,
from_status=_START_OK_FROM | {proj_svc.STATUS_PAUSED},
to_status=proj_svc.STATUS_QUEUED,
cancel_requested=False,
execution_name=None,
)
except proj_svc.StatusConflict:
raise HTTPException(409, "Pipeline is busy; cancel first then retry")
try:
execution_name = job_runner.enqueue_pipeline_regen(
project_id, owner_user_id, stages=req.stages,
)
except Exception:
logger.exception("enqueue_pipeline_regen failed for %s", project_id)
proj_svc.update_project(
storage, owner_user_id, project_id,
status=proj_svc.STATUS_ERROR,
pipeline_state={"error": "Failed to enqueue worker"},
)
raise HTTPException(503, "Failed to enqueue pipeline worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id, execution_name=execution_name,
)
return {"status": "regen_started", "project_id": project_id, "stages": req.stages}
# ---------------------------------------------------------------------------
# SSE events
# ---------------------------------------------------------------------------
_EXEC_TERMINAL = frozenset({"succeeded", "failed", "cancelled"})
_ANALYSIS_SSE_TERMINAL = frozenset({
"pipeline_complete",
"pipeline_error",
"pipeline_cancelled",
"pipeline_paused",
})
@router.get("/pipeline/{project_id}/events")
async def events(project_id: str, request: Request):
"""SSE stream of pipeline progress events.
Tails the GCS-backed event log written by the worker. Stops on
analysis terminal events, but also has two hard-crash escape
hatches: the project's status reaching a terminal value *after*
having been active, and the Cloud Run execution reaching a
terminal state. Either of those triggers a synthetic
``pipeline_error`` so the SSE doesn't hang forever when the worker
dies without writing its terminal event.
"""
owner_user_id, meta = await resolve_or_404(request, project_id)
storage = get_storage(request)
async def event_generator():
execution_name = meta.execution_name
crash_detected: dict[str, str | None] = {"reason": None}
# Only treat a terminal status as a crash if we observed the
# project as queued/running first — otherwise a finished project
# reconnecting to /events would immediately synthesize an error
# (or race with a historical pipeline_complete replay).
active_state = {
"saw": meta.status in (
proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING,
),
}
async def watch_status() -> None:
poll_interval = 2.0
while True:
await asyncio.sleep(poll_interval)
try:
cur = proj_svc.get_project(storage, owner_user_id, project_id)
except Exception:
continue
if cur is None:
continue
if cur.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING):
active_state["saw"] = True
elif active_state["saw"] and cur.status in proj_svc.TERMINAL_STATUSES:
crash_detected["reason"] = (
f"project status={cur.status} (terminal)"
)
return
# Cloud Run hard-crash detection
if execution_name and (
active_state["saw"]
or cur.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING)
):
try:
state = job_runner.get_execution_state(execution_name)
except Exception:
state = "unknown"
if state in _EXEC_TERMINAL:
crash_detected["reason"] = (
f"execution state={state}"
)
return
watcher = asyncio.create_task(watch_status())
try:
async for msg in event_bridge.tail_events(
storage, owner_user_id, project_id,
terminal_events=_ANALYSIS_SSE_TERMINAL,
):
if crash_detected["reason"] is not None:
break
ev = msg["event"]
# Skip placement events in the shared log.
if ev.startswith("placement_") or ev.startswith("pcb_"):
continue
yield {
"event": ev,
"data": json.dumps(msg.get("data", {})),
}
if ev in _ANALYSIS_SSE_TERMINAL:
return
# tail_events exited without a terminal event — escape hatch
if crash_detected["reason"] is not None:
# Re-read the current meta so the synthetic event has
# the most up-to-date error information.
cur = proj_svc.get_project(storage, owner_user_id, project_id)
err = (
(cur.pipeline_state or {}).get("error")
if cur and cur.pipeline_state
else crash_detected["reason"]
)
yield {
"event": "pipeline_error",
"data": json.dumps({
"error": err or "worker terminated without writing a terminal event",
"synthetic": True,
}),
}
finally:
watcher.cancel()
try:
await watcher
except (asyncio.CancelledError, Exception):
pass
return EventSourceResponse(
event_generator(),
ping=15,
headers={
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
},
)
@router.get("/pipeline/{project_id}/status")
async def status(project_id: str, request: Request):
"""Polling fallback — returns current project state.
Also heals zombie ``running``/``queued`` projects whose event log
already ends with ``pipeline_complete`` (worker died after finishing).
"""
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
healed = proj_svc.heal_if_pipeline_finished(storage, owner_user_id, project_id)
if healed is not None:
meta = healed
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": (meta.placement_status or "draft") in ("queued", "running"),
"pcb_status": meta.pcb_status,
"pcb_state": meta.pcb_state,
"pcb_running": (meta.pcb_status or "draft") in ("queued", "running"),
"healed": healed is not None,
}
# ---------------------------------------------------------------------------
# Placement pipeline (parallel — topology only, no LLM / no credits)
# ---------------------------------------------------------------------------
_PLACEMENT_START_OK = frozenset({"draft", "complete", "error", "cancelled"})
_PLACEMENT_SSE_TERMINAL = frozenset({
"placement_complete",
"placement_error",
"placement_cancelled",
})
@router.post("/pipeline/{project_id}/placement/start", status_code=202)
async def start_placement(project_id: str, request: Request):
"""Enqueue the Placement topology pipeline (free, no analysis status change)."""
from backend.services.placement_pipeline import analysis_busy, placement_busy
from backend.services.pcb_pipeline import pcb_busy
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Upload BOM and netlist before starting placement")
if 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_START_OK:
raise HTTPException(
409,
f"Cannot start placement from placement_status={meta.placement_status}",
)
proj_svc.update_project(
storage, owner_user_id, project_id,
placement_status="queued",
placement_cancel_requested=False,
placement_state=None,
placement_execution_name=None,
)
# Clear before enqueue so the placement SSE client never stops on a
# leftover analysis ``pipeline_complete`` in the shared event log.
try:
event_bridge.GCSEventBroker(storage, owner_user_id).clear_history(project_id)
except Exception:
logger.exception("failed to clear events before placement start for %s", project_id)
try:
execution_name = job_runner.enqueue_placement_pipeline(
project_id, owner_user_id,
)
except Exception:
logger.exception("enqueue_placement_pipeline failed for %s", project_id)
proj_svc.update_project(
storage, owner_user_id, project_id,
placement_status="error",
placement_state={"error": "Failed to enqueue placement worker"},
)
raise HTTPException(503, "Failed to enqueue placement worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id,
placement_execution_name=execution_name,
)
return {"status": "started", "project_id": project_id}
@router.post("/pipeline/{project_id}/placement/cancel")
async def cancel_placement(project_id: str, request: Request):
"""Soft-cancel the Placement pipeline via ``placement_cancel_requested``."""
from backend.services.placement_pipeline import placement_busy
storage = get_storage(request)
owner_user_id, 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_user_id, 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):
"""Return ``placement_plan.json`` (F1 topology — no coordinates)."""
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
key = f"{proj_svc.project_prefix(owner_user_id, project_id)}/placement_plan.json"
if not storage.exists(key):
# Fallback for plans written only as functional_groups during analysis.
key = f"{proj_svc.project_prefix(owner_user_id, project_id)}/functional_groups.json"
if not storage.exists(key):
raise HTTPException(404, "Placement plan not found — run placement first")
return storage.read_json(key)
@router.get("/pipeline/{project_id}/placement/pack")
async def get_placement_pack(project_id: str, request: Request):
"""Return ``placement_pack.json`` (F2 — skipped without PCB + numeric rules)."""
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
key = f"{proj_svc.project_prefix(owner_user_id, 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):
"""SSE stream for Placement pipeline progress (watches placement_* only)."""
owner_user_id, meta = await resolve_or_404(request, project_id)
storage = get_storage(request)
async def event_generator():
execution_name = meta.placement_execution_name
crash_detected: dict[str, str | None] = {"reason": None}
async def watch_status() -> None:
poll_interval = 2.0
saw_active = (meta.placement_status or "draft") in ("queued", "running")
while True:
await asyncio.sleep(poll_interval)
try:
cur = proj_svc.get_project(storage, owner_user_id, project_id)
except Exception:
continue
if cur is None:
continue
pst = cur.placement_status or "draft"
if pst in ("queued", "running"):
saw_active = True
elif saw_active and pst in ("complete", "error", "cancelled"):
# Worker wrote terminal status; if SSE missed the event,
# surface a synthetic terminal after a short grace.
crash_detected["reason"] = f"placement_status={pst} (terminal)"
return
if execution_name:
try:
state = job_runner.get_execution_state(execution_name)
except Exception:
state = "unknown"
if state in _EXEC_TERMINAL and (
saw_active or pst in ("queued", "running")
):
crash_detected["reason"] = f"execution state={state}"
return
watcher = asyncio.create_task(watch_status())
try:
async for msg in event_bridge.tail_events(
storage, owner_user_id, project_id,
terminal_events=_PLACEMENT_SSE_TERMINAL,
):
if crash_detected["reason"] is not None:
break
ev = msg["event"]
# Skip leftover analysis events if the log was not cleared yet.
if not (
ev.startswith("placement_")
or ev == "heartbeat"
) or ev.startswith("pcb_"):
continue
yield {
"event": ev,
"data": json.dumps(msg.get("data", {})),
}
if ev in _PLACEMENT_SSE_TERMINAL:
return
if crash_detected["reason"] is not None:
cur = proj_svc.get_project(storage, owner_user_id, project_id)
err = None
if cur and cur.placement_state:
err = cur.placement_state.get("error")
yield {
"event": "placement_error",
"data": json.dumps({
"error": err or crash_detected["reason"]
or "placement worker terminated without a terminal event",
"synthetic": True,
}),
}
finally:
watcher.cancel()
try:
await watcher
except (asyncio.CancelledError, Exception):
pass
return EventSourceResponse(
event_generator(),
ping=15,
headers={
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
},
)
# ---------------------------------------------------------------------------
# PCB review pipeline (parallel — exam, not auto-place)
# ---------------------------------------------------------------------------
_PCB_START_OK = frozenset({"draft", "complete", "error", "cancelled"})
_PCB_SSE_TERMINAL = frozenset({
"pcb_complete",
"pcb_error",
"pcb_cancelled",
})
@router.post("/pipeline/{project_id}/pcb/start", status_code=202)
async def start_pcb(project_id: str, request: Request):
from backend.services.pcb_pipeline import analysis_busy, pcb_busy, placement_busy
storage = get_storage(request)
owner_user_id, meta = await resolve_or_404(request, project_id)
if not meta.has_pcb:
raise HTTPException(400, "Upload a .kicad_pcb before starting PCB review")
if not meta.has_bom or not meta.has_netlist:
raise HTTPException(400, "Upload BOM and netlist before starting PCB review")
if analysis_busy(meta):
raise HTTPException(409, "Analysis pipeline is running; wait or cancel it first")
if 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_START_OK:
raise HTTPException(
409,
f"Cannot start PCB review from pcb_status={meta.pcb_status}",
)
proj_svc.update_project(
storage, owner_user_id, project_id,
pcb_status="queued",
pcb_cancel_requested=False,
pcb_state=None,
pcb_execution_name=None,
)
try:
event_bridge.GCSEventBroker(storage, owner_user_id).clear_history(project_id)
except Exception:
logger.exception("failed to clear events before PCB start for %s", project_id)
try:
execution_name = job_runner.enqueue_pcb_pipeline(
project_id, owner_user_id,
)
except Exception:
logger.exception("enqueue_pcb_pipeline failed for %s", project_id)
proj_svc.update_project(
storage, owner_user_id, project_id,
pcb_status="error",
pcb_state={"error": "Failed to enqueue PCB worker"},
)
raise HTTPException(503, "Failed to enqueue PCB worker; please retry")
proj_svc.update_project(
storage, owner_user_id, project_id,
pcb_execution_name=execution_name,
)
return {"status": "started", "project_id": project_id}
@router.post("/pipeline/{project_id}/pcb/cancel")
async def cancel_pcb(project_id: str, request: Request):
from backend.services.pcb_pipeline import pcb_busy
storage = get_storage(request)
owner_user_id, 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_user_id, 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_user_id, _ = await resolve_or_404(request, project_id)
key = f"{proj_svc.project_prefix(owner_user_id, 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_user_id, meta = await resolve_or_404(request, project_id)
storage = get_storage(request)
async def event_generator():
execution_name = meta.pcb_execution_name
crash_detected: dict[str, str | None] = {"reason": None}
async def watch_status() -> None:
poll_interval = 2.0
saw_active = (meta.pcb_status or "draft") in ("queued", "running")
while True:
await asyncio.sleep(poll_interval)
try:
cur = proj_svc.get_project(storage, owner_user_id, project_id)
except Exception:
continue
if cur is None:
continue
pst = cur.pcb_status or "draft"
if pst in ("queued", "running"):
saw_active = True
elif saw_active and pst in ("complete", "error", "cancelled"):
crash_detected["reason"] = f"pcb_status={pst} (terminal)"
return
if execution_name:
try:
state = job_runner.get_execution_state(execution_name)
except Exception:
state = "unknown"
if state in _EXEC_TERMINAL and (
saw_active or pst in ("queued", "running")
):
crash_detected["reason"] = f"execution state={state}"
return
watcher = asyncio.create_task(watch_status())
try:
async for msg in event_bridge.tail_events(
storage, owner_user_id, project_id,
terminal_events=_PCB_SSE_TERMINAL,
):
if crash_detected["reason"] is not None:
break
ev = msg["event"]
if not (ev.startswith("pcb_") or ev == "heartbeat"):
continue
yield {
"event": ev,
"data": json.dumps(msg.get("data", {})),
}
if ev in _PCB_SSE_TERMINAL:
return
if crash_detected["reason"] is not None:
cur = proj_svc.get_project(storage, owner_user_id, project_id)
from backend.services.pcb_pipeline import pcb_sse_terminal_from_status
ev, payload = pcb_sse_terminal_from_status(
cur.pcb_status if cur else None,
cur.pcb_state if cur else None,
crash_detected["reason"],
)
yield {
"event": ev,
"data": json.dumps(payload),
}
finally:
watcher.cancel()
try:
await watcher
except (asyncio.CancelledError, Exception):
pass
return EventSourceResponse(
event_generator(),
ping=15,
headers={
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
},
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
async def _interrupt_active_pipeline(
storage, user_id: str, project_id: str,
) -> None:
"""Cancel a queued/running worker so a new run can be enqueued.
If the worker is already dead (docker rebuild, OOM) the status can
stay ``running``; force it to cancelled after a short wait.
"""
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 _await_terminal(
storage, user_id: str, project_id: str, *, timeout_s: float,
) -> None:
"""Poll project status until it reaches a terminal state or the timeout
elapses. Used by /restart and /regen between cancel and re-enqueue.
"""
poll = 0.5
elapsed = 0.0
while elapsed < timeout_s:
try:
meta = proj_svc.get_project(storage, user_id, project_id)
except Exception:
meta = None
if meta is None:
return
if meta.status in proj_svc.TERMINAL_STATUSES:
return
await asyncio.sleep(poll)
elapsed += poll
File diff suppressed because it is too large Load Diff
+418
View File
@@ -0,0 +1,418 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Report, graph, datasheet, and API log serving endpoints."""
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.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
router = APIRouter(tags=["reports"])
log = logging.getLogger(__name__)
# Allow alphanumeric, dash, underscore, dot, colon, forward-slash, plus, hash, space
_SAFE_MPN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_\-\.:/ +#,()]*$")
def _validate_mpn(mpn: str) -> None:
"""Reject MPN values that could cause path traversal."""
if not _SAFE_MPN.match(mpn) or ".." in mpn:
raise HTTPException(400, "Invalid MPN format")
@router.get("/report/{project_id}")
async def get_report(project_id: str, request: Request):
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
prefix = proj_svc.project_prefix(owner_user_id, 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
from backend.periscopex.pcb_checks import merge_schema_pcb_reports
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:
pass
merged["findings"] = [json.loads(f.model_dump_json()) for f in findings]
summary = {"ERROR": 0, "WARNING": 0, "INFO": 0, "total": len(findings)}
for f in findings:
if f.status in summary:
summary[f.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_user_id, _ = await resolve_or_404(request, project_id)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/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_user_id, _ = await resolve_or_404(request, project_id)
user_id = get_user_id(request)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/report.json"
if not storage.exists(key):
raise HTTPException(404, "Report not found")
report_data = 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(),
}
comments = report_data.setdefault("comments", {})
comments.setdefault(body.finding_id, []).append(comment)
storage.write_json(key, report_data)
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_user_id, meta = await resolve_or_404(request, project_id)
user_id = get_user_id(request)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/report.json"
if not storage.exists(key):
raise HTTPException(404, "Report not found")
report_data = storage.read_json(key)
comments = report_data.get("comments", {})
for finding_id, comment_list in comments.items():
for i, c in enumerate(comment_list):
if c["comment_id"] == comment_id:
if c["user_id"] != user_id and user_id != owner_user_id:
raise HTTPException(403, "Cannot delete another user's comment")
comment_list.pop(i)
if not comment_list:
del comments[finding_id]
storage.write_json(key, report_data)
return JSONResponse({"ok": True})
class ReviewBody(BaseModel):
state: str
reason: str = ""
user_name: str = ""
def _load_report(storage, owner_user_id: str, project_id: str) -> tuple[str, dict]:
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/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.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_user_id, _ = await resolve_or_404(request, project_id)
user_id = get_user_id(request)
key, report_data = _load_report(storage, owner_user_id, project_id)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
findings = _findings_from_report(report_data)
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_data = storage.read_json(pcb_key)
pcb_findings = _findings_from_report(pcb_data)
if finding_id in {f.finding_id for f in pcb_findings if f.finding_id}:
key, report_data, findings = pcb_key, pcb_data, 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_data.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_data["review_states"] = states
storage.write_json(key, report_data)
if body.state in {"wontfix", "false_positive"}:
found = next(
(f for f in _findings_from_report(report_data) 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:
prefix = proj_svc.project_prefix(owner_user_id, project_id)
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_user_id, _ = await resolve_or_404(request, project_id)
_, report_data = _load_report(storage, owner_user_id, project_id)
items = build_eco(_findings_from_report(report_data), report_data.get("review_states") or {})
return JSONResponse({"items": items})
@router.get("/report/{project_id}/eco.csv")
async def get_eco_csv(project_id: str, request: Request):
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
_, report_data = _load_report(storage, owner_user_id, project_id)
items = build_eco(_findings_from_report(report_data), report_data.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_user_id, _ = await resolve_or_404(request, project_id)
user_id = get_user_id(request)
key, report_data = _load_report(storage, owner_user_id, project_id)
release = sign_report(report_data, user_id=user_id)
report_data["release"] = release
storage.write_json(key, report_data)
return JSONResponse(release)
raise HTTPException(404, "Comment not found")
@router.get("/bom/{project_id}")
async def get_bom_summary(project_id: str, request: Request):
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/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_user_id, _ = await resolve_or_404(request, project_id)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/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_user_id, _ = await resolve_or_404(request, project_id)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/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):
"""Return API call logs for a project pipeline run."""
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
prefix = proj_svc.project_prefix(owner_user_id, project_id)
key = f"{prefix}/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_user_id: str, project_id: str, safe: str,
mpn: str | None = None,
) -> str | None:
"""Return the storage key for a datasheet PDF, or None."""
from backend.services.datasheet_store import resolve_datasheet
# 1. Project uploads
key = f"{proj_svc.project_prefix(owner_user_id, project_id)}/uploads/datasheets/{safe}.pdf"
if storage.exists(key):
return key
# 2. Content-addressed ref lookup
resolved = resolve_datasheet(storage, safe)
if resolved:
return resolved
# 3. Legacy flat file fallback (remove after migration confirmed)
key = f"library/datasheets/{safe}.pdf"
if storage.exists(key):
return key
# 4. Pattern-based fallback (passives with shared datasheets)
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):
"""Return a URL for accessing a datasheet PDF.
Returns a backend proxy URL that streams the PDF through Cloud Run.
This avoids GCS signed-URL issues (IAM signBlob scope problems) and
works identically for local and cloud storage.
"""
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
_validate_mpn(mpn)
safe = safe_mpn(mpn)
key = _find_datasheet_key(storage, owner_user_id, project_id, safe, mpn=mpn)
if key is None:
raise HTTPException(404, f"Datasheet not found for MPN: {mpn}")
# Return a proxy URL that points back to this backend
proxy_path = f"/api/projects/{project_id}/datasheet/{mpn}"
base = str(request.base_url).rstrip("/")
return {"url": f"{base}{proxy_path}"}
@router.get("/projects/{project_id}/datasheet/{mpn:path}")
async def get_datasheet_proxy(project_id: str, mpn: str, request: Request):
"""Stream a datasheet PDF from storage (GCS or local).
This is the proxy endpoint returned by get_datasheet_url.
"""
from fastapi.responses import Response
storage = get_storage(request)
owner_user_id, _ = await resolve_or_404(request, project_id)
_validate_mpn(mpn)
safe = safe_mpn(mpn)
key = _find_datasheet_key(storage, owner_user_id, project_id, safe, mpn=mpn)
if key is None:
raise HTTPException(404, f"Datasheet not found for MPN: {mpn}")
data = storage.read_bytes(key)
return Response(
content=data,
media_type="application/pdf",
headers={"Content-Disposition": f'inline; filename="{safe}.pdf"'},
)
@router.get("/datasheets/{mpn}")
async def get_datasheet(mpn: str, request: Request):
"""Serve a datasheet PDF (legacy local-dev endpoint)."""
from fastapi.responses import FileResponse
from backend.services.storage import LocalStorageBackend
storage = get_storage(request)
user_id = get_user_id(request)
_validate_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",
)
user_prefix = f"users/{user_id}/projects/"
for entry in storage.list_prefix(user_prefix):
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}")
+72
View File
@@ -0,0 +1,72 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Onboarding survey endpoints."""
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):
storage = get_storage(request)
user_id = get_user_id(request)
return {"completed": survey_svc.is_completed(storage, user_id)}
@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"}
# Resolve user email/name from Clerk if available
email = "unknown"
name = "unknown"
if settings.use_auth:
try:
from backend.services.email import _resolve_clerk_user
clerk_user = await _resolve_clerk_user(user_id)
if clerk_user:
emails = clerk_user.get("email_addresses", [])
email = emails[0].get("email_address", "unknown") if emails else "unknown"
first = clerk_user.get("first_name") or ""
last = clerk_user.get("last_name") or ""
name = f"{first} {last}".strip() or "unknown"
except Exception:
logger.warning("Failed to resolve Clerk user %s for survey", 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}
# Sheet write failed — don't mark completed so the user can retry
return {"ok": False, "detail": "sheet_write_failed"}
@@ -0,0 +1,51 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Global admin settings, persisted via StorageBackend.
Settings are stored at ``admin/settings.json`` in storage (GCS or local
``data/``). The module mirrors the pattern in ``limits.py``.
"""
from __future__ import annotations
from packaging.version import Version
from backend.services.storage import StorageBackend
_SETTINGS_KEY = "admin/settings.json"
_DEFAULTS: dict[str, str] = {
"min_model_version": "0.0.0", # no threshold by default
}
def get_admin_settings(storage: StorageBackend) -> dict:
"""Return the full admin settings dict, with defaults."""
if storage.exists(_SETTINGS_KEY):
data = storage.read_json(_SETTINGS_KEY)
return {**_DEFAULTS, **data}
return dict(_DEFAULTS)
def get_min_model_version(storage: StorageBackend) -> str:
"""Return the min_model_version threshold."""
return get_admin_settings(storage).get("min_model_version", "0.0.0")
def set_min_model_version(storage: StorageBackend, version: str) -> None:
"""Set the min_model_version threshold. Validates semver format."""
Version(version) # raises InvalidVersion if bad
data = get_admin_settings(storage)
data["min_model_version"] = version
storage.write_json(_SETTINGS_KEY, data)
def version_is_stale(component_version: str, min_version: str) -> bool:
"""Return True if *component_version* < *min_version* (semver)."""
if min_version == "0.0.0":
return False
try:
return Version(component_version) < Version(min_version)
except Exception:
return True # unparseable → treat as stale
@@ -0,0 +1,186 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Cross-process event bridge for pipeline progress.
Today the FastAPI API process and the pipeline worker (Cloud Run Job
execution, or a local subprocess in dev) live in different processes, so
the in-memory ``EventBroker`` in ``services.pipeline`` can't span them.
The bridge:
* Worker writes one object per event to
``users/{user_id}/projects/{project_id}/events/{seq:010d}.json``.
The object holds ``{seq, ts, event, data}``. The worker is the only
writer for a given run, so its local monotonic ``seq`` counter
needs no coordination.
* API SSE handler tails the same prefix via ``StorageBackend.list_prefix_after``,
yielding events in order until a terminal one arrives or the caller
cancels.
This avoids appending to a single JSONL on GCS (no append API; full
rewrite or compose-per-event has worse semantics) and naturally survives
SSE reconnects (consumer just resumes from its last seen ``seq``).
"""
from __future__ import annotations
import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import AsyncIterator
from backend.services.projects import project_prefix
from backend.services.storage import StorageBackend
logger = logging.getLogger(__name__)
# Filename pattern: 10-digit zero-padded seq + .json. Lexicographic order
# matches numeric order so list_prefix_after pages cleanly.
_SEQ_WIDTH = 10
_FILENAME_FMT = f"{{seq:0{_SEQ_WIDTH}d}}.json"
# Terminal event names — the SSE loop stops on these.
TERMINAL_EVENTS = frozenset({
"pipeline_complete",
"pipeline_error",
"pipeline_cancelled",
"pipeline_paused",
"placement_complete",
"placement_error",
"placement_cancelled",
"pcb_complete",
"pcb_error",
"pcb_cancelled",
})
def _events_prefix(user_id: str, project_id: str) -> str:
return f"{project_prefix(user_id, project_id)}/events/"
def _seq_from_key(key: str) -> int | None:
"""Extract the integer seq from an event key; ``None`` on parse failure."""
name = key.rsplit("/", 1)[-1]
if not name.endswith(".json"):
return None
stem = name[:-5]
try:
return int(stem)
except ValueError:
return None
class GCSEventBroker:
"""Drop-in for the in-memory ``EventBroker`` that persists to storage.
Same interface (``publish``, ``subscribe``, ``unsubscribe``,
``clear_history``) so the worker can swap it in for the module-level
``broker`` singleton without touching call sites. Subscription is a
no-op — the API consumes events via :func:`tail_events` instead.
"""
def __init__(self, storage: StorageBackend, user_id: str) -> None:
self.storage = storage
self.user_id = user_id
# Per-project local counter. Workers handle one project per
# execution, but the dict shape keeps parity with ``EventBroker``.
self._seq: dict[str, int] = {}
def subscribe(self, project_id: str) -> asyncio.Queue:
# Workers never subscribe — only the API tails the GCS event log.
# Returning an unfed queue is acceptable but raising is more
# honest about the contract.
raise NotImplementedError(
"GCSEventBroker is publish-only; subscribers should call "
"event_bridge.tail_events(...) instead."
)
def unsubscribe(self, project_id: str, q: asyncio.Queue) -> None:
# No-op for symmetry with the in-memory broker.
return
def clear_history(self, project_id: str) -> None:
"""Wipe all prior event objects for this project.
Called at the start of a fresh run so resumed/restarted runs
don't intermix with stale events from earlier attempts.
"""
prefix = _events_prefix(self.user_id, project_id)
try:
self.storage.delete_prefix(prefix)
except Exception:
logger.exception("failed to clear event history at %s", prefix)
self._seq[project_id] = 0
def publish(self, project_id: str, event: str, data: dict) -> None:
seq = self._seq.get(project_id, 0)
self._seq[project_id] = seq + 1
key = _events_prefix(self.user_id, project_id) + _FILENAME_FMT.format(seq=seq)
msg = {
"seq": seq,
"ts": datetime.now(timezone.utc).isoformat(),
"event": event,
"data": data,
}
try:
self.storage.write_json(key, msg)
except Exception:
# An event-write failure should never crash the pipeline.
logger.exception("failed to write event %s to %s", event, key)
async def tail_events(
storage: StorageBackend,
user_id: str,
project_id: str,
*,
poll_interval: float = 0.5,
heartbeat_interval: float = 15.0,
terminal_events: frozenset[str] | None = None,
) -> AsyncIterator[dict]:
"""Yield events from the GCS-backed event log in order.
Stops yielding after a terminal event (default ``TERMINAL_EVENTS``).
Emits a ``{"event": "heartbeat", "data": {}}`` synthetic event roughly
every ``heartbeat_interval`` seconds when no real events arrive.
The caller is expected to handle disconnects/cancellations and
secondary terminal-detection (``meta.status``, Cloud Run execution
state) on top of this iterator.
"""
stop_on = terminal_events if terminal_events is not None else TERMINAL_EVENTS
prefix = _events_prefix(user_id, project_id)
last_seen_key: str | None = None
last_emit_ts = 0.0
while True:
try:
keys = storage.list_prefix_after(prefix, after_key=last_seen_key)
except Exception:
logger.exception("event tail: list_prefix_after failed for %s", prefix)
keys = []
emitted_any = False
for key in keys:
try:
msg = storage.read_json(key)
except Exception:
logger.exception("event tail: read_json failed for %s", key)
continue
yield msg
emitted_any = True
last_seen_key = key
last_emit_ts = asyncio.get_event_loop().time()
if msg.get("event") in stop_on:
return
now = asyncio.get_event_loop().time()
if not emitted_any and now - last_emit_ts >= heartbeat_interval:
yield {"event": "heartbeat", "data": {}}
last_emit_ts = now
await asyncio.sleep(poll_interval)
@@ -0,0 +1,408 @@
# Native Periscope overlay: leftover app module still imported from src.
# PinScope original remains in dependency/.
"""Pipeline-worker dispatcher.
In production: enqueues a Cloud Run Job execution that runs the
``backend.pipeline_worker`` entrypoint with project_id/user_id/resume/free
passed as env-var overrides.
In local dev (no ``GCS_BUCKET``): launches the worker as a child process
so the same code path runs end-to-end. Removes the in-process
``BackgroundTask`` divergence between dev and prod.
"""
from __future__ import annotations
import logging
import os
import subprocess
import sys
import threading
from pathlib import Path
from typing import Literal
from backend.config import settings
logger = logging.getLogger(__name__)
ExecutionState = Literal[
"pending", "running", "succeeded", "failed", "cancelled", "unknown"
]
# ---------------------------------------------------------------------------
# Local subprocess fallback (dev mode)
# ---------------------------------------------------------------------------
# Track child processes so the API can query "is it still running?" in
# dev. In prod the Cloud Run Jobs admin API answers the same question.
_local_procs: dict[str, subprocess.Popen] = {}
_local_procs_lock = threading.Lock()
def _local_execution_name(project_id: str) -> str:
"""Stable synthetic execution name for the dev subprocess path.
Lets the rest of the codebase treat dev runs uniformly with prod
runs (we always have an ``execution_name`` to store on ProjectMeta
and pass to status / cancel calls).
"""
return f"local/projects/{project_id}"
def _spawn_local_subprocess(
project_id: str,
user_id: str,
*,
resume: bool,
free: bool,
mode: str = "run",
regen_stages: list[str] | None = None,
proc_key: str | None = None,
execution_name: str | None = None,
) -> str:
key = proc_key or project_id
name = execution_name or _local_execution_name(project_id)
env = os.environ.copy()
env["PROJECT_ID"] = project_id
env["USER_ID"] = user_id
env["RESUME"] = "1" if resume else "0"
env["FREE"] = "1" if free else "0"
env["MODE"] = mode
if regen_stages:
env["REGEN_STAGES"] = ",".join(regen_stages)
env["EXECUTION_NAME"] = name
proc = subprocess.Popen(
[sys.executable, "-m", "backend.pipeline_worker"],
env=env,
stdin=subprocess.DEVNULL,
)
_write_pid(key, proc.pid)
with _local_procs_lock:
# Reap any old proc for the same key before tracking the new one.
prior = _local_procs.pop(key, None)
if prior is not None:
try:
prior.terminate()
except Exception:
pass
_local_procs[key] = proc
logger.info(
"dev: spawned worker subprocess pid=%s for %s mode=%s",
proc.pid, project_id, mode,
)
return name
def _pid_path(project_id: str) -> Path:
return settings.data_dir / "workers" / f"{project_id}.pid"
def _write_pid(project_id: str, pid: int) -> None:
path = _pid_path(project_id)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(str(pid))
def _pid_alive(project_id: str) -> bool | None:
"""True/False if a pid file exists; None if there is no file."""
path = _pid_path(project_id)
if not path.is_file():
return None
try:
pid = int(path.read_text().strip())
except ValueError:
return False
try:
os.kill(pid, 0)
except OSError:
return False
return True
def _local_state(project_id: str) -> ExecutionState:
with _local_procs_lock:
proc = _local_procs.get(project_id)
if proc is not None:
rc = proc.poll()
if rc is None:
return "running"
if rc == 0:
return "succeeded"
if rc < 0:
# Negative return = terminated by signal
return "cancelled"
return "failed"
alive = _pid_alive(project_id)
if alive is True:
return "running"
if alive is False:
return "failed"
return "unknown"
def _local_cancel(project_id: str) -> None:
with _local_procs_lock:
proc = _local_procs.get(project_id)
if proc is None or proc.poll() is not None:
return
try:
proc.terminate()
except Exception:
logger.exception("dev: failed to terminate worker subprocess for %s", project_id)
# ---------------------------------------------------------------------------
# Cloud Run Jobs (prod path)
# ---------------------------------------------------------------------------
def _gcp_project() -> str:
"""Resolve the GCP project id for the Cloud Run Jobs admin API."""
if settings.pipeline_worker_project:
return settings.pipeline_worker_project
proj = os.environ.get("GOOGLE_CLOUD_PROJECT") or os.environ.get("GCLOUD_PROJECT")
if proj:
return proj
# Fall back to the metadata server (works on Cloud Run).
try:
import requests # type: ignore[import-not-found]
resp = requests.get(
"http://metadata.google.internal/computeMetadata/v1/project/project-id",
headers={"Metadata-Flavor": "Google"},
timeout=2.0,
)
if resp.ok:
return resp.text.strip()
except Exception:
pass
raise RuntimeError(
"Could not resolve GCP project for Cloud Run Jobs. Set "
"PIPELINE_WORKER_PROJECT or GOOGLE_CLOUD_PROJECT."
)
def _job_resource_name() -> str:
return (
f"projects/{_gcp_project()}/locations/{settings.pipeline_worker_region}"
f"/jobs/{settings.pipeline_worker_job_name}"
)
def _jobs_client():
# Lazy import: keeps the API process startup fast in local dev where
# google-cloud-run isn't even installed (it's an optional dep there).
from google.cloud import run_v2 # type: ignore[import-not-found]
return run_v2.JobsClient()
def _executions_client():
from google.cloud import run_v2 # type: ignore[import-not-found]
return run_v2.ExecutionsClient()
def _enqueue_cloud_run_job(
project_id: str,
user_id: str,
*,
resume: bool,
free: bool,
mode: str = "run",
regen_stages: list[str] | None = None,
) -> str:
"""Issue ``RunJob`` with env-var overrides; return the execution name."""
from google.cloud import run_v2 # type: ignore[import-not-found]
env_overrides = [
run_v2.EnvVar(name="PROJECT_ID", value=project_id),
run_v2.EnvVar(name="USER_ID", value=user_id),
run_v2.EnvVar(name="RESUME", value="1" if resume else "0"),
run_v2.EnvVar(name="FREE", value="1" if free else "0"),
run_v2.EnvVar(name="MODE", value=mode),
]
if regen_stages:
env_overrides.append(
run_v2.EnvVar(name="REGEN_STAGES", value=",".join(regen_stages)),
)
overrides = run_v2.RunJobRequest.Overrides(
container_overrides=[
run_v2.RunJobRequest.Overrides.ContainerOverride(env=env_overrides),
],
)
request = run_v2.RunJobRequest(name=_job_resource_name(), overrides=overrides)
operation = _jobs_client().run_job(request=request)
# Don't wait for completion — fire and forget. The metadata is enough
# to extract the execution resource name.
metadata = operation.metadata
name = getattr(metadata, "name", None) if metadata is not None else None
if not name:
# As a fallback, peek at the operation; on Cloud Run RunJob this
# is a long-running op whose initial metadata holds the execution.
name = operation.operation.name # type: ignore[union-attr]
if not name:
raise RuntimeError("Cloud Run RunJob returned no execution name")
logger.info("enqueued Cloud Run Job execution %s for project %s", name, project_id)
return name
def _cloud_run_state(execution_name: str) -> ExecutionState:
"""Map Cloud Run Execution state to our enum."""
try:
from google.cloud import run_v2 # type: ignore[import-not-found]
client = _executions_client()
ex = client.get_execution(name=execution_name)
except Exception:
logger.exception("get_execution failed for %s", execution_name)
return "unknown"
# An Execution has reconciliation_started, completion_time, conditions.
# Map to our enum based on completion + conditions.
if ex.completion_time is None or ex.completion_time.seconds == 0:
if ex.start_time and ex.start_time.seconds:
return "running"
return "pending"
# Completed — figure out success vs failure.
failed = int(getattr(ex, "failed_count", 0) or 0)
cancelled = int(getattr(ex, "cancelled_count", 0) or 0)
succeeded = int(getattr(ex, "succeeded_count", 0) or 0)
if cancelled > 0 and succeeded == 0:
return "cancelled"
if failed > 0:
return "failed"
if succeeded > 0:
return "succeeded"
return "unknown"
def _cloud_run_cancel(execution_name: str) -> None:
try:
from google.cloud import run_v2 # type: ignore[import-not-found]
request = run_v2.CancelExecutionRequest(name=execution_name)
_executions_client().cancel_execution(request=request)
except Exception:
logger.exception("cancel_execution failed for %s", execution_name)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def use_cloud_run_jobs() -> bool:
"""True iff we should dispatch via Cloud Run Jobs.
Tied to whether GCS storage is configured — Jobs and GCS go together
in prod, and local dev uses neither.
"""
return bool(settings.gcs_bucket)
def enqueue_pipeline(
project_id: str, user_id: str, *, resume: bool = False, free: bool = False,
) -> str:
"""Dispatch a pipeline run.
In prod, returns the Cloud Run Execution resource name. In dev,
returns a synthetic ``local/projects/{id}`` name. Either way, callers
should persist the returned name on ``ProjectMeta.execution_name``.
"""
if use_cloud_run_jobs():
return _enqueue_cloud_run_job(project_id, user_id, resume=resume, free=free)
return _spawn_local_subprocess(project_id, user_id, resume=resume, free=free)
def enqueue_pipeline_regen(
project_id: str, user_id: str, *, stages: list[str],
) -> str:
"""Dispatch a regen run (graph + selected stages, free).
Same image, same worker; differs only in the env-var-driven mode.
"""
if not stages:
raise ValueError("regen requires at least one stage")
if use_cloud_run_jobs():
return _enqueue_cloud_run_job(
project_id, user_id, resume=False, free=True,
mode="regen", regen_stages=stages,
)
return _spawn_local_subprocess(
project_id, user_id, resume=False, free=True,
mode="regen", regen_stages=stages,
)
def get_execution_state(execution_name: str | None) -> ExecutionState:
"""Return current state of a previously-enqueued execution.
Used by the SSE handler's hard-crash escape hatch and by the
stale-running sweeper. ``None`` -> ``"unknown"``.
"""
if not execution_name:
return "unknown"
if execution_name.startswith("local/projects/"):
project_id = execution_name.split("/", 2)[-1]
return _local_state(project_id)
if execution_name.startswith("local/placement/"):
project_id = execution_name.split("/", 2)[-1]
return _local_state(f"placement:{project_id}")
if execution_name.startswith("local/pcb/"):
project_id = execution_name.split("/", 2)[-1]
return _local_state(f"pcb:{project_id}")
return _cloud_run_state(execution_name)
def cancel_execution(execution_name: str | None) -> None:
"""Hard-cancel an execution (Cloud Run cancel or local SIGTERM).
Best-effort. Soft cancel via ``meta.cancel_requested`` is preferred —
only fall back to this when the worker has already gone unresponsive.
"""
if not execution_name:
return
if execution_name.startswith("local/projects/"):
project_id = execution_name.split("/", 2)[-1]
_local_cancel(project_id)
return
if execution_name.startswith("local/placement/"):
project_id = execution_name.split("/", 2)[-1]
_local_cancel(f"placement:{project_id}")
return
if execution_name.startswith("local/pcb/"):
project_id = execution_name.split("/", 2)[-1]
_local_cancel(f"pcb:{project_id}")
return
_cloud_run_cancel(execution_name)
def enqueue_placement_pipeline(project_id: str, user_id: str) -> str:
"""Dispatch the parallel Placement pipeline (topology plan, no LLM)."""
if use_cloud_run_jobs():
return _enqueue_cloud_run_job(
project_id, user_id, resume=False, free=True, mode="placement",
)
return _spawn_local_subprocess(
project_id, user_id, resume=False, free=True, mode="placement",
proc_key=f"placement:{project_id}",
execution_name=f"local/placement/{project_id}",
)
def enqueue_pcb_pipeline(project_id: str, user_id: str) -> str:
"""Dispatch the parallel PCB review pipeline (deterministic + AI exam)."""
if use_cloud_run_jobs():
return _enqueue_cloud_run_job(
project_id, user_id, resume=False, free=False, mode="pcb",
)
return _spawn_local_subprocess(
project_id, user_id, resume=False, free=False, mode="pcb",
proc_key=f"pcb:{project_id}",
execution_name=f"local/pcb/{project_id}",
)
File diff suppressed because it is too large Load Diff
@@ -1,6 +1,6 @@
# Piano — indipendenza architettonica e di licenza da PinScope
**Stato:** split **2.38.0**. C2C4 as before. **2.41.0** job workspace. **2.42.0** graph/parsers/models/taxonomy overlay. **2.43.0** leftover helpers + live `pipeline`/`validation`/`extraction`/`validate` overlay; skills + taxonomy JSON copied to `periscope/src`. **2.44.0** leftover services + LLM factory/types/base overlay. Originals **not** deleted. Fork non staccato.
**Stato:** split **2.38.0**. C2C4 as before. **2.41.0** job workspace. **2.42.0** graph/parsers/models/taxonomy overlay. **2.43.0** leftover helpers + live `pipeline`/`validation`/`extraction`/`validate` overlay; skills + taxonomy JSON copied to `periscope/src`. **2.44.0** leftover services + LLM factory/types/base overlay. **2.45.0** leftover app entry / routers / jobs overlay. Originals **not** deleted. Fork non staccato.
**Gate Michele:** sostituire/smettere di chiamare un modulo `dependency/` solo dopo pytest + deploy smoke. Se la verifica fallisce, resta il path ereditato.
**Sequenza:** split → sostituzione incrementale (C2…C5 overlay). **Mai** empty-delete. Auto-place fuori scope. AGPL resta.
@@ -32,7 +32,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope:
| Pacchetto logico | Path **dopo lo split** | Licenza da audit |
| --- | --- | --- |
| Core schematico PinScope | `periscope/dependency/backend/periscopex/…`**overlay 2.42.02.43.0** in `periscope/src` (same names; keep inherited files) | AGPL-3.0 del fork (blob `LICENSE` identico a upstream) |
| Orchestrazione review | `pipeline_worker.py` still dependency-only; `pipeline.py` / `validation.py` / `extraction.py` **overlay 2.43.0**; leftover services **2.44.0** in src | stessa |
| Orchestrazione review | `pipeline.py` / `validation.py` / `extraction.py` **overlay 2.43.0**; leftover services **2.44.0**; `main` / `config` / middleware / leftover routers / `pipeline_worker` / jobs **2.45.0** in src | stessa |
| Skills Anthropic/Console | `periscope/dependency/skills/…` kept; **src copy 2.43.0** `periscope/src/skills/` + `src/backend/skills_manifest.json`. `upload_skills.py` still dependency | stessa + contratto Claude |
| UI OSS / marketing shell | `periscope/dependency/frontend/` (UPSTREAM/DERIVED); file nativi in `periscope/src/frontend/` con symlink nel recinto | AGPL |
| Gateway seams | `billing_hook.py`, `proxy.ts`, `clerk-theme-provider.tsx`, … sotto `periscope/dependency/` | stub open-core PinScope |
@@ -64,6 +64,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope:
| Graph / parsers / models / taxonomy | `periscope/src/backend/periscopex/{graph,parsers,parsers_edif,models,taxonomy}.py` | OVERLAY 2.42.0; inherited copies kept |
| Leftover helpers + analysis overlay | `utils`, `resolve_passives`, `derating`, `bom_summary`, `pin_mux_check`, `led_current_check`, `pin_function_tokens`, `validate`, `validation_tools`, `services/{pipeline,validation,extraction}.py`, `src/taxonomy`, `src/skills` | OVERLAY 2.43.0 |
| Leftover services overlay | `api_logs`, `normalize_findings`, `dedupe_findings`, `billing_hook`, `datasheet_store`, `cost_estimator`, `storage`, `purple_parts`, `email`, `digikey`, `llm/{factory,types,base}` | OVERLAY 2.44.0 |
| Leftover app/entry overlay | `main.py`, `config.py`, `middleware/auth.py`, leftover `routers/*`, `pipeline_worker.py`, `job_runner`, `event_bridge`, `projects`, `admin_settings` | OVERLAY 2.45.0 |
| Deploy | `scripts/update-periscope.sh`, `docker-compose.yml`, `periscope/src/backend/Dockerfile` | NEW (nomi `pinscope_*` ancora WEAK) |
| Plugin KiCad | `periscope/src/plugins/kicad/` | NEW |
| Check deterministici fork | `dnp_check`, `sequencing_check`, `layout_rules`, … under `periscope/src` | NEW ma **INDIRECT**: usano `models` / graph |
+73
View File
@@ -0,0 +1,73 @@
"""Native overlay: leftover app/entry modules resolve from src."""
from __future__ import annotations
from pathlib import Path
import backend.config as config
import backend.main as main
import backend.middleware.auth as middleware_auth
import backend.pipeline_worker as pipeline_worker
import backend.routers.admin as routers_admin
import backend.routers.auth as routers_auth
import backend.routers.contact as routers_contact
import backend.routers.deps as routers_deps
import backend.routers.feedback as routers_feedback
import backend.routers.impedance as routers_impedance
import backend.routers.pipeline as routers_pipeline
import backend.routers.projects as routers_projects
import backend.routers.reports as routers_reports
import backend.routers.survey as routers_survey
import backend.services.admin_settings as admin_settings
import backend.services.event_bridge as event_bridge
import backend.services.job_runner as job_runner
import backend.services.projects as projects
def _src(mod) -> Path:
return Path(mod.__file__).resolve()
def _overlay_text(mod) -> str:
return _src(mod).read_text(encoding="utf-8")[:500]
def test_leftover_app_modules_load_from_src():
root = Path(__file__).resolve().parents[1]
be = (root / "periscope" / "src" / "backend").resolve()
for mod, rel in (
(main, "main.py"),
(config, "config.py"),
(pipeline_worker, "pipeline_worker.py"),
(middleware_auth, "middleware/auth.py"),
(job_runner, "services/job_runner.py"),
(event_bridge, "services/event_bridge.py"),
(projects, "services/projects.py"),
(admin_settings, "services/admin_settings.py"),
(routers_admin, "routers/admin.py"),
(routers_contact, "routers/contact.py"),
(routers_deps, "routers/deps.py"),
(routers_feedback, "routers/feedback.py"),
(routers_pipeline, "routers/pipeline.py"),
(routers_projects, "routers/projects.py"),
(routers_reports, "routers/reports.py"),
(routers_survey, "routers/survey.py"),
):
path = _src(mod)
assert path == be / rel, path
assert "Native Periscope overlay" in _overlay_text(mod)
def test_native_auth_and_impedance_routers_untouched():
root = Path(__file__).resolve().parents[1]
be = (root / "periscope" / "src" / "backend").resolve()
assert _src(routers_auth) == be / "routers" / "auth.py"
assert _src(routers_impedance) == be / "routers" / "impedance.py"
assert "Local Periscope auth endpoints" in _src(routers_auth).read_text(encoding="utf-8")[:200]
def test_main_still_exports_fastapi_app():
from fastapi import FastAPI
assert isinstance(main.app, FastAPI)
assert main.LOCAL_DEV_USER == "local"