Merge pull request #197 from openqj/feat-stream-waiting

feat: browser stream waits through round gaps instead of disconnecting

cloud/server.py

@@ -5408,9 +5408,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         current holder at least every 5s). The payload also
         carries the session url/title, so the console panel can
         track the header from the stream alone instead of
-        polling the state endpoint while watching. Ends the
-        stream when the state file goes stale, so a dead
-        session reads as a disconnect, not a frozen picture.
+        polling the state endpoint while watching.
+
+        Stale state does NOT end the stream while the task is
+        still alive: between rounds (and across a worker
+        restart) the MCP is gone for a while and the state file
+        ages past the 30s staleness mark, which used to kill
+        every watch at the first round gap. Now a stale state
+        only switches the stream to waiting mode -- a heartbeat
+        with waiting:true every 5s, task status checked at the
+        same cadence -- and frames resume when the next round's
+        MCP starts polling again. The stream ends only when the
+        task reaches a terminal state (completed, merged,
+        failed), the task row disappears, or the state stays
+        stale for 10 minutes straight (abandoned watch).
         """
         import time
         from cloud.playwright_remote import (
@@ -5431,23 +5442,62 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         paths = bridge_paths(worktree)
         last_frame = None
         last_heartbeat = 0.0
+        last_holder = ""
+        stale_since = None
+        last_status_check = 0.0
+        task_status = None
         try:
             while True:
                 state, serr = read_state(worktree)
+                now = time.time()
                 if serr is not None:
-                    break
+                    # No live session right now (round gap,
+                    # worker restart, or a blocked drain loop).
+                    # Wait it out while the task is still alive
+                    # instead of dropping the watch.
+                    if stale_since is None:
+                        stale_since = now
+                    if now - stale_since > 600:
+                        break  # abandoned watch
+                    if now - last_status_check >= 5:
+                        last_status_check = now
+                        try:
+                            row = self._fetch_task(task_id)
+                        except Exception:  # noqa: BLE001 - keep waiting
+                            row = False  # transient error, not deletion
+                        if row is None:
+                            break  # task row gone
+                        if row is not False:
+                            task_status = row["status"]
+                        if task_status in (
+                                "completed", "merged", "failed"):
+                            break
+                    if now - last_heartbeat >= 5:
+                        last_heartbeat = now
+                        payload = json.dumps(
+                            {"image": "",
+                             "holder": last_holder,
+                             "waiting": True,
+                             "task_status": task_status or "",
+                             "ts": int(now)}, ensure_ascii=False)
+                        self.wfile.write(
+                            ("data: %s\n\n" % payload).encode("utf-8"))
+                        self.wfile.flush()
+                    time.sleep(0.02)
+                    continue
+                stale_since = None
                 try:
                     with open(paths["frame"], "rb") as fh:
                         frame = fh.read().decode("utf-8")
                 except (OSError, UnicodeDecodeError):
                     frame = ""
-                now = time.time()
                 if frame != last_frame or now - last_heartbeat >= 5:
                     last_frame = frame
                     last_heartbeat = now
+                    last_holder = effective_holder(worktree, state)
                     payload = json.dumps(
                         {"image": frame,
-                         "holder": effective_holder(worktree, state),
+                         "holder": last_holder,
                          "url": state.get("url", ""),
                          "title": state.get("title", ""),
                          "ts": int(now)}, ensure_ascii=False)
@@ -5457,6 +5507,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 time.sleep(0.02)
         except (BrokenPipeError, ConnectionResetError, OSError):
             pass
+        # An SSE stream has no Content-Length, so the client can
+        # only tell "stream ended" from the connection closing.
+        # HTTP/1.1 keep-alive would otherwise leave it hanging
+        # after a terminal break (completed/failed/cap) with no
+        # more bytes ever coming. Force the close on stream end.
+        self.close_connection = True
 
     def _task_chat_history(self, user_id, task_id):
         """GET /tasks/{id}/chat-history：真正的交替历史，一次性