File size: 4,639 Bytes
42d88ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | import asyncio
import json
from dataclasses import asdict
from typing import Optional
from uuid import UUID
from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from Backend.database.connection import get_db
from Backend.database.models import Issue, IssueEvent
from Backend.core.flow_tracker import get_flow_tracker, _active_flows
router = APIRouter()
async def event_generator(issue_id: UUID, timeout: int = 300):
tracker = get_flow_tracker(issue_id)
if not tracker:
yield f"data: {json.dumps({'type': 'error', 'message': 'No active flow for this issue'})}\n\n"
return
queue = tracker.subscribe()
try:
start_msg = {
"type": "connected",
"issue_id": str(issue_id),
"message": "Connected to agent flow stream",
"current_steps": [asdict(s) for s in tracker.flow.steps]
}
yield f"data: {json.dumps(start_msg)}\n\n"
if tracker.flow.status in ["completed", "error"]:
yield f"data: {json.dumps({'type': 'flow_' + tracker.flow.status, 'data': tracker.flow.to_dict()})}\n\n"
return
while True:
try:
message = await asyncio.wait_for(queue.get(), timeout=30)
yield f"data: {json.dumps(message)}\n\n"
if message.get("type") in ["flow_completed", "flow_error"]:
break
except asyncio.TimeoutError:
yield f"data: {json.dumps({'type': 'heartbeat'})}\n\n"
finally:
tracker.unsubscribe(queue)
@router.get("/flow/{issue_id}")
async def stream_agent_flow(issue_id: UUID):
return StreamingResponse(
event_generator(issue_id),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
)
@router.get("/flow/active")
async def list_active_flows():
return {
"active_flows": [
{
"issue_id": str(issue_id),
"status": tracker.flow.status,
"steps_count": len(tracker.flow.steps),
"started_at": tracker.flow.started_at,
}
for issue_id, tracker in _active_flows.items()
]
}
@router.get("/events/{issue_id}")
async def get_issue_events(
issue_id: UUID,
limit: int = Query(50, ge=1, le=200),
db: AsyncSession = Depends(get_db),
):
query = (
select(IssueEvent)
.where(IssueEvent.issue_id == issue_id)
.order_by(IssueEvent.created_at.asc())
.limit(limit)
)
result = await db.execute(query)
events = result.scalars().all()
return {
"issue_id": str(issue_id),
"events": [
{
"id": str(e.id),
"event_type": e.event_type,
"agent_name": e.agent_name,
"event_data": json.loads(e.event_data) if e.event_data else None,
"created_at": e.created_at.isoformat(),
}
for e in events
]
}
@router.get("/timeline/{issue_id}")
async def get_issue_timeline(
issue_id: UUID,
db: AsyncSession = Depends(get_db),
):
issue = await db.get(Issue, issue_id)
if not issue:
return {"error": "Issue not found"}
query = (
select(IssueEvent)
.where(IssueEvent.issue_id == issue_id)
.order_by(IssueEvent.created_at.asc())
)
result = await db.execute(query)
events = result.scalars().all()
timeline = []
timeline.append({
"timestamp": issue.created_at.isoformat(),
"event": "issue_created",
"agent": "System",
"details": {
"latitude": issue.latitude,
"longitude": issue.longitude,
"description": issue.description,
}
})
for event in events:
event_data = json.loads(event.event_data) if event.event_data else {}
timeline.append({
"timestamp": event.created_at.isoformat(),
"event": event.event_type,
"agent": event.agent_name or "Unknown",
"details": event_data,
})
return {
"issue_id": str(issue_id),
"current_state": issue.state,
"priority": issue.priority,
"is_duplicate": issue.is_duplicate,
"timeline": timeline,
}
|