395 lines
15 KiB
Python
395 lines
15 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
CodeEditor Feature Routes.
|
|
SSE-based endpoints for Cursor-style AI file editing.
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
import asyncio
|
|
from typing import Optional, Dict, Any, List
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Body, Path, Query, Request
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from modules.auth import limiter, getRequestContext, RequestContext
|
|
from modules.interfaces import interfaceDbChat, interfaceDbManagement
|
|
from modules.interfaces.interfaceAiObjects import AiObjects
|
|
from modules.datamodels.datamodelChat import UserInputRequest
|
|
from modules.services.serviceStreaming import get_event_manager
|
|
from modules.features.codeeditor import codeEditorProcessor, fileContextManager
|
|
from modules.features.codeeditor.datamodelCodeeditor import FileEditProposal, EditStatusEnum
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/codeeditor",
|
|
tags=["Code Editor Feature"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
_aiObjects: Optional[AiObjects] = None
|
|
|
|
|
|
async def _getAiObjects() -> AiObjects:
|
|
"""Lazy-init singleton for AiObjects."""
|
|
global _aiObjects
|
|
if _aiObjects is None:
|
|
_aiObjects = await AiObjects.create()
|
|
return _aiObjects
|
|
|
|
|
|
def _getServiceChat(context: RequestContext, featureInstanceId: str = None):
|
|
"""Get chat interface with feature instance context."""
|
|
return interfaceDbChat.getInterface(
|
|
context.user,
|
|
mandateId=str(context.mandateId) if context.mandateId else None,
|
|
featureInstanceId=featureInstanceId
|
|
)
|
|
|
|
|
|
def _getDbManagement(context: RequestContext, featureInstanceId: str = None):
|
|
"""Get management interface with user context for file access."""
|
|
return interfaceDbManagement.getInterface(
|
|
context.user,
|
|
mandateId=str(context.mandateId) if context.mandateId else None,
|
|
featureInstanceId=featureInstanceId
|
|
)
|
|
|
|
|
|
def _validateInstanceAccess(instanceId: str, context: RequestContext) -> str:
|
|
"""Validate user has access to the feature instance. Returns mandateId."""
|
|
from modules.interfaces.interfaceDbApp import getRootInterface
|
|
|
|
rootInterface = getRootInterface()
|
|
instance = rootInterface.getFeatureInstance(instanceId)
|
|
if not instance:
|
|
raise HTTPException(status_code=404, detail=f"Feature instance {instanceId} not found")
|
|
|
|
featureAccess = rootInterface.getFeatureAccess(str(context.user.id), instanceId)
|
|
if not featureAccess or not featureAccess.enabled:
|
|
raise HTTPException(status_code=403, detail="Access denied to this feature instance")
|
|
|
|
return str(instance.mandateId) if instance.mandateId else None
|
|
|
|
|
|
@router.post("/{instanceId}/start/stream")
|
|
@limiter.limit("60/minute")
|
|
async def streamCodeeditorStart(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: Optional[str] = Query(None, description="Optional workflow ID to continue"),
|
|
mode: str = Query("simple", description="Processing mode: 'simple' (single AI call) or 'agent' (multi-step with tools)"),
|
|
userInput: UserInputRequest = Body(...),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
):
|
|
"""Start or continue a CodeEditor workflow with SSE streaming. Supports simple and agent mode."""
|
|
try:
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
chatInterface = _getServiceChat(context, featureInstanceId=instanceId)
|
|
dbManagement = _getDbManagement(context, featureInstanceId=instanceId)
|
|
aiObjects = await _getAiObjects()
|
|
eventManager = get_event_manager()
|
|
|
|
if workflowId:
|
|
workflow = chatInterface.getWorkflow(workflowId)
|
|
if not workflow:
|
|
raise HTTPException(status_code=404, detail=f"Workflow {workflowId} not found")
|
|
else:
|
|
workflow = chatInterface.createWorkflow({
|
|
"workflowMode": "CodeEditor",
|
|
"status": "running",
|
|
"label": userInput.prompt[:80] if userInput.prompt else "CodeEditor Session",
|
|
})
|
|
workflowId = workflow.get("id") if isinstance(workflow, dict) else workflow.id
|
|
|
|
queue = eventManager.create_queue(workflowId)
|
|
|
|
userMessage = {
|
|
"role": "user",
|
|
"content": userInput.prompt,
|
|
"selectedFiles": userInput.listFileId or []
|
|
}
|
|
await eventManager.emit_event(workflowId, "chatdata", {
|
|
"type": "message", "item": userMessage
|
|
})
|
|
|
|
selectedFileIds = userInput.listFileId or []
|
|
|
|
agentMode = mode.lower() == "agent"
|
|
|
|
asyncio.create_task(
|
|
codeEditorProcessor.processMessage(
|
|
workflowId=workflowId,
|
|
userPrompt=userInput.prompt,
|
|
selectedFileIds=selectedFileIds,
|
|
dbManagement=dbManagement,
|
|
interfaceAi=aiObjects,
|
|
chatInterface=chatInterface,
|
|
eventManager=eventManager,
|
|
agentMode=agentMode
|
|
)
|
|
)
|
|
|
|
async def _eventStream():
|
|
streamTimeout = 300
|
|
lastActivity = asyncio.get_event_loop().time()
|
|
|
|
while True:
|
|
now = asyncio.get_event_loop().time()
|
|
if now - lastActivity > streamTimeout:
|
|
yield f"data: {json.dumps({'type': 'error', 'error': 'Stream timeout'})}\n\n"
|
|
break
|
|
|
|
if await request.is_disconnected():
|
|
logger.info(f"Client disconnected for workflow {workflowId}")
|
|
break
|
|
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout=1.0)
|
|
lastActivity = asyncio.get_event_loop().time()
|
|
|
|
eventType = event.get("type", "")
|
|
|
|
if eventType == "chatdata":
|
|
yield f"data: {json.dumps(event.get('data', {}))}\n\n"
|
|
elif eventType in ("complete", "stopped", "error"):
|
|
yield f"data: {json.dumps({'type': eventType, **event.get('data', {})})}\n\n"
|
|
break
|
|
else:
|
|
yield f"data: {json.dumps(event)}\n\n"
|
|
|
|
except asyncio.TimeoutError:
|
|
yield ": keepalive\n\n"
|
|
|
|
await eventManager.cleanup(workflowId)
|
|
|
|
return StreamingResponse(
|
|
_eventStream(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no"
|
|
}
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error in streamCodeeditorStart: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/{instanceId}/{workflowId}/stop")
|
|
@limiter.limit("120/minute")
|
|
async def stopWorkflow(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
):
|
|
"""Stop a running CodeEditor workflow."""
|
|
try:
|
|
_validateInstanceAccess(instanceId, context)
|
|
eventManager = get_event_manager()
|
|
await eventManager.emit_event(workflowId, "stopped", {
|
|
"workflowId": workflowId
|
|
}, event_category="workflow", message="Workflow stopped by user")
|
|
return {"status": "stopped", "workflowId": workflowId}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error stopping workflow: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/{instanceId}/{workflowId}/chatData")
|
|
@limiter.limit("120/minute")
|
|
def getWorkflowChatData(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
afterTimestamp: Optional[float] = Query(None, description="Unix timestamp for incremental fetch"),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
) -> Dict[str, Any]:
|
|
"""Get chat data for a workflow (polling fallback)."""
|
|
try:
|
|
_validateInstanceAccess(instanceId, context)
|
|
chatInterface = _getServiceChat(context, featureInstanceId=instanceId)
|
|
workflow = chatInterface.getWorkflow(workflowId)
|
|
if not workflow:
|
|
raise HTTPException(status_code=404, detail=f"Workflow {workflowId} not found")
|
|
return chatInterface.getUnifiedChatData(workflowId, afterTimestamp)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting chat data: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/{instanceId}/workflows")
|
|
@limiter.limit("120/minute")
|
|
def getWorkflows(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(20, ge=1, le=100),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
) -> Dict[str, Any]:
|
|
"""List workflows for this feature instance."""
|
|
try:
|
|
_validateInstanceAccess(instanceId, context)
|
|
chatInterface = _getServiceChat(context, featureInstanceId=instanceId)
|
|
from modules.datamodels.datamodelPagination import PaginationParams
|
|
pagination = PaginationParams(page=page, pageSize=pageSize)
|
|
return chatInterface.getWorkflows(pagination=pagination)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting workflows: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/{instanceId}/files")
|
|
@limiter.limit("120/minute")
|
|
def getFiles(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
) -> Dict[str, Any]:
|
|
"""List all text files accessible to the user."""
|
|
try:
|
|
_validateInstanceAccess(instanceId, context)
|
|
dbManagement = _getDbManagement(context, featureInstanceId=instanceId)
|
|
textFiles = fileContextManager.listTextFiles(dbManagement)
|
|
return {
|
|
"files": [f.model_dump(exclude={"content"}) for f in textFiles],
|
|
"count": len(textFiles)
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error listing files: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/{instanceId}/files/{fileId}/content")
|
|
@limiter.limit("120/minute")
|
|
def getFileContent(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
fileId: str = Path(..., description="File ID"),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
) -> Dict[str, Any]:
|
|
"""Get the text content of a file."""
|
|
try:
|
|
_validateInstanceAccess(instanceId, context)
|
|
dbManagement = _getDbManagement(context, featureInstanceId=instanceId)
|
|
|
|
fileItem = dbManagement.getFile(fileId)
|
|
if not fileItem:
|
|
raise HTTPException(status_code=404, detail=f"File {fileId} not found")
|
|
|
|
fileData = dbManagement.getFileData(fileId)
|
|
if not fileData:
|
|
raise HTTPException(status_code=404, detail=f"No data for file {fileId}")
|
|
|
|
try:
|
|
content = fileData.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise HTTPException(status_code=400, detail="File is not valid UTF-8 text")
|
|
|
|
return {
|
|
"fileId": fileId,
|
|
"fileName": fileItem.fileName,
|
|
"mimeType": fileItem.mimeType,
|
|
"content": content
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting file content: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/{instanceId}/{workflowId}/apply")
|
|
@limiter.limit("60/minute")
|
|
async def applyEdit(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
proposalData: Dict[str, Any] = Body(...),
|
|
context: RequestContext = Depends(getRequestContext)
|
|
) -> Dict[str, Any]:
|
|
"""Accept a file edit proposal. Updates existing file or creates new one."""
|
|
try:
|
|
_validateInstanceAccess(instanceId, context)
|
|
dbManagement = _getDbManagement(context, featureInstanceId=instanceId)
|
|
|
|
fileId = proposalData.get("fileId", "")
|
|
newContent = proposalData.get("newContent")
|
|
fileName = proposalData.get("fileName", "")
|
|
|
|
if newContent is None:
|
|
raise HTTPException(status_code=400, detail="newContent is required")
|
|
|
|
contentBytes = newContent.encode("utf-8")
|
|
isNewFile = not fileId or fileId.startswith("unknown-")
|
|
|
|
if isNewFile:
|
|
mimeType = _guessMimeType(fileName)
|
|
fileItem = dbManagement.createFile(fileName, mimeType, contentBytes)
|
|
resultFileId = fileItem.id
|
|
resultFileName = fileItem.fileName
|
|
else:
|
|
fileItem = dbManagement.getFile(fileId)
|
|
if not fileItem:
|
|
raise HTTPException(status_code=404, detail=f"File {fileId} not found")
|
|
success = dbManagement.createFileData(fileId, contentBytes)
|
|
if not success:
|
|
raise HTTPException(status_code=500, detail="Failed to store updated file content")
|
|
resultFileId = fileId
|
|
resultFileName = fileName or fileItem.fileName
|
|
|
|
eventManager = get_event_manager()
|
|
await eventManager.emit_event(workflowId, "chatdata", {
|
|
"type": "file_version",
|
|
"item": {
|
|
"fileId": resultFileId,
|
|
"fileName": resultFileName,
|
|
"status": "accepted",
|
|
"isNew": isNewFile
|
|
}
|
|
})
|
|
|
|
return {
|
|
"status": "accepted",
|
|
"fileId": resultFileId,
|
|
"fileName": resultFileName,
|
|
"isNew": isNewFile
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error applying edit: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
_MIME_MAP = {
|
|
".md": "text/markdown", ".txt": "text/plain", ".json": "application/json",
|
|
".yaml": "application/yaml", ".yml": "application/yaml", ".xml": "application/xml",
|
|
".csv": "text/csv", ".py": "text/x-python", ".js": "text/javascript",
|
|
".ts": "text/x-typescript", ".html": "text/html", ".css": "text/css",
|
|
".sql": "text/x-sql", ".sh": "text/x-shellscript",
|
|
}
|
|
|
|
|
|
def _guessMimeType(fileName: str) -> str:
|
|
"""Guess MIME type from file extension."""
|
|
if not fileName or "." not in fileName:
|
|
return "text/plain"
|
|
ext = "." + fileName.rsplit(".", 1)[-1].lower()
|
|
return _MIME_MAP.get(ext, "text/plain")
|