michele
16c606ae3d
Parallel pcb_status pipeline: parse board, classify domains/groups, inventory traces, PE-LAY/PLC/SI/DRT checks, per-IC datasheet review. Findings merge into the report UI. No auto-place or pcbnew write-back.
978 lines
38 KiB
Python
978 lines
38 KiB
Python
"""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 1–60s.
|
||
"""
|
||
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)
|
||
err = None
|
||
if cur and cur.pcb_state:
|
||
err = cur.pcb_state.get("error")
|
||
yield {
|
||
"event": "pcb_error",
|
||
"data": json.dumps({
|
||
"error": err or crash_detected["reason"]
|
||
or "pcb 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",
|
||
},
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|