fix: abort sets should_stop before close; llmcore skip retry on user stop; hub serve-on-connect + stapp hub optional
This commit is contained in:
1 parent
41e1fcebe8
commit
d8d90eef8c
4 files changed
+21
-3
No files matched your search
+2
-1
@@ -113,10 +113,11 @@ class GenericAgent:
|
||||
if not self.is_running: return
|
||||
print('Abort current task...')
|
||||
self.stop_sig = True
|
||||
if self.handler is not None: self.handler.code_stop_signal.append(1)
|
||||
for sess in getattr(self.llmclient.backend, '_sessions', [self.llmclient.backend]):
|
||||
sess.should_stop = lambda: self.stop_sig # live read; cleared by run()'s finally
|
||||
try: sess.active_response.close()
|
||||
except Exception: pass
|
||||
if self.handler is not None: self.handler.code_stop_signal.append(1)
|
||||
|
||||
def put_task(self, query, source="user", images=None):
|
||||
display_queue = queue.Queue()
|
||||
|
||||
@@ -64,6 +64,19 @@ class HubClient:
|
||||
elif op == 'abort': data = (await asyncio.to_thread(self.abort) or {'ok': 1}) if self.abort else {'error': 'no abort hook', 'code': 'nosupport'}
|
||||
await ws.send(json.dumps({'op': 'r', 'id': c.get('id'), 'name': self.name, 'data': data}, default=str))
|
||||
|
||||
def serve():
|
||||
"""Bring the hub up on demand: any host may spawn it, the port is the lock (a loser just exits).
|
||||
Detached + windowless, so it outlives its spawner and no console ever flashes."""
|
||||
import socket, subprocess
|
||||
if '127.0.0.1' not in URL and 'localhost' not in URL: return # a remote hub is not ours to start
|
||||
with socket.socket() as s:
|
||||
if s.connect_ex(('127.0.0.1', PORT)) == 0: return # already listening
|
||||
exe = os.path.join(os.path.dirname(sys.executable), 'pythonw.exe')
|
||||
subprocess.Popen([exe if os.path.exists(exe) else sys.executable, os.path.abspath(__file__)],
|
||||
cwd=os.path.dirname(os.path.abspath(__file__)), close_fds=True,
|
||||
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
creationflags=0x08000008 if os.name == 'nt' else 0) # DETACHED | NO_WINDOW
|
||||
|
||||
def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=None):
|
||||
"""One line to wire a GA host: `hub.connect(agent, 'stapp')`; any hook can still be overridden.
|
||||
Default put refuses while the agent is busy (a remote must not cut in line), else it parks
|
||||
@@ -73,6 +86,8 @@ def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=
|
||||
if getattr(agent, 'is_running', False): return {'error': f'peer {name} is busy', 'code': 'busy'}
|
||||
agent._hub_inbox.append({'text': text, 'q': agent.put_task(text, source='hub')})
|
||||
try:
|
||||
try: serve() # best effort: bring up a local hub if none is listening
|
||||
except Exception: pass
|
||||
if not hasattr(agent, '_hub_inbox'): agent._hub_inbox = []
|
||||
return HubClient(name or getattr(agent, 'name', 'agent'), put_task or _put,
|
||||
get_outputs or (lambda: agent.all_outputs), abort or agent.abort).start()
|
||||
|
||||
+3
-2
@@ -360,8 +360,9 @@ def mount_main_stream():
|
||||
if not hasattr(agent, "_ui_messages"): agent._ui_messages = st.session_state.get("messages", [])
|
||||
if "messages" not in st.session_state: st.session_state.messages = agent._ui_messages
|
||||
if not hasattr(agent, "_hub"):
|
||||
import hub
|
||||
agent._hub = hub.connect(agent, 'stapp') # busy-rejects; accepted tasks land in agent._hub_inbox
|
||||
try:
|
||||
import hub; agent._hub = hub.connect(agent, 'stapp')
|
||||
except Exception: agent._hub = None
|
||||
# Lazy history: long sessions (esp. after loop) render thousands of elements on every
|
||||
# full-app rerun → seconds of gray/RUNNING. Only render the tail unless expanded.
|
||||
_HIST_TAIL = 10
|
||||
|
||||
@@ -454,6 +454,7 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn):
|
||||
return e.value or []
|
||||
except (requests.Timeout, requests.ConnectionError, requests.exceptions.ChunkedEncodingError) as e:
|
||||
err = f"!!!Error: {type(e).__name__}: {e}" if str(e) else f"!!!Error: {type(e).__name__}"
|
||||
if getattr(sess, 'should_stop', None) and sess.should_stop(): return []
|
||||
if attempt < sess.max_retries:
|
||||
d = _delay(None, attempt)
|
||||
print(f"[LLM Retry] {type(e).__name__}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})")
|
||||
|
||||
Reference in new issue
Block a user