periscope/backend/routers/pipeline.py

"""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_"):
                    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"),
        "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

    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 (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"
                ):
                    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",
        },
    )


# ---------------------------------------------------------------------------
# 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