Merge pull request #39 from mezonai/feature/record_Agent_session

fix bug update status final room

Author: hoangdohuy

Architect_MultiClient_Server/orchestrator_service/api/room_registry_api.py

@@ -73,22 +73,21 @@ async def register_room(
     try:
         registry = get_room_registry()
 
-        # Tạo ISO string timestamp duy nhất tại thời điểm này
+        # create actual_start_time string in ISO format
         if request.start_time is not None:
-            # Nếu được cung cấp, convert float timestamp sang ISO string
-            actual_start_time = datetime.fromtimestamp(request.start_time).isoformat()
+            # if provided, use the given start_time
+            actual_start_time = datetime.fromtimestamp(request.start_time).strftime("%Y%m%d_%H%M%S")
         else:
-            # Nếu không, dùng thời gian hiện tại
-            actual_start_time = datetime.utcnow().isoformat()
-        
-        # Register room vào registry với ISO string
+            # If not provided, use the current time
+            actual_start_time = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+
+        # Register room in registry 
         if not registry.register_room(request.room_name, actual_start_time):
             raise HTTPException(
                 status_code=409,
                 detail=f"Room '{request.room_name}' is already registered"
             )
-        print(f"Room '{request.room_name}' registered with start_time: {registry.get_room_start_time(request.room_name)}")
-        # Lấy tất cả tracks đang có trong room và bắt đầu recording
+        # get all tracks in the room and start recording for audio tracks
         tracks_started = 0
         try:
             livekit_service = get_livekit_service()
@@ -177,12 +176,12 @@ async def unregister_room(
     auth: Dict[str, Any] = Depends(verify_api_key)
 ):
     """
-    Unregister một room khỏi registry.
+    Unregister a room from the registry.
     
-    Sau khi unregister, webhook sẽ không xử lý các events của room này nữa.
-    Đồng thời:
-    - Dừng tất cả egress recordings đang chạy
-    - Finalize room status trong STT service
+    After unregistering, the webhook will no longer process events for this room.
+    Additionally:
+    - Stop all running egress recordings
+    - Finalize room status in the STT service
     
     **Example:**
     ```json
@@ -194,7 +193,7 @@ async def unregister_room(
     try:
         registry = get_room_registry()
 
-        # Lấy start_session_time trước khi unregister (ISO string, luôn có giá trị)
+        # get start_session_time after unregister (ISO string)
         start_session_time = registry.get_room_start_time(request.room_name)
 
         # Unregister room from registry
@@ -254,9 +253,9 @@ async def get_room_status(
     auth: Dict[str, Any] = Depends(verify_api_key)
 ):
     """
-    Kiểm tra trạng thái registration của một room.
+    Check status registration for a room.
     
-    Returns thông tin về room bao gồm start_time và duration nếu room đã được register.
+    Returns information about the room including start_time and duration if the room is registered.
     """
     try:
         registry = get_room_registry()
@@ -284,9 +283,9 @@ async def list_registered_rooms(
     auth: Dict[str, Any] = Depends(verify_api_key)
 ):
     """
-    Lấy danh sách tất cả rooms đang được register.
+    Get a list of all currently registered rooms.
     
-    Returns dictionary với key là room_name và value là start_time.
+    Returns a dictionary with keys as room_name and values as start_time.
     """
     try:
         registry = get_room_registry()
@@ -310,9 +309,9 @@ async def clear_all_rooms(
     auth: Dict[str, Any] = Depends(verify_api_key)
 ):
     """
-    Clear tất cả rooms khỏi registry (dùng cho testing hoặc cleanup).
+    clear all registered rooms from the registry.
     
-    **Cảnh báo**: Action này sẽ xóa tất cả rooms đang được register.
+    **Warning**: This action will delete all currently registered rooms.
     """
     try:
         registry = get_room_registry()

Architect_MultiClient_Server/orchestrator_service/api/stream_message_api.py

@@ -18,7 +18,7 @@ class PushMessageRequest(BaseModel):
 
 @router.post("/push_message")
 async def push_message_api(req: PushMessageRequest):
-    # Kiểm tra có client nào đang listen không
+    # check room has active connections
     if not manager.has_active_connections(req.room_name):
         logger.warning(f"No active connections for room {req.room_name}, message may be lost")
 
@@ -34,25 +34,26 @@ async def push_message_api(req: PushMessageRequest):
 
 async def event_generator(room_name: str, connection_id: str):
     """
-    Generator SSE với:
-    - Timeout để detect client disconnect
-    - Proper cleanup khi connection đóng
-    - Heartbeat để giữ connection alive
+    Generator SSE events for a specific room and connection.
+    Features:
+    - Timeout to detect client disconnect
+    - Proper cleanup when connection closes
+    - Heartbeat to keep connection alive
     """
     logger.info(f"[SSE] Connection started: {connection_id} for room: {room_name}")
     q = manager.get_queue(room_name)
     loop = asyncio.get_event_loop()
 
-    # Gửi event đầu tiên để confirm connection
+    # Send initial event to confirm connection
     yield f"event: connected\ndata: {connection_id}\n\n"
 
-    heartbeat_interval = 15  # Gửi heartbeat mỗi 15 giây
+    heartbeat_interval = 15  # Send heartbeat every 15 seconds
     last_heartbeat = asyncio.get_event_loop().time()
 
     try:
         while True:
             try:
-                # Sử dụng timeout để có thể check connection và gửi heartbeat
+                # Use timeout to check connection and send heartbeat
                 text = await asyncio.wait_for(
                     loop.run_in_executor(None, lambda: q.get(timeout=1.0)),
                     timeout=2.0
@@ -61,7 +62,7 @@ async def event_generator(room_name: str, connection_id: str):
                 yield f"data: {text}\n\n"
 
             except (asyncio.TimeoutError, queue.Empty):
-                # Không có message mới - kiểm tra có cần gửi heartbeat không
+                # No new message - check if heartbeat needs to be sent
                 current_time = asyncio.get_event_loop().time()
                 if current_time - last_heartbeat >= heartbeat_interval:
                     yield f"event: heartbeat\ndata: ping\n\n"
@@ -76,7 +77,7 @@ async def event_generator(room_name: str, connection_id: str):
     except GeneratorExit:
         logger.info(f"[SSE] Generator exit: {connection_id}")
     finally:
-        # Cleanup khi connection đóng
+        # Cleanup when connection closes
         manager.unregister_connection(room_name, connection_id)
         logger.info(f"[SSE] Connection closed and unregistered: {connection_id}")
 

Architect_MultiClient_Server/orchestrator_service/api/stream_message_manager.py

@@ -21,14 +21,14 @@ def __init__(self):
         self._connection_counter = 0
 
     def get_queue(self, room_id: str) -> queue.Queue:
-        """Lấy queue cho room, tạo mới nếu chưa có"""
+        """Get queue for room, and create if not exists"""
         with self._lock:
             if room_id not in self.queues:
                 self.queues[room_id] = queue.Queue()
             return self.queues[room_id]
 
     def register_connection(self, room_id: str) -> str:
-        """Đăng ký một connection mới, trả về connection_id"""
+        """Register a new connection, return connection_id"""
         with self._lock:
             self._connection_counter += 1
             connection_id = f"{room_id}_{self._connection_counter}_{int(time.time())}"
@@ -40,16 +40,16 @@ def register_connection(self, room_id: str) -> str:
             return connection_id
 
     def unregister_connection(self, room_id: str, connection_id: str):
-        """Hủy đăng ký connection khi client disconnect"""
+        """Unregister connection when client disconnects"""
         with self._lock:
             if room_id in self.connections:
                 self.connections[room_id].discard(connection_id)
 
-                # Nếu không còn connection nào, xóa queue để tránh memory leak
+                # If no more connections, delete queue to avoid memory leak
                 if not self.connections[room_id]:
                     del self.connections[room_id]
                     if room_id in self.queues:
-                        # Clear queue trước khi xóa
+                        # Clear queue before deleting
                         q = self.queues[room_id]
                         while not q.empty():
                             try:
@@ -59,12 +59,12 @@ def unregister_connection(self, room_id: str, connection_id: str):
                         del self.queues[room_id]
 
     def has_active_connections(self, room_id: str) -> bool:
-        """Kiểm tra room có connection active không"""
+        """Check if room has active connections"""
         with self._lock:
             return room_id in self.connections and len(self.connections[room_id]) > 0
 
     def get_connection_count(self, room_id: str) -> int:
-        """Lấy số connection active của room"""
+        """Get number of active connections for room"""
         with self._lock:
             return len(self.connections.get(room_id, set()))
 

Architect_MultiClient_Server/orchestrator_service/controller/webhook_handler.py

@@ -126,10 +126,6 @@ async def _handle_room_finished(self, event: Dict) -> WebhookResponse:
         room_name = event.get("room", {}).get("name", "unknown")
         logger.info(f"  Room finished: {room_name}")
 
-        # Get start_session_time from registry (ISO string)
-        start_session_time = self.room_registry.get_room_start_time(room_name)
-        
-        await self.transcription_service.final_room(room_name, start_session_time)
         return WebhookResponse(received=True, action="room_finished_logged")
 
     async def _handle_egress_ended(self, event: Dict) -> WebhookResponse:
@@ -165,7 +161,7 @@ def _build_egress_info(self, egress: Dict, file_data: Dict,
         """Build EgressInfo object from event data"""
         egress_data = {
             "egressId": egress.get("egressId", "unknown"),
-            "room": {"name": egress.get("roomName", ""), "start_session_time": parsed.get("timestamp_iso")},
+            "room": {"name": egress.get("roomName", ""), "start_session_time": parsed.get("timestamp")},
             "participant": {"identity": parsed.get("identity", "unknown")},
             "track": {
                 "id": egress.get("track", {}).get("trackId", "unknown"),

Architect_MultiClient_Server/orchestrator_service/services/egress_service.py

@@ -29,15 +29,8 @@ def _build_filepath(self, room_name: str, identity: str, source: str,
         ext = "ogg" if track_type == "AUDIO" else "webm"
 
         # Parse ISO string và format lại thành YYYYmmdd_HHMMSS
-        try:
-            dt = datetime.fromisoformat(room_start_time)
-            timestamp = dt.strftime("%Y%m%d_%H%M%S")
-        except (ValueError, AttributeError):
-            # Fallback nếu parse thất bại
-            logger.warning(f"Failed to parse room_start_time: {room_start_time}, using current time")
-            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
 
-        return f"{room_name}/{identity}-{source}-{track_type.lower()}-{timestamp}.{ext}"
+        return f"{room_name}/{identity}-{source}-{track_type.lower()}-{room_start_time}.{ext}"
 
     def _get_s3_upload(self) -> api.S3Upload:
         if self._s3_upload is None:

Architect_MultiClient_Server/orchestrator_service/services/transcription_service.py

@@ -47,13 +47,13 @@ async def enqueue(self, egress_info: Dict) -> bool:
             logger.error(f"✗ Error sending to queue: {e}")
             return False
 
-    async def final_room(self, room_name: str, start_session_time: str = None) -> bool:
+    async def final_room(self, room_name: str, start_session_time: str ) -> bool:
         """
         Notify transcription service to finalize room
         
         Args:
             room_name: Name of the room to finalize
-            start_session_time: Optional start session time
+            start_session_time: start session time
             
         Returns:
             True if successful, False if failed

Architect_MultiClient_Server/orchestrator_service/utils/filepath_parser.py

@@ -32,12 +32,4 @@ def parse(cls, filepath: str) -> Dict[str, str]:
 
         result = match.groupdict()
 
-        # Convert timestamp from "YYYYmmdd_HHMMSS" to ISO format "YYYY-MM-DDTHH:MM:SS"
-        try:
-            timestamp_str = result.get("timestamp", "")
-            dt = datetime.strptime(timestamp_str, "%Y%m%d_%H%M%S")
-            result["timestamp_iso"] = dt.isoformat()
-        except (ValueError, AttributeError):
-            result["timestamp_iso"] = None
-        
         return result

Architect_MultiClient_Server/stt_service/controller/transcription_api.py

@@ -25,7 +25,7 @@
 
 class RoomInfo(BaseModel):
     name: str
-    start_session_time: Optional[str]
+    start_session_time: str
 
 class ParticipantInfo(BaseModel):
     identity: str

Architect_MultiClient_Server/stt_service/service/mongodb_service.py

@@ -345,6 +345,12 @@ async def get_room_by_name(self, room_name: str) -> Optional[Dict]:
             return None
         return await self.rooms_collection.find_one({"room_name": room_name})
 
+    async def get_room_session_by_name(self, room_name: str, start_session_time: str) -> Optional[Dict]:
+        """Get room by name"""
+        if not self.connected:
+            return None
+        return await self.rooms_collection.find_one({"room_name": room_name, "start_session_time": start_session_time})
+
     async def update_room_status(
         self,
         room_ref_id: ObjectId,
@@ -370,18 +376,18 @@ async def update_room_status(
             logger.error(f"Failed to update room status: {e}")
             return False
 
-    async def final_room_status(self, room_name: str, start_session_time: Optional[str]) -> bool:
+    async def final_room_status(self, room_name: str, start_session_time: str) -> bool:
 
         if not self.connected:
             return False
 
         try:
-            room = await self.get_room_by_name(room_name)
+            room = await self.get_room_session_by_name(room_name, start_session_time)
             if not room:
                 logger.error(f"Room not found: {room_name}")
                 return False
 
-            # Chỉ update nếu chưa finalized
+            # only update when status not finalized
             if room["status"] in ["final_room", "completed"]:
                 logger.warning(f"Room already finalized: {room_name}")
                 return True
@@ -394,11 +400,6 @@ async def final_room_status(self, room_name: str, start_session_time: Optional[s
                 }
             }
 
-            # Chỉ update start_session_time nếu được cung cấp VÀ room chưa có
-            if start_session_time and not room.get("start_session_time"):
-                update_doc["$set"]["start_session_time"] = start_session_time
-                logger.info(f"Setting start_session_time for room {room_name}: {start_session_time}")
-            
             await self.rooms_collection.update_one(
                 {"_id": room["_id"]},
                 update_doc