diff --git a/app/chat_service.py b/app/chat_service.py index 8974d32..40c725f 100644 --- a/app/chat_service.py +++ b/app/chat_service.py @@ -43,7 +43,6 @@ async def get_workspace_cookie(universe_id: str) -> str: raise RuntimeError(f"Failed to authenticate with {universe_id}: {resp.status_code} {resp.text}") cookie_header = resp.headers.get("set-cookie", "") - # Extract claude-auth cookie token_part = "" for part in cookie_header.split(";"): if "claude-auth=" in part: @@ -55,7 +54,7 @@ async def get_workspace_cookie(universe_id: str) -> str: cookie_str = token_part or cookie_header.split(";")[0].strip() _AUTH_COOKIES[universe_id] = { "cookie": cookie_str, - "expires_at": now + (25 * 86400) # 25 days validity + "expires_at": now + (25 * 86400) } return cookie_str @@ -85,14 +84,26 @@ async def get_nabil_session_cookie() -> str: cookie_str = "; ".join(cookie_parts) _AUTH_COOKIES["nabil"] = { "cookie": cookie_str, - "expires_at": now + 3600 # 1 hour + "expires_at": now + 3600 } return cookie_str +async def get_nabil_ws_ticket() -> str: + """Acquires a single-use WebSocket ticket from Nabil.""" + cookie = await get_nabil_session_cookie() + async with httpx.AsyncClient(timeout=8.0) as client: + resp = await client.post( + "http://hermes-nabil:9119/api/auth/ws-ticket", + json={}, + headers={"Cookie": cookie} + ) + if resp.status_code == 200: + return resp.json().get("ticket", "") + return "" + async def list_universe_conversations(universe_id: str) -> List[Dict[str, Any]]: """Lists conversations for a given universe.""" if universe_id == "nabil": - # Sync from Nabil backend and merge with local SQLite store try: cookie = await get_nabil_session_cookie() async with httpx.AsyncClient(timeout=6.0) as client: @@ -108,18 +119,16 @@ async def list_universe_conversations(universe_id: str) -> List[Dict[str, Any]]: started_at = s.get("started_at") c_at = int(started_at * 1000) if isinstance(started_at, (int, float)) and started_at < 1e11 else int(time.time() * 1000) save_conversation("nabil", sid, title=stitle, pinned=spinned, created_at=c_at) - except Exception as e: - # Fallback to local store if backend unreachable + except Exception: pass return list_conversations("nabil") elif universe_id in ("tt", "nyora", "perso"): - # For workspace backends, SQLite is the canonical index return list_conversations(universe_id) else: return [] async def create_universe_conversation(universe_id: str, title: Optional[str] = None) -> Dict[str, Any]: - """Creates a new conversation in the universe and indexes it in Hub SQLite.""" + """Creates a new conversation on backend and indexes it in Hub SQLite.""" if universe_id in ("tt", "nyora", "perso"): cookie = await get_workspace_cookie(universe_id) universe = UNIVERSES[universe_id] @@ -134,8 +143,41 @@ async def create_universe_conversation(universe_id: str, title: Optional[str] = conv = save_conversation(universe_id, session_key, title=title or "Nouvelle conversation") return conv elif universe_id == "nabil": - session_key = str(uuid.uuid4()) - conv = save_conversation(universe_id, session_key, title=title or "Nouvelle conversation") + # Create session explicitly on hermes-nabil backend via session.create RPC + cookie = await get_nabil_session_cookie() + ticket = await get_nabil_ws_ticket() + session_title = title or "Nouvelle conversation" + session_key = None + + ws_url = f"ws://hermes-nabil:9119/api/ws?ticket={ticket}" + try: + async with websockets.connect( + ws_url, + additional_headers={"Cookie": cookie}, + ping_interval=20, + ping_timeout=20 + ) as ws: + req_id = f"create-{uuid.uuid4().hex[:8]}" + await ws.send(json.dumps({ + "jsonrpc": "2.0", + "id": req_id, + "method": "session.create", + "params": {"title": session_title} + })) + + # Wait for session.create result frame + while True: + raw = await asyncio.wait_for(ws.recv(), timeout=6.0) + data = json.loads(raw) + if data.get("id") == req_id and "result" in data: + res = data["result"] + session_key = res.get("session_id") or res.get("stored_session_id") + break + except Exception as e: + # Fallback if WS fails + session_key = f"nabil_{uuid.uuid4().hex[:12]}" + + conv = save_conversation(universe_id, session_key, title=session_title) return conv else: raise ValueError(f"Unknown universe {universe_id}") @@ -176,7 +218,6 @@ async def get_conversation_messages(universe_id: str, session_key: str) -> List[ normalized = [] for m in raw_messages: content = m.get("content", "") - # If assistant only has tool calls, present readable summary if not content and m.get("tool_calls"): tool_names = [tc.get("function", {}).get("name", "tool") for tc in m.get("tool_calls", [])] content = f"[Outils exécutés : {', '.join(tool_names)}]" @@ -215,15 +256,11 @@ async def stream_chat_messages(universe_id: str, session_key: str, message: str) yield f"{line}\n" else: yield "\n" - # Touch updated_at save_conversation(universe_id, session_key) elif universe_id == "nabil": cookie = await get_nabil_session_cookie() - # Acquire ws ticket - async with httpx.AsyncClient(timeout=8.0) as client: - resp_ticket = await client.post("http://hermes-nabil:9119/api/auth/ws-ticket", json={}, headers={"Cookie": cookie}) - ticket = resp_ticket.json().get("ticket") if resp_ticket.status_code == 200 else "" + ticket = await get_nabil_ws_ticket() yield "event: started\ndata: {\"status\": \"connecting\"}\n\n" @@ -231,17 +268,18 @@ async def stream_chat_messages(universe_id: str, session_key: str, message: str) try: async with websockets.connect( ws_url, - additional_headers={"Cookie": cookie} + additional_headers={"Cookie": cookie}, + ping_interval=20, + ping_timeout=20 ) as ws: - # Send chat message via JSON-RPC - req_id = str(uuid.uuid4()) + req_id = f"prompt-{uuid.uuid4().hex[:8]}" send_payload = { "jsonrpc": "2.0", "id": req_id, - "method": "chat.send", + "method": "prompt.submit", "params": { "session_id": session_key, - "message": message + "text": message } } await ws.send(json.dumps(send_payload)) @@ -249,23 +287,22 @@ async def stream_chat_messages(universe_id: str, session_key: str, message: str) accumulated_text = "" while True: try: - raw = await asyncio.wait_for(ws.recv(), timeout=45.0) + raw = await asyncio.wait_for(ws.recv(), timeout=60.0) data = json.loads(raw) - method = data.get("method") params = data.get("params", {}) + ev_type = params.get("type", "") + payload = params.get("payload", {}) - if method == "chat.chunk": - chunk_text = params.get("text", "") - accumulated_text += chunk_text - yield f"event: chunk\ndata: {json.dumps({'text': accumulated_text, 'chunk': chunk_text, 'fullReplace': True})}\n\n" - elif method in ("chat.done", "session.done", "gateway.ready"): - if method != "gateway.ready": - yield f"event: done\ndata: {json.dumps({'state': 'complete', 'text': accumulated_text})}\n\n" - break - elif data.get("id") == req_id and "result" in data: - res_text = data["result"].get("content") or data["result"].get("text") or accumulated_text - yield f"event: done\ndata: {json.dumps({'state': 'complete', 'text': res_text})}\n\n" + if ev_type in ("message.delta", "chat.chunk"): + delta = payload.get("text") or params.get("text") or "" + accumulated_text += delta + yield f"event: chunk\ndata: {json.dumps({'text': accumulated_text, 'chunk': delta, 'fullReplace': True})}\n\n" + elif ev_type in ("reasoning.delta", "thinking.delta"): + delta = payload.get("text", "") + # We can stream reasoning as needed + elif ev_type in ("message.finish", "session.idle", "reasoning.available") or data.get("method") in ("chat.done", "session.done"): + yield f"event: done\ndata: {json.dumps({'state': 'complete', 'text': accumulated_text})}\n\n" break except asyncio.TimeoutError: break @@ -276,7 +313,6 @@ async def stream_chat_messages(universe_id: str, session_key: str, message: str) save_conversation(universe_id, session_key) async def rename_universe_conversation(universe_id: str, session_key: str, title: str) -> bool: - """Renames a conversation in SQLite and on Nabil backend if applicable.""" update_conversation(universe_id, session_key, title=title) if universe_id == "nabil": try: @@ -292,7 +328,6 @@ async def rename_universe_conversation(universe_id: str, session_key: str, title return True async def pin_universe_conversation(universe_id: str, session_key: str, pinned: bool) -> bool: - """Pins/unpins a conversation in SQLite and on Nabil backend if applicable.""" update_conversation(universe_id, session_key, pinned=pinned) if universe_id == "nabil": try: @@ -308,7 +343,6 @@ async def pin_universe_conversation(universe_id: str, session_key: str, pinned: return True async def delete_universe_conversation(universe_id: str, session_key: str) -> bool: - """Deletes a conversation from SQLite and from backend if supported.""" delete_conversation_record(universe_id, session_key) if universe_id in ("tt", "nyora", "perso"): try: diff --git a/app/proxy.py b/app/proxy.py index 239980d..d84d3a7 100644 --- a/app/proxy.py +++ b/app/proxy.py @@ -1,5 +1,6 @@ import httpx import re +import socket import asyncio from typing import AsyncGenerator, Dict, Any, Optional, List, Tuple from fastapi import Request, Response, WebSocket, WebSocketDisconnect @@ -30,7 +31,9 @@ WS_HOP_BY_HOP_HEADERS = { "sec-websocket-extensions", "sec-websocket-accept", "host", - "origin" + "origin", + "sec-fetch-site", + "sec-fetch-mode" } _http_transport: Optional[httpx.AsyncHTTPTransport] = None @@ -98,14 +101,15 @@ async def proxy_request( k_str = raw_k.decode("latin-1").lower() if k_str in HOP_BY_HOP_HEADERS or k_str == "host": continue - if is_dsh and k_str == "origin": + if is_dsh and k_str in ("origin", "sec-fetch-site"): continue req_headers.append((raw_k, raw_v)) - # If DSH backend, replicate localhost:3080 Host/Origin for internal origin check + # If DSH backend, replicate localhost:3080 Host/Origin and same-origin site if is_dsh: req_headers.append((b"host", b"localhost:3080")) req_headers.append((b"origin", b"http://localhost:3080")) + req_headers.append((b"sec-fetch-site", b"same-origin")) else: req_headers.append((b"x-forwarded-host", request.headers.get("host", "").encode("latin-1"))) @@ -180,13 +184,7 @@ async def proxy_websocket( path: str, universe_id: Optional[str] = None ): - ws_base = backend_url.replace("https://", "wss://").replace("http://", "ws://").rstrip("/") sub_path = path.lstrip("/") - upstream_url = f"{ws_base}/{sub_path}" if sub_path else ws_base - - if client_ws.url.query: - upstream_url = f"{upstream_url}?{client_ws.url.query}" - is_dsh = "dsh-vps:3080" in backend_url or universe_id == "dsh" upstream_headers = {} @@ -194,10 +192,27 @@ async def proxy_websocket( if k.lower() not in WS_HOP_BY_HOP_HEADERS: upstream_headers[k] = v + subprotocols_raw = client_ws.headers.get("sec-websocket-protocol", "") + subprotocols = [s.strip() for s in subprotocols_raw.split(",") if s.strip()] or None + + sock = None if is_dsh: - upstream_headers["host"] = "localhost:3080" + # DSH requires exact Host: localhost:3080, Origin: http://localhost:3080 and Sec-Fetch-Site: same-origin + upstream_headers["Sec-Fetch-Site"] = "same-origin" + upstream_url = f"ws://localhost:3080/{sub_path}" if sub_path else "ws://localhost:3080" + if client_ws.url.query: + upstream_url = f"{upstream_url}?{client_ws.url.query}" + + # Connect raw socket directly to dsh-vps:3080 + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect(("dsh-vps", 3080)) + sock.setblocking(False) origin_val = "http://localhost:3080" else: + ws_base = backend_url.replace("https://", "wss://").replace("http://", "ws://").rstrip("/") + upstream_url = f"{ws_base}/{sub_path}" if sub_path else ws_base + if client_ws.url.query: + upstream_url = f"{upstream_url}?{client_ws.url.query}" upstream_headers["x-forwarded-host"] = client_ws.headers.get("host", "") upstream_headers["x-forwarded-proto"] = client_ws.url.scheme or "http" origin_val = None @@ -205,9 +220,6 @@ async def proxy_websocket( if client_ws.client: upstream_headers["x-forwarded-for"] = client_ws.client.host - subprotocols_raw = client_ws.headers.get("sec-websocket-protocol", "") - subprotocols = [s.strip() for s in subprotocols_raw.split(",") if s.strip()] or None - try: connect_kwargs = { "additional_headers": upstream_headers, @@ -216,6 +228,8 @@ async def proxy_websocket( "ping_timeout": 20, "max_size": 10 * 1024 * 1024 } + if sock: + connect_kwargs["sock"] = sock if origin_val: connect_kwargs["origin"] = origin_val diff --git a/static/js/hub.js b/static/js/hub.js index d940315..cea306e 100644 --- a/static/js/hub.js +++ b/static/js/hub.js @@ -630,7 +630,7 @@ document.addEventListener('DOMContentLoaded', () => { tabFiles.addEventListener('click', () => { tabFiles.classList.add('active'); tabSession.classList.remove('active'); - showIframeView(`https://dsh-hub.yesminedor.tn/`); + showIframeView(`https://files-hub.yesminedor.tn/`); }); }