fix(voice): keep watcher alive through quiet checks and reconnects

apps/desktop/assets/codex-bridge/bekci_watch.py

@@ -124,48 +124,70 @@ class WatchOwner:
         binding = Binding(opaque('member',key),scope,opaque('history-receipt',p['subscription']),c.connection_id,c.thread_id,c.generation)
         deadline = p['expires_at']/1000
         try:
-            for attempt in range(8):
-                if stop.is_set() or time.time() >= deadline or not self.live(c): break
+            backoff = 1
+            while not stop.is_set() and time.time() < deadline and self.live(c):
                 conn = (http.client.HTTPSConnection(self.origin.hostname,self.origin.port,context=ssl.create_default_context(),timeout=10)
                         if self.origin.scheme == 'https' else http.client.HTTPConnection(self.origin.hostname,self.origin.port,timeout=10))
                 finished = threading.Event()
+                transport, resp, guard_thread = None, None, None
+                def interrupt_transport():
+                    # getresponse may detach conn.sock for HTTP/1.0 or
+                    # Connection: close. Keep our actual socket until cleanup.
+                    if transport is not None:
+                        try: transport.shutdown(socket.SHUT_RDWR)
+                        except OSError: pass
                 def guard():
                     while not finished.wait(.05):
                         if stop.is_set() or time.time() >= deadline or not self.live(c):
                             stop.set()
-                            if conn.sock:
-                                try: conn.sock.shutdown(socket.SHUT_RDWR)
-                                except OSError: pass
-                            conn.close()
+                            interrupt_transport()
                             return
-                guard_thread = threading.Thread(target=guard,daemon=True)
-                guard_thread.start()
                 try:
+                    # Bound TCP/TLS connection establishment separately from
+                    # headers/stream reads: quiet research has no idle deadline.
+                    conn.connect()
+                    transport = conn.sock
+                    transport.settimeout(None)
+                    if stop.is_set() or time.time() >= deadline or not self.live(c): break
+                    guard_thread = threading.Thread(target=guard,daemon=True)
+                    guard_thread.start()
                     body = {'subscription':p['subscription']}
                     cursor = self.ledger.cursor(scope)
                     if cursor: body['cursor'] = cursor
                     conn.request('POST', EVENTS_PATH, json.dumps(body), {'Authorization':'Bearer '+key,'Content-Type':'application/json'})
                     resp = conn.getresponse()
-                    if resp.status != 200 or 'text/event-stream' not in resp.getheader('Content-Type',''): break
-                    parser, ready = Frames(), False
-                    while self.live(c) and not stop.is_set() and time.time() < deadline:
-                        chunk = resp.read1(4096)
-                        if not chunk: break
-                        for event,data,event_id,_ in parser.feed(chunk):
-                            if not data: continue # keepalive
-                            value = json.loads(data)
-                            if event == 'ready':
-                                if not isinstance(value,dict) or not isinstance(value.get('cursor'),str): raise ValueError('invalid ready')
-                                ready = self.live(c) and self.mux.bind(binding)
-                                if not ready: return
-                            elif event == 'notice':
-                                if not ready: raise ValueError('notice before ready')
-                                self._notice(binding,scope,value,event_id,deadline,c)
+                    if resp.status != 200:
+                        if resp.status not in (408,429) and not 500 <= resp.status < 600: break
+                    elif 'text/event-stream' not in resp.getheader('Content-Type',''):
+                        break
+                    else:
+                        parser, ready = Frames(), False
+                        while self.live(c) and not stop.is_set() and time.time() < deadline:
+                            chunk = resp.read1(4096)
+                            if not chunk: break
+                            for event,data,event_id,_ in parser.feed(chunk):
+                                if stop.is_set() or time.time() >= deadline or not self.live(c): break
+                                if not data: continue # keepalive
+                                value = json.loads(data)
+                                if event == 'ready':
+                                    if not isinstance(value,dict) or not isinstance(value.get('cursor'),str): raise ValueError('invalid ready')
+                                    ready = self.live(c) and self.mux.bind(binding)
+                                    if not ready: return
+                                elif event == 'notice':
+                                    if not ready: raise ValueError('notice before ready')
+                                    self._notice(binding,scope,value,event_id,deadline,c)
                 except (OSError,ValueError,http.client.HTTPException):
                     pass # no credential-bearing exceptions in logs
                 finally:
-                    finished.set(); conn.close(); guard_thread.join(timeout=1)
-                if stop.wait(min(2**attempt,15)): break
+                    finished.set()
+                    interrupt_transport()
+                    if guard_thread is not None: guard_thread.join()
+                    if resp is not None: resp.close()
+                    conn.close()
+                    if transport is not None: transport.close()
+                if stop.is_set() or time.time() >= deadline or not self.live(c): break
+                if stop.wait(min(backoff,max(0,deadline-time.time()))): break
+                backoff = min(backoff*2,15)
         finally:
             with self.mux._lock:
                 k = (binding.tenant_id,binding.bekci_session_id)

apps/desktop/assets/codex-bridge/test_bekci_watch.py

