Function manual_transition

backend/app/api/workflow.py:135–165

Manually transition to a different phase.

(
    pipeline_id: str,
    target_phase: WorkflowPhase,
    reason: Optional[str] = None,
    manager: WorkflowManager = Depends(get_manager)
)

@router.post("/workflow/{pipeline_id}/transition")
async def manual_transition(
    pipeline_id: str,
    target_phase: WorkflowPhase,
    reason: Optional[str] = None,
    manager: WorkflowManager = Depends(get_manager)
):
    """Manually transition to a different phase."""
    workflow = manager.get_workflow(pipeline_id)
    if not workflow:
        raise HTTPException(status_code=404, detail="Workflow not found")
    
    if not workflow.can_transition_to(target_phase):
        raise HTTPException(
            status_code=400,
            detail=f"Cannot transition from {workflow.phase} to {target_phase}"
        )
    
    workflow.add_transition(
        target_phase,
        reason or "Manual transition",
        "user"
    )
    
    # Broadcast update
    await manager._broadcast_workflow_update(workflow)
    
    return {
        "success": True,
        "workflow": workflow.dict(),
        "message": f"Transitioned to {target_phase}"
    }


@router.get("/workflow/{pipeline_id}/history")