state-hub/api/routers/execution.py

import uuid

from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.database import get_session
from api.models.task import Task, TaskStatus
from api.models.workplan_launch_request import WorkplanLaunchRequest
from api.models.workplan import Workplan
from api.models.workplan_dependency import WorkplanDependency
from api.schemas.execution import (
    ExecutionIntentRead,
    ExecutionIntentUpdate,
    ExecutionSemantics,
    LaunchRequestCreate,
    LaunchRequestRead,
    WorkplanQueueItem,
)
from api.services.execution_queue import (
    ACTIVITY_CORE_RESPONSIBILITIES,
    CONCURRENCY_MODES,
    EXECUTION_STATES,
    EXECUTION_REPLACEMENTS,
    LAUNCH_MODES,
    STATE_HUB_RESPONSIBILITIES,
    queue_sort_key,
    workplan_blockers,
)
from api.routers.workstreams import _legacy_key
from api.services.legacy_compat import retire_legacy_route
from api.services.legacy_compat import meter_legacy_query_param
from api.workplan_status import CLOSED_WORKPLAN_STATUSES, normalize_workplan_status

router = APIRouter(prefix="/execution", tags=["execution"])


@router.get("/semantics", response_model=ExecutionSemantics)
async def execution_semantics() -> ExecutionSemantics:
    return ExecutionSemantics(
        execution_states=EXECUTION_STATES,
        launch_modes=LAUNCH_MODES,
        concurrency_modes=CONCURRENCY_MODES,
        state_hub_responsibility=STATE_HUB_RESPONSIBILITIES,
        activity_core_responsibility=ACTIVITY_CORE_RESPONSIBILITIES,
        launch_requests_accepted=False,
        replacements=EXECUTION_REPLACEMENTS,
    )


async def _update_execution_intent(
    *,
    workstream_id: uuid.UUID,
    body: ExecutionIntentUpdate,
    session: AsyncSession,
) -> ExecutionIntentRead:
    ws = await session.get(Workplan, workstream_id)
    if ws is None:
        raise HTTPException(status_code=404, detail="Workplan not found")

    for field, value in body.model_dump(exclude_unset=True).items():
        setattr(ws, field, value)
    await session.commit()
    await session.refresh(ws)
    return _intent_read(ws)


@router.patch("/workstreams/{workstream_id}/intent", status_code=status.HTTP_410_GONE)
async def update_execution_intent(
    request: Request,
    response: Response,
    workstream_id: uuid.UUID,
    body: ExecutionIntentUpdate,
    session: AsyncSession = Depends(get_session),
) -> None:
    await retire_legacy_route(
        session=session,
        request=request,
        response=response,
        interface_key=_legacy_key("PATCH", "/execution/workstreams/{workstream_id}/intent"),
        replacement_ref="/execution/workplans/{workplan_id}/intent",
        detail="Legacy PATCH /execution/workstreams/{workstream_id}/intent retired; "
        "use PATCH /execution/workplans/{workplan_id}/intent",
    )


@router.patch("/workplans/{workplan_id}/intent", response_model=ExecutionIntentRead)
async def update_workplan_execution_intent(
    workplan_id: uuid.UUID,
    body: ExecutionIntentUpdate,
    session: AsyncSession = Depends(get_session),
) -> ExecutionIntentRead:
    return await _update_execution_intent(workstream_id=workplan_id, body=body, session=session)