@@ -256,3 +256,102 @@ def test_actual_handler_nonvoice_preserves_response_pipeline(tmp_path, monkeypat
         wrapped.assert_not_called();shortcut.assert_not_called();owner.accept.assert_not_called()
     finally:
         server.shutdown();server.server_close();upstream.shutdown();upstream.server_close()
+
+
+class QuietWatchServer:
+    """Real sockets, with no heartbeat; observes client shutdown at the peer."""
+    def __init__(self, protocol='HTTP/1.0', delay=0, headers_delay=False, reconnects=False, delayed_headers=False):
+        self.requests=[];self.closed=threading.Event();self.peer_closed=threading.Event()
+        self.ready=threading.Event();owner=self
+        class H(BaseHTTPRequestHandler):
+            protocol_version='HTTP/1.1' if protocol=='HTTP/1.1-close' else protocol
+            def log_message(self,*args): pass
+            def do_POST(self):
+                body=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+                owner.requests.append(body);number=len(owner.requests)
+                try:
+                    if headers_delay:
+                        owner.ready.set()
+                        self.connection.settimeout(5)
+                        if self.connection.recv(1)==b'': owner.peer_closed.set()
+                        return
+                    if delayed_headers and owner.closed.wait(delay): return
+                    self.send_response(200);self.send_header('Content-Type','text/event-stream')
+                    if protocol=='HTTP/1.1-close': self.send_header('Connection','close')
+                    self.end_headers()
+                    self.wfile.write(frame('ready',{'cursor':'start'}));self.wfile.flush();owner.ready.set()
+                    if reconnects:
+                        n=notice('first' if number<9 else 'ninth')
+                        self.wfile.write(frame('notice',n,n['cursor']));self.wfile.flush()
+                        if number<9:
+                            self.close_connection=True
+                            return
+                    elif delay and number==1:
+                        if not delayed_headers and owner.closed.wait(delay): return
+                        n=notice('delayed');self.wfile.write(frame('notice',n,n['cursor']));self.wfile.flush()
+                    self.connection.settimeout(5 if not delay else 20)
+                    if self.connection.recv(1)==b'': owner.peer_closed.set()
+                except OSError: pass
+        self.server=ThreadingHTTPServer(('127.0.0.1',0),H);self.server.daemon_threads=True
+        self.origin='http://127.0.0.1:'+str(self.server.server_port)
+        threading.Thread(target=self.server.serve_forever,daemon=True).start()
+    def close(self):
+        self.closed.set();self.server.shutdown();self.server.server_close()
+
+
+@pytest.mark.parametrize('delayed_headers',[False,True])
+def test_quiet_watch_delivers_after_more_than_ten_seconds(tmp_path,delayed_headers):
+    server=QuietWatchServer(delay=11.2,delayed_headers=delayed_headers)
+    mux=NativeSpeechMux(['unused'],observer=True);started(mux,'A')
+    owner=WatchOwner(mux,Ledger(tmp_path/'quiet'),origin=server.origin,test_loopback=True)
+    try:
+        assert owner.accept(meta(),capture(owner,'A'),'member',server.origin)=='subscribing'
+        assert server.ready.wait(14)
+        until=time.monotonic()+14
+        while mux._queue.empty() and time.monotonic()<until: time.sleep(.02)
+        assert not mux._queue.empty(), 'valid quiet stream lost delayed notice'
+        assert len(server.requests)==1, 'quiet channel spuriously reconnected'
+    finally:owner.close();server.close()
+
+
+@pytest.mark.parametrize('protocol',['HTTP/1.0','HTTP/1.1','HTTP/1.1-close'])
+@pytest.mark.parametrize('cancel',['close','expiry','generation'])
+@pytest.mark.parametrize('headers_delay',[False,True])
+def test_quiet_owned_socket_is_interrupted(tmp_path,protocol,cancel,headers_delay):
+    server=QuietWatchServer(protocol=protocol,headers_delay=headers_delay)
+    mux=NativeSpeechMux(['unused'],observer=True);started(mux,'A')
+    owner=WatchOwner(mux,Ledger(tmp_path/'cancel'),origin=server.origin,test_loopback=True)
+    try:
+        p=meta()
+        if cancel=='expiry':p['expires_at']=(time.time()+.7)*1000
+        owner.accept(p,capture(owner,'A'),'member',server.origin)
+        assert server.ready.wait(3)
+        if not headers_delay:wait(lambda:bool(mux._bindings))
+        if cancel=='close':owner.close()
+        elif cancel=='generation':mux._observe_client({'method':'thread/realtime/stop','params':{'threadId':'A'}})
+        assert server.peer_closed.wait(2), 'owned response/header socket remained open'
+        wait(lambda:not owner.watches)
+        assert not mux._bindings and mux._queue.empty()
+    finally:owner.close();server.close()
+
+
+def test_ninth_reconnect_delivers_and_preserves_cursor_dedup(tmp_path):
+    class FastBackoff:
+        def __init__(self):self.event=threading.Event()
+        def is_set(self):return self.event.is_set()
+        def set(self):self.event.set()
+        def wait(self,seconds):return self.event.wait(min(seconds,.01))
+    server=QuietWatchServer(reconnects=True)
+    mux=NativeSpeechMux(['unused'],observer=True);started(mux,'A')
+    owner=WatchOwner(mux,Ledger(tmp_path/'reconnect'),origin=server.origin,test_loopback=True)
+    p=meta();c=capture(owner,'A');scope=opaque('public-watch','member',p['subscription']);stop=FastBackoff()
+    owner.watches[scope]=stop
+    worker=threading.Thread(target=owner._watch,args=(p,c,'member',scope,stop),daemon=True);worker.start()
+    try:
+        wait(lambda:len(server.requests)>=9)
+        wait(lambda:owner.ledger.cursor(scope)=='cursor-ninth')
+        assert mux._queue.qsize()==2
+        assert all(req.get('cursor')=='cursor-first' for req in server.requests[1:9])
+    finally:
+        owner.close();worker.join(timeout=3);server.close()
+    assert not worker.is_alive()