feat: add viewer support for shell sessions and enhance file harvest logic

- Introduced `viewer` role in `v1.py` for read-only WebSocket connections.
- Improved file harvest logic to scan multiple directories and broadcast completion events.
- Updated sandbox warning handling and refined cleanup of uploaded files.

Author: Francisco

src/api/entities_api/orchestration/mixins/shell_execution_mixin.py

@@ -217,8 +217,13 @@ async def handle_shell_action(
                     }
                 )
 
+                # Filter out known benign sandbox startup warnings
+                chunk_lower = chunk.lower()
+                eval_chunk = chunk_lower.replace("bash: /root/.bashrc: permission denied", "")
+                eval_chunk = eval_chunk.replace("warning: an existing sandbox was detected", "")
+
                 if any(
-                    marker in chunk.lower()
+                    marker in eval_chunk
                     for marker in [
                         "not found",
                         "permission denied",

src/api/sandbox/routers/v1.py

@@ -118,6 +118,7 @@ async def websocket_endpoint(
     room: str = Query(..., description="The Thread/Room ID"),
     elevated: bool = Query(False),
     token: str = Query(..., description="Signed JWT from Main API"),
+    role: str = Query("executor", description="Role of the client (viewer or executor)"),
 ):
     """
     SECURED: Interactive Shell Session.
@@ -133,7 +134,6 @@ async def websocket_endpoint(
         return
 
     # 2. Validate Room Access (Multi-Tenancy Security)
-    # The token MUST contain a "room" claim matching the requested room
     allowed_room = payload.get("room")
     user_id = payload.get("sub")
 
@@ -144,8 +144,20 @@ async def websocket_endpoint(
         await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
         return
 
-    logging_utility.info(f"Shell session started for User: {user_id} in Room: {room}")
+    # 3. Handle Frontend UI (Viewer)
+    if role == "viewer":
+        logging_utility.info(f"Viewer UI connected to Room: {room}")
+        await room_manager.connect(room, websocket)
+        try:
+            # Keep the connection alive and listen for the client disconnecting
+            while True:
+                await websocket.receive_text()
+        except WebSocketDisconnect:
+            await room_manager.disconnect(room, websocket)
+            logging_utility.info(f"Viewer UI disconnected from Room: {room}")
+        return
 
-    # 3. Start Session
+    # 4. Handle AI SDK (Executor)
+    logging_utility.info(f"Shell session started for User: {user_id} in Room: {room}")
     session = PersistentShellSession(websocket, room, room_manager, elevated=elevated)
     await session.start()

src/api/sandbox/services/shell_session.py

@@ -272,14 +272,30 @@ async def _message_loop(self) -> None:
                 await self.websocket.send_json({"type": "pong"})
 
             elif action == "harvest_files":
+
                 # Explicit mid-session harvest requested by the assistant.
+
                 # Does NOT terminate the session — shell stays alive.
+
                 self._reset_idle_timer()
+
                 await self._harvest_and_upload_files(
                     context="explicit_harvest",
                     wipe_after=True,
                 )
 
+                # FIX: Broadcast command_complete so the SDK knows the harvest is done!
+
+                asyncio.create_task(
+                    self.room_manager.broadcast(
+                        self.room,
+                        {
+                            "type": "command_complete",
+                            "thread_id": self.room,
+                        },
+                    )
+                )
+
             elif action == "disconnect":
                 break
 
@@ -429,25 +445,38 @@ async def _harvest_and_upload_files(
             Delete successfully uploaded files after upload.  Always True in
             normal operation; False only for debugging.
         """
-        if not os.path.isdir(self.session_dir):
-            return
+        # 1. Scan BOTH the room's session directory AND the global generated_files directory
+        directories_to_check = [self.session_dir, "/app/generated_files"]
 
-        # Collect harvestable files, skipping oversized ones
         candidates: list[tuple[str, str]] = []
-        for fname in os.listdir(self.session_dir):
-            fpath = os.path.join(self.session_dir, fname)
-            if not os.path.isfile(fpath):
-                continue
-            size_mb = os.path.getsize(fpath) / (1024 * 1024)
-            if size_mb > MAX_HARVEST_FILE_SIZE_MB:
-                logger.warning(
-                    "Skipping %s (%.1f MB exceeds %d MB limit)",
-                    fname,
-                    size_mb,
-                    MAX_HARVEST_FILE_SIZE_MB,
-                )
+        seen_paths = set()
+
+        for target_dir in directories_to_check:
+            if not os.path.isdir(target_dir):
                 continue
-            candidates.append((fname, fpath))
+
+            for fname in os.listdir(target_dir):
+                fpath = os.path.join(target_dir, fname)
+
+                # Prevent double-processing just in case paths overlap
+                abs_path = os.path.abspath(fpath)
+                if abs_path in seen_paths:
+                    continue
+                seen_paths.add(abs_path)
+
+                if not os.path.isfile(fpath):
+                    continue
+
+                size_mb = os.path.getsize(fpath) / (1024 * 1024)
+                if size_mb > MAX_HARVEST_FILE_SIZE_MB:
+                    logger.warning(
+                        "Skipping %s (%.1f MB exceeds %d MB limit)",
+                        fname,
+                        size_mb,
+                        MAX_HARVEST_FILE_SIZE_MB,
+                    )
+                    continue
+                candidates.append((fname, fpath))
 
         if not candidates:
             logger.info("Harvest [%s] room=%s: no files found.", context, self.room)
@@ -534,6 +563,7 @@ async def _upload_one(fname: str, fpath: str) -> Optional[dict]:
         if wipe_after:
             for fpath in uploaded_paths:
                 try:
+                    # We only delete the specific files we successfully uploaded!
                     os.remove(fpath)
                 except Exception:
                     pass