diff --git a/agentmain.py b/agentmain.py index 6f04ccef..2debf2ec 100644 --- a/agentmain.py +++ b/agentmain.py @@ -47,9 +47,12 @@ class GenericAgent: os.makedirs(os.path.join(script_dir, 'temp'), exist_ok=True) self.lock = threading.Lock() self.task_dir = None - self.history = []; self.handler = None; + self.history = []; self.handler = None; self.all_outputs = [] self.task_queue = queue.Queue() - self.is_running = False; self.stop_sig = False; self.llm_no = 0; + self.is_running = False; self.stop_sig = False; self.llm_no = 0; + # Output queue of the task currently executing (None when idle). Lets a UI that + # lost its own handle (page refresh, second client) re-attach to the live task. + self._current_queue = None self.inc_out = False; self.verbose = True self.peer_hint = True self.force_non_stream = False @@ -144,11 +147,13 @@ class GenericAgent: raw_query = self._handle_slash_cmd(raw_query, display_queue) if raw_query is None: self.task_queue.task_done(); continue - self.is_running = True + self.is_running = True; self._current_queue = display_queue if len(raw_query) > 2000: task_file = os.path.join(script_dir, 'temp', f'user_prompt_{os.getpid()}_{time.time_ns()}.md') with open(task_file, 'w', encoding='utf-8') as f: f.write(raw_query) raw_query = f'Long user prompt saved to {task_file}. Read and execute.' + self.all_outputs.append({"input": raw_query, "outputs": []}) + if len(self.all_outputs) > 10000: self.all_outputs = self.all_outputs[-5000:] rquery = smart_format(raw_query.replace('\n', ' '), max_str_len=200) self.history.append(f"[USER]: {rquery}") sys_prompt = get_system_prompt() + '\n'.join(self.extra_sys_prompts) + getattr(self.llmclient.backend, 'extra_sys_prompt', '') @@ -168,7 +173,7 @@ class GenericAgent: gen = agent_runner_loop(self.llmclient, sys_prompt, raw_query, handler, TOOLS_SCHEMA, max_turns=180, verbose=self.verbose, yield_info=True) try: - full_resp = ""; last_pos = 0; curr_turn = 0; turn_resps = [] + full_resp = ""; last_pos = 0; curr_turn = 0; turn_resps = self.all_outputs[-1]["outputs"] for chunk in gen: if consume_file(self.task_dir, '_stop'): self.abort() if self.stop_sig: break @@ -189,7 +194,7 @@ class GenericAgent: display_queue.put({'done': full_resp + f'\n```\n{format_error(e)}\n```', 'source': source, 'turn': curr_turn, 'outputs': turn_resps.copy()}) finally: if self.stop_sig: print('User aborted the task.') - self.is_running = self.stop_sig = False + self.is_running = self.stop_sig = False # keep _current_queue: its final 'done' may still be unclaimed (refreshed UI salvages it); next task overwrites it self.task_queue.task_done() if self.handler is not None: self.handler.code_stop_signal.append(1) diff --git a/frontends/conductor.html b/frontends/conductor.html index b3a19520..5e1621a1 100644 --- a/frontends/conductor.html +++ b/frontends/conductor.html @@ -61,9 +61,9 @@ overflow: hidden; } - .left-col { min-height: 0; display: flex; flex-direction: column; gap: 18px; } + .left-col { min-height: 0; min-width: 0; display: flex; flex-direction: column; gap: 18px; } .left-col > .pane.left { flex: 1; min-height: 0; } - .left, .right { display: flex; flex-direction: column; min-height: 0; } + .left, .right { display: flex; flex-direction: column; min-height: 0; min-width: 0; } .right { background: transparent; border: 0; box-shadow: none; backdrop-filter: none; gap: 18px; } .head { @@ -200,10 +200,15 @@ .btn-no { background: linear-gradient(135deg, #f87171, #ef4444); box-shadow: 0 8px 20px rgba(239,68,68,.22); } /* ---------- 聊天框 ---------- */ - .chat { flex: 1; min-height: 0; display: flex; flex-direction: column; } - .messages { flex: 1; overflow: auto; padding: 20px; display: flex; flex-direction: column; gap: 13px; } + .chat { flex: 1; min-height: 0; min-width: 0; display: flex; flex-direction: column; } + .messages { flex: 1; min-width: 0; overflow: auto; padding: 20px; display: flex; flex-direction: column; gap: 13px; } .msg { max-width: 80%; padding: 13px 15px; border-radius: 18px; line-height: 1.45; font-size: 14px; - word-wrap: break-word; box-shadow: 0 8px 22px rgba(30,45,80,.055); } + overflow-wrap: anywhere; box-shadow: 0 8px 22px rgba(30,45,80,.055); } + .msg :is(p, ul, ol, pre, table, img) { max-width: 100%; } + .msg pre { white-space: pre-wrap; overflow-wrap: anywhere; } + .msg code { overflow-wrap: anywhere; } + .msg table { display: block; overflow-x: auto; } + .msg img { height: auto; } .msg p { margin: 0 0 .3em 0; } .msg p:last-child { margin-bottom: 0; } .msg.me { align-self: flex-end; color: white; background: linear-gradient(135deg, #3b82f6, #7c3aed); border: 1px solid rgba(59,130,246,.20); diff --git a/frontends/conductor.py b/frontends/conductor.py index 627e6a6a..9202ce7d 100644 --- a/frontends/conductor.py +++ b/frontends/conductor.py @@ -234,9 +234,12 @@ def add_chat(msg: str, role: str = "conductor", files: list = None, images: list files = list(files or []) known = {x.get("path") for x in files} for ref in re.findall(r"`([^`\n]+)`", msg): - try: path = file_path(ref) - except ValueError: continue - if os.path.isfile(path) and path not in known: files.append({"ref": ref, "path": path, "name": os.path.basename(path)}) + for candidate in ([ref] if os.path.isabs(ref) else [os.path.join(FILE_HOME, ref), ref]): + try: path = file_path(candidate) + except ValueError: continue + if os.path.isfile(path): + if path not in known: files.append({"ref": ref, "path": path, "name": os.path.basename(path)}) + break item = {"id": short_id(), "role": role, "msg": msg, "ts": now_ms(), "read": role != "user", "files": files, "images": images or []} chat_messages.append(item) if len(chat_messages) > 200: del chat_messages[:-200] diff --git a/frontends/hub.html b/frontends/hub.html new file mode 100644 index 00000000..bc9298d2 --- /dev/null +++ b/frontends/hub.html @@ -0,0 +1,269 @@ + + + + +GA Hub + + + + +
+

GA Hub

+
+ +
+ + + + + diff --git a/frontends/hub.py b/frontends/hub.py new file mode 100644 index 00000000..d50b926d --- /dev/null +++ b/frontends/hub.py @@ -0,0 +1,150 @@ +"""GA Hub: `import hub` -> client (silent if no server); `python hub.py` -> WS server. + hub.connect(agent, 'stapp') # or override any hook: connect(a, n, put_task=, get_outputs=, abort=) + default put -> 'busy' if agent.is_running, else parks {'text','q'} in agent._hub_inbox for the UI +HTTP (errors = {'error','code'} + status: offline/gone 404, busy 409, timeout 504, nosupport 501, badop 400): + GET peers -> [{name,title,n_msgs,sig}] | {name}/messages?detail=1&sig= -> {title,tasks:[{i,input,steps: + [{j,title,n}]}],sig} or {same:1,sig} | {name}/seg/{i}/{j}?off=N -> {content,off,n} (step bodies, tailable) + POST {name}/put {"text":..} -> {ok:1} | {name}/abort -> {ok:1} +""" +import os, re, sys, json, time, asyncio, threading +PORT = int(os.environ.get('GA_HUB_PORT', 19736)) +URL = os.environ.get('GA_HUB', f'ws://127.0.0.1:{PORT}/ws') +TITLE_MIN = 12 # a too short last input gets the previous one prepended +NOISE = re.compile(r'\**LLM Running \(Turn \d+\) \.\.\.\**|`{3,}.*?`{3,}|.*?', re.DOTALL) + +class HubClient: + def __init__(self, name, put_task, get_outputs, abort=None): + self.name, self.put_task, self.get_outputs, self.abort = name, put_task, get_outputs, abort + self._tc = {} # (i,j) -> (fingerprint, title): a finished step is never re-scanned + def start(self): + threading.Thread(target=lambda: asyncio.run(self._loop()), daemon=True).start(); return self + async def _loop(self): + try: import websockets + except ImportError: return + while True: + try: + async with websockets.connect(URL, open_timeout=3, max_size=None) as ws: + await ws.send(json.dumps({'op': 'hello', 'name': self.name, 'pid': os.getpid()})) + async for raw in ws: await self._on_cmd(ws, json.loads(raw)) + except Exception: pass # silent: the hub is optional, it must not disturb the host + await asyncio.sleep(5) + def _stitle(self, i, j, s): + s, e = s or '', self._tc.get((i, j)) + fp = (len(s), s[:24] + s[-24:]) # memoised on length+edges: guards index reuse after /clear + if e and e[0] == fp: return e[1] + body = NOISE.sub('', s) # title = if present, else first meaningful line + m = re.search(r'\s*(.*?)\s*', body, re.DOTALL) + t = (m.group(1) if m else body.strip()).strip().split('\n')[0]; t = t[:50] + '...' if len(t) > 50 else t or f'step {j + 1}' + self._tc[(i, j)] = (fp, t); return t + def _build(self, c): + """Skeleton only (no bodies), off-loop: regex/serialisation must not stall the WS reader.""" + tasks = list(self.get_outputs() or []); outs = [list(t.get('outputs') or []) for t in tasks] # copy: host mutates + sig = '.'.join(str(len(o)) for o in outs) + ':' + str(len(outs[-1][-1] or '') if outs and outs[-1] else 0) + if c.get('sig') == sig: return {'same': 1, 'sig': sig} # nothing moved -> no payload at all + ins = [(t.get('input') or '').strip() for t in tasks] + ins = [u for u in ins if u and not u.startswith('/')] or [u for u in ins if u] # /clear is plumbing + title = ins[-1].split('\n')[0] if ins else '(empty)' + if len(ins) > 1 and len(title) < TITLE_MIN: title = ins[-2].split('\n')[0] + ' <- ' + title + det, rows = int(c.get('detail', 1)), [] + for i, (t, o) in enumerate(zip(tasks, outs)): + steps = [{'j': j, 'title': self._stitle(i, j, s), 'n': len(s or '')} for j, s in enumerate(o)] + rows.append({'i': i, 'input': t.get('input', ''), **({'steps': steps} if det else {'ns': len(o)})}) + return {'title': title[:80], 'tasks': rows, 'sig': sig} + def _seg(self, c): + tasks, off = list(self.get_outputs() or []), max(0, int(c.get('off') or 0)) + i, j = int(c.get('i', -1)), int(c.get('j', -1)) + o = (tasks[i].get('outputs') or []) if -len(tasks) <= i < len(tasks) else None + if o is None or not (-len(o) <= j < len(o)): + return {'error': f'segment {i}/{j} gone (trimmed or cleared)', 'code': 'gone'} + s = o[j] or ''; return {'content': s[off:], 'off': off, 'n': len(s)} # off lets a UI tail a growing step + async def _on_cmd(self, ws, c): + op = c.get('op'); data = {'error': f'unknown op {op}', 'code': 'badop'} + if op in ('get', 'seg'): data = await asyncio.to_thread(self._build if op == 'get' else self._seg, c) + elif op == 'put_task': data = (c.get('text') and await asyncio.to_thread(self.put_task, c['text'])) or {'ok': 1} + 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 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 + (text, queue) in agent._hub_inbox so a UI can pop it and echo the task as its own bubble. + Never raises: a broken hub must not break its host.""" + def _put(text): + 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: + 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() + except Exception: return None + +# ------------------------------- server (python hub.py) ------------------------------- +if __name__ == '__main__': + from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, Response + from fastapi.responses import FileResponse, JSONResponse; import uvicorn + HERE, OK_ORIGIN = os.path.dirname(os.path.abspath(__file__)), ('', f'http://127.0.0.1:{PORT}', f'http://localhost:{PORT}') + STATUS = {'offline': 404, 'gone': 404, 'timeout': 504, 'nosupport': 501, 'badop': 400, 'busy': 409} + app, peers, waits, seq, pcache, pdead = FastAPI(), {}, {}, [0], {}, {} # ws / futures / last row / next probe + @app.middleware('http') + async def same_origin(req: Request, call_next): # browsers send Origin, native clients don't + return await call_next(req) if (req.headers.get('origin') or '') in OK_ORIGIN else Response(status_code=403) + async def ask(name, msg, timeout=15): + if name not in peers: return {'error': f'peer {name} offline', 'code': 'offline'} + seq[0] += 1; rid = seq[0] + waits[rid] = fut = asyncio.get_running_loop().create_future() + try: + await peers[name].send_text(json.dumps(dict(msg, id=rid))) + return await asyncio.wait_for(fut, timeout) # the WS reader resolves it + except asyncio.TimeoutError: return {'error': f'peer busy (>{timeout}s)', 'code': 'timeout'} + except Exception as e: return {'error': str(e) or type(e).__name__, 'code': 'offline'} + finally: waits.pop(rid, None) + def out(d): # a peer-level error becomes a real HTTP status, not a 200 with a body + c = d.get('code') if isinstance(d, dict) else None; return JSONResponse(d, STATUS.get(c, 400)) if c else d + @app.websocket('/ws') + async def endpoint(ws: WebSocket): + if (ws.headers.get('origin') or '') not in OK_ORIGIN: return await ws.close(code=1008) + await ws.accept(); name = None + try: + while True: + m = json.loads(await ws.receive_text()) + if m.get('op') == 'hello': + name = re.sub(r'[^\w.-]', '_', f"{m.get('name')}-{m.get('pid')}") # it lands in a URL path + peers[name] = ws; print(f'[+] {name} ({len(peers)} online)') + f = waits.get(m.get('id')) if m.get('op') == 'r' else None + if f and not f.done(): f.set_result(m.get('data')) + except WebSocketDisconnect: pass + finally: + if name: peers.pop(name, None); print(f'[-] {name} ({len(peers)} online)') + @app.get('/api/peers') + async def api_peers(): # one cheap round-trip per peer: counts only, and only if its sig moved + now = time.time() + names = [n for n in peers if now >= pdead.get(n, 0)] + rs = await asyncio.gather(*[ask(n, {'op': 'get', 'detail': 0, + 'sig': (pcache.get(n) or {}).get('sig')}, 3) for n in names]) + for n, r in zip(names, rs): pdead[n] = now + 15 if (r or {}).get('error') else 0 + rs, rows = dict(zip(names, rs)), [] + for n in peers: + r = rs.get(n) + if r is None or r.get('error') or (r.get('same') and n in pcache): # busy/skipped/idle -> last row + rows.append(pcache.get(n) or {'name': n, 'title': '?', 'n_msgs': 0, 'sig': None}); continue + msgs = sum(1 for t in r.get('tasks', []) if not (t.get('input') or '').lstrip().startswith('/')) + rows.append({'name': n, 'title': r.get('title', '?'), 'n_msgs': msgs, 'sig': r.get('sig')}) + if r.get('sig'): pcache[n] = rows[-1] + for n in [x for x in pcache if x not in peers]: pcache.pop(n, None) + return rows + @app.get('/api/{name}/messages') + async def api_messages(name: str, detail: int = 1, sig: str = None): return out(await ask(name, {'op': 'get', 'detail': detail, 'sig': sig})) + @app.get('/api/{name}/seg/{i}/{j}') + async def api_seg(name: str, i: int, j: int, off: int = 0): return out(await ask(name, {'op': 'seg', 'i': i, 'j': j, 'off': off})) + @app.post('/api/{name}/put') + async def api_put(name: str, body: dict): return out(await ask(name, {'op': 'put_task', 'text': body.get('text', '')})) + @app.post('/api/{name}/abort') + async def api_abort(name: str): return out(await ask(name, {'op': 'abort'})) + @app.get('/') + async def index(): return FileResponse(os.path.join(HERE, 'hub.html')) + @app.get('/vendor/{f}') + async def vendor(f: str): + p = os.path.join(HERE, 'desktop', 'static', 'vendor', os.path.basename(f)) + return FileResponse(p) if os.path.exists(p) else Response(status_code=404) + try: uvicorn.run(app, host='127.0.0.1', port=PORT, log_level='warning') + except OSError: sys.exit(f'hub already running on {PORT}') diff --git a/frontends/stapp.py b/frontends/stapp.py index e32c0b75..4d2357e2 100644 --- a/frontends/stapp.py +++ b/frontends/stapp.py @@ -49,6 +49,7 @@ I18N = { 'auto_on_cap': '🟢 已允许：约1分钟后启动，之后空闲30分钟再触发', 'auto_off_cap': '🔴 自主行动已停止', 'auto_prompt': '[AUTO]🤖 用户已经离开超过30分钟，作为自主智能体，请阅读自动化sop，执行自动任务。', + 'detached_running': '⏳ 后台任务运行中…（本页已刷新，实时流不再接入；完成后自动刷新）', 'get_token': '🔑 获取 Token', 'get_token_toast': '已打开浏览器', 'need_mykey': '⚠️ 请配置 mykey.py', @@ -66,6 +67,7 @@ I18N = { 'auto_on_cap': '🟢 On: first run ~1min, then every 30min idle', 'auto_off_cap': '🔴 Auto-action disabled', 'auto_prompt': '[AUTO]🤖 User has been idle for over 30 minutes. As an autonomous agent, read the automation SOP and execute automatic tasks.', + 'detached_running': '⏳ A task is running in the background… (this page was refreshed, live stream not attached; will refresh when done)', 'get_token': '🔑 Get Token', 'get_token_toast': 'Opened in browser', 'need_mykey': '⚠️ Please set mykey.py', @@ -83,6 +85,8 @@ def init(): return agent agent = init() +# NOTE: never abort merely because a new session appeared (F5 / 2nd tab / external probe). +# Aborting is an *intent*, triggered only by a new prompt or the Stop button. _sp = getattr(agentmain, "start_subscription_portal", None) @st.fragment(run_every=timedelta(seconds=2)) @@ -300,8 +304,8 @@ def _poll_main_task(max_items=256): return done -def _render_stat_badge(is_running, target=None): - """Render task usage/time into one stable element, without clearing its container.""" +def _render_stat_badge(is_running): + """Render task usage/time in the current app or fragment context.""" if 'task_start_ts' not in st.session_state or not hasattr(llmcore, 'STATS'): return end_ts = time.time() if is_running else st.session_state.get('task_end_ts', time.time()) @@ -312,13 +316,12 @@ def _render_stat_badge(is_running, target=None): f"{short(stats['ctx'])} chars·{stats['msgs']}msgs │ " f"in {short(stats.get('inp', 0))} toks·cached{short(stats.get('cached', 0))}·out{short(stats.get('out', 0))} │ " if 'ctx' in stats else '') - renderer = target.markdown if target is not None else st.markdown - renderer(f'
{usage}{secs // 60}:{secs % 60:02d}
', - unsafe_allow_html=True) + st.markdown(f'
{usage}{secs // 60}:{secs % 60:02d}
', + unsafe_allow_html=True) @st.fragment(run_every=timedelta(seconds=1)) -def render_main_stream(frozen_host, live_slot, stats_slot, render_state): +def render_main_stream(frozen_host, live_slot, render_state): """Append completed turns outside the fragment; redraw only the active turn.""" done = _poll_main_task() if done is not None: @@ -344,21 +347,24 @@ def render_main_stream(frozen_host, live_slot, stats_slot, render_state): # These external slots keep stable delta paths; only their contents change. with live_slot.container(): render_segments([segments[-1]], suffix=" ▌") - _render_stat_badge(is_running=True, target=stats_slot) + _render_stat_badge(is_running=True) -def mount_main_stream(stats_slot): - """Create persistent targets captured by the fragment until the next full rerun.""" +def mount_main_stream(): + """Create persistent response targets captured by the fragment until the next full rerun.""" with st.chat_message("assistant"): frozen_host = st.container() live_slot = st.empty() - render_main_stream(frozen_host, live_slot, stats_slot, {'frozen': 0}) + render_main_stream(frozen_host, live_slot, {'frozen': 0}) 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 # 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 = 6 +_HIST_TAIL = 10 _msgs = st.session_state.messages if len(_msgs) > _HIST_TAIL: if not st.session_state.get("show_full_history"): @@ -398,10 +404,6 @@ _js_ime_fix = ("" if os.name == 'nt' else "f();new MutationObserver(f).observe(d.body,{childList:1,subtree:1})}()") _embed_html(f'', height=0) -# Fixed target shared by full-app rendering and the one-second stream fragment. -# The badge is position:fixed, so this placeholder does not affect its visual position. -stats_slot = st.empty() - _injected = st.session_state.pop('_inject_prompt', None) prompt = st.chat_input("any task?") or _injected if prompt: @@ -415,7 +417,7 @@ if prompt: st.session_state.last_reply_time = int(time.time()) st.rerun() if cmd == "/new": - st.session_state.messages = [{"role": "assistant", "content": reset_conversation(agent), "time": ts}] + st.session_state.messages[:] = [{"role": "assistant", "content": reset_conversation(agent), "time": ts}] _reset_and_rerun() if cmd.startswith("/continue"): m = re.match(r'/continue\s+(\d+)\s*$', cmd.strip()) @@ -426,15 +428,15 @@ if prompt: result = handle_frontend_command(agent, cmd) history = extract_ui_messages(target) if target and result.startswith('✅') else None tail = [{"role": "assistant", "content": result, "time": ts}] - if history: st.session_state.messages = history + tail - else: st.session_state.messages = list(st.session_state.messages)+[{"role": "user", "content": cmd, "time": ts}]+tail + if history: st.session_state.messages[:] = history + tail + else: st.session_state.messages.extend([{"role": "user", "content": cmd, "time": ts}] + tail) _reset_and_rerun() if cmd.startswith("/btw"): answer = btw_handle_frontend(agent, cmd) # sync; bypasses put_task → main agent.run() untouched - st.session_state.messages = list(st.session_state.messages) + [ + st.session_state.messages.extend([ {"role": "user", "content": prompt, "time": ts}, {"role": "assistant", "content": answer, "time": ts}, - ] + ]) st.rerun() # preserve display_queue/partial_response so resume path drains the running main task if cmd.startswith("/export"): parts = cmd.split(maxsplit=1) @@ -463,29 +465,73 @@ if prompt: result = f"✅ 已导出:\n\n`{path}`" except Exception as e: result = f"❌ 导出失败: {e}" - st.session_state.messages = list(st.session_state.messages) + [ + st.session_state.messages.extend([ {"role": "user", "content": cmd, "time": ts}, {"role": "assistant", "content": result, "time": ts}, - ] + ]) _reset_and_rerun() # Regular prompt starts a new main task. Explicitly detach any prior task first; # sidebar-only reruns never pass through this branch, so they keep the queue. - if st.session_state.get("display_queue") is not None: + if agent.is_running or st.session_state.get("display_queue") is not None: _cancel_main_task() st.session_state.messages.append({"role": "user", "content": prompt}) if hasattr(agent, '_pet_req') and not prompt.startswith('/'): agent._pet_req('state=walk') with st.chat_message("user"): st.markdown(prompt) _start_main_task(prompt) - mount_main_stream(stats_slot) + mount_main_stream() elif st.session_state.get('display_queue') is not None: # No new prompt but a task is mid-flight (typically a /btw rerun) — resume drain. - mount_main_stream(stats_slot) + mount_main_stream() +elif getattr(agent, '_hub_inbox', None): + # A task pushed in over the hub: adopt it as if typed here, so it gets a user bubble and a + # live stream instead of the anonymous "something is running" notice. pop() -> the first tab + # to rerun claims it; later tabs fall through to the detached view below. + _hm = agent._hub_inbox.pop(0) + st.session_state.messages.append({"role": "user", "content": _hm['text']}) + with st.chat_message("user"): st.markdown(_hm['text']) + st.session_state.display_queue = _hm['q'] + st.session_state.partial_response = "" + st.session_state.task_start_ts = time.time() + mount_main_stream() +elif agent.is_running or (getattr(agent, '_current_queue', None) is not None + and not agent._current_queue.empty()): + # A task is running but this session owns no queue (page was refreshed / opened in a + # 2nd tab). The stream is per-session and intentionally gone; only show liveness so the + # page is never blank-and-silent, and refresh once the backend goes idle. + @st.fragment(run_every=timedelta(seconds=1)) + def _watch_detached_task(): + # Re-attach to the live task's queue only to salvage its final ``done`` so the + # answer still lands in history. The ``next`` stream is per-session and is + # deliberately dropped rather than resumed. + dq = getattr(agent, "_current_queue", None) + if dq is not None: + while True: + try: item = dq.get_nowait() + except Exception: break + if isinstance(item, dict) and "done" in item: + if item["done"]: + st.session_state.messages.append({"role": "assistant", "content": item["done"]}) + st.rerun(scope="app"); return + if not agent.is_running: + st.rerun(scope="app"); return + st.chat_message("assistant").markdown(T("detached_running")) + + _watch_detached_task() + +@st.fragment(run_every=timedelta(seconds=2)) +def _watch_hub_inbox(): + # An idle page has no heartbeat, so a task arriving from the hub would sit unnoticed until the + # next click. Poll cheaply and hand it to the claim branch above via a full rerun. + if getattr(agent, '_hub_inbox', None) and st.session_state.get('display_queue') is None: + st.rerun(scope="app") + +_watch_hub_inbox() # Usage/time label remains after completion. During a task render_main_stream refreshes it. _has_task_stats = 'task_start_ts' in st.session_state _is_running = st.session_state.get('display_queue') is not None if _has_task_stats and not _is_running: - _render_stat_badge(is_running=False, target=stats_slot) + _render_stat_badge(is_running=False) if _has_task_stats or _is_running: st.markdown(