@router.get("/workplan-stack", response_model=list[WorkplanQueueItem])
async def workplan_stack(
    include_manual: bool = Query(True),
    include_blocked: bool = Query(True),
    session: AsyncSession = Depends(get_session),
) -> list[WorkplanQueueItem]:
    result = await session.execute(select(Workplan))
    workstreams = [
        ws for ws in result.scalars().all()
        if normalize_workplan_status(ws.status) not in CLOSED_WORKPLAN_STATUSES
    ]
    ws_by_id = {ws.id: ws for ws in workstreams}
    ws_status = {ws.id: normalize_workplan_status(ws.status) for ws in workstreams}

    dep_result = await session.execute(select(WorkplanDependency))
    ws_deps: dict[uuid.UUID, list[uuid.UUID]] = {}
    task_deps: dict[uuid.UUID, list[uuid.UUID]] = {}
    for dep in dep_result.scalars().all():
        if dep.to_workplan_id is not None:
            ws_deps.setdefault(dep.from_workplan_id, []).append(dep.to_workplan_id)
        if dep.to_task_id is not None:
            task_deps.setdefault(dep.from_workplan_id, []).append(dep.to_task_id)

    task_ids = [task_id for ids in task_deps.values() for task_id in ids]
    task_status: dict[uuid.UUID, str] = {}
    if task_ids:
        task_result = await session.execute(select(Task).where(Task.id.in_(task_ids)))
        task_status = {task.id: _task_status(task.status) for task in task_result.scalars().all()}

    items: list[WorkplanQueueItem] = []
    for ws in workstreams:
        if not include_manual and ws.execution_state == "manual":
            continue
        lifecycle_status = normalize_workplan_status(ws.status)
        blocked_ws = [
            blocker for blocker in workplan_blockers(ws.id, ws_deps, ws_status)
            if blocker in ws_by_id or blocker in ws_status
        ]
        blocked_tasks = [
            task_id for task_id in task_deps.get(ws.id, [])
            if task_status.get(task_id) not in {"done", "cancel"}
        ]
        eligible = lifecycle_status != "blocked" and not blocked_ws and not blocked_tasks
        if not include_blocked and not eligible:
            continue
        sort_key = queue_sort_key(ws, eligible=eligible)
        items.append(WorkplanQueueItem(
            workplan_id=ws.id,
            slug=ws.slug,
            title=ws.title,
            status=lifecycle_status,
            repo_id=ws.repo_id,
            planning_priority=ws.planning_priority,
            planning_order=ws.planning_order,
            execution_state=ws.execution_state,
            launch_mode=ws.launch_mode,
            concurrency_mode=ws.concurrency_mode,
            queue_rank=ws.queue_rank,
            execution_group=ws.execution_group,
            scheduled_for=ws.scheduled_for,
            eligible=eligible,
            blocked_by_workplan_ids=blocked_ws,
            blocked_by_task_ids=blocked_tasks,
            sort_key=sort_key,
        ))
    return sorted(items, key=lambda item: item.sort_key)


@router.post(
    "/launch-requests",
    status_code=status.HTTP_410_GONE,
)
async def create_launch_request(
    request: Request,
    response: Response,
    body: LaunchRequestCreate,
    session: AsyncSession = Depends(get_session),
) -> None:
    del body  # Request shape stays documented while the retired route returns 410.
    await retire_legacy_route(
        session=session,
        request=request,
        response=response,
        interface_key="rest_api:POST /execution/launch-requests",
        replacement_ref="repo file queue or activity-core ActivityDefinition + ops_run",
        detail=(
            "State Hub workplan launch requests are retired: no consumer picks up these rows. "
            "Queue development work in the authoritative repository file; use an "
            "ActivityDefinition and activity-core ops_run only for recurring or operational fires."
        ),
    )


@router.get("/launch-requests", response_model=list[LaunchRequestRead])
async def list_launch_requests(
    request: Request,
    response: Response,
    workplan_id: uuid.UUID | None = None,
    workstream_id: uuid.UUID | None = None,
    request_status: str | None = None,
    session: AsyncSession = Depends(get_session),
) -> list[WorkplanLaunchRequest]:
    if workstream_id is not None and workplan_id is None:
        await meter_legacy_query_param(
            session=session,
            request=request,
            response=response,
            method="GET",
            route="/execution/launch-requests",
            replacement_ref="/execution/launch-requests?workplan_id=<workplan_id>",
        )
    q = select(WorkplanLaunchRequest).order_by(WorkplanLaunchRequest.created_at.desc())
    scope_id = workplan_id or workstream_id
    if scope_id:
        q = q.where(WorkplanLaunchRequest.workplan_id == scope_id)
    if request_status:
        q = q.where(WorkplanLaunchRequest.status == request_status)
    result = await session.execute(q)
    return list(result.scalars().all())


def _intent_read(ws: Workplan) -> ExecutionIntentRead:
    return ExecutionIntentRead(
        workplan_id=ws.id,
        execution_state=ws.execution_state,
        launch_mode=ws.launch_mode,
        concurrency_mode=ws.concurrency_mode,
        queue_rank=ws.queue_rank,
        execution_group=ws.execution_group,
        scheduled_for=ws.scheduled_for,
    )


def _task_status(status_value: TaskStatus | str) -> str:
    if hasattr(status_value, "value"):
        return status_value.value
    return str(status_value or "").strip().lower()