diff --git a/tools/server/README-dev.md b/tools/server/README-dev.md index b4ec9f17d3..b41d70c63a 100644 --- a/tools/server/README-dev.md +++ b/tools/server/README-dev.md @@ -136,13 +136,13 @@ Producer side: `server_res_generator` extends `server_res_spipe`, which keeps al Lifetime safety: the session holds no back reference to the response, so `spipe` is a plain `unique_ptr` touched only by the http worker. `cancel` raises an atomic the producer polls; the producer finalizes the session from its destructor, which also runs `~server_response_reader::stop()` to cancel the generation at the queue level. A `DELETE` stops work by raising the flag and letting the worker unwind. -Consumer side: `GET /v1/stream/?from=N` opens a `text/event-stream` that replays buffered bytes from offset `N` and blocks for live bytes, so the browser reattaches like a fresh EventSource. An offset below the dropped prefix returns 400. +Consumer side: `GET /v1/stream?conv_id=&from=N` opens a `text/event-stream` that replays buffered bytes from offset `N` and blocks for live bytes, so the browser reattaches like a fresh EventSource. An offset below the dropped prefix returns 400. Routes: -- `GET /v1/stream/:conv_id?from=N`: replay or live reattach. +- `GET /v1/stream?conv_id=&from=N`: replay or live reattach. The id travels in the query string because it can embed a model name containing slashes. - `POST /v1/streams/lookup` with `{"conversation_ids": [...]}`: returns session status only for ids the caller already owns. There is no listing route, so live sessions cannot be enumerated (an earlier `GET /v1/streams` was removed for exactly this reason). -- `DELETE /v1/stream/:conv_id`: explicit Stop, idempotent (`evict_and_cancel`). +- `DELETE /v1/stream?conv_id=`: explicit Stop, idempotent (`evict_and_cancel`). Router mode binds the same paths to proxy handlers. A `conv_id -> child` map (`conv_models`), populated when a POST is routed, resolves the owning child in one lookup with no polling. The lookup groups ids per child; GET and DELETE proxy straight to the owner. This loopback REST hop is expected to move to a websocket IPC later, swapping only the transport. @@ -166,8 +166,8 @@ graph TD GC[GC thread] -- drop after TTL --> Sess end Sess -- read_from offset --> Cons[stream_pipe_consumer] - Cons -- "GET /v1/stream/:id?from=N" --> Client - DEL[DELETE /v1/stream/:id] -- evict_and_cancel --> Sess + Cons -- "GET /v1/stream?conv_id=id&from=N" --> Client + DEL[DELETE /v1/stream?conv_id=id] -- evict_and_cancel --> Sess ``` The diagram shows the buffer touch points. The live wire (chunks streamed to the original client during a normal generation) is the producer's default output, described under "Producer side" above. diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp index 923b3533e9..188a72a374 100644 --- a/tools/server/server-models.cpp +++ b/tools/server/server-models.cpp @@ -1172,7 +1172,7 @@ bool server_models::ensure_model_ready(const std::string & name) { return true; } -server_http_res_ptr server_models::proxy_request(const server_http_req & req, const std::string & method, const std::string & name, bool update_last_used) { +server_http_res_ptr server_models::proxy_request(const server_http_req & req, const std::string & method, const std::string & name, bool update_last_used, bool detached) { auto meta = get_meta(name); if (!meta.has_value()) { throw std::runtime_error("model name=" + name + " is not found"); @@ -1198,7 +1198,10 @@ server_http_res_ptr server_models::proxy_request(const server_http_req & req, co req.headers, req.body, req.files, - req.should_stop, + // a detached request belongs to a replay session that outlives the client socket: + // it reaches the child even when the downstream died during the load wait, the + // session buffer is the recipient and DELETE remains the stop + detached ? std::function([]() { return false; }) : req.should_stop, base_params.timeout_read, base_params.timeout_write ); @@ -1469,13 +1472,9 @@ static bool router_validate_model(std::string & name, server_models & models, bo } // resolve alias to canonical model name name = meta->name; - if (models_autoload) { - models.ensure_model_ready(name); - } else { - if (!meta->is_running()) { - res_err(res, format_error_response("model is not loaded", ERROR_TYPE_INVALID_REQUEST)); - return false; - } + if (!models_autoload && !meta->is_running()) { + res_err(res, format_error_response("model is not loaded", ERROR_TYPE_INVALID_REQUEST)); + return false; } return true; } @@ -1568,6 +1567,9 @@ void server_models_routes::init_routes() { if (!router_validate_model(name, models, autoload, error_res)) { return error_res; } + if (autoload) { + models.ensure_model_ready(name); + } return models.proxy_request(req, method, name, false); }; @@ -1581,12 +1583,23 @@ void server_models_routes::init_routes() { return error_res; } // remember which child serves this conversation so the stream routes can route straight - // to it without polling, keyed on the exact conv id from the header + // to it without polling, keyed on the exact conv id from the header. registered before + // the load wait so a stop issued while the model loads can erase the entry and cancel + // this request instead of leaving an orphan generation std::string conv_id = server_stream_conv_id_from_headers(req.headers); - if (!conv_id.empty()) { - models.conv_models.remember(conv_id, name); + uint64_t ticket = models.conv_models.remember(conv_id, name); + bool waited = autoload && models.ensure_model_ready(name); + if (ticket != 0 && !models.conv_models.alive(conv_id, ticket)) { + SRV_INF("request for conv_id=%s cancelled while model name=%s was loading\n", + conv_id.c_str(), name.c_str()); + res_err(error_res, format_error_response( + "request cancelled by a stop while the model was loading", ERROR_TYPE_INVALID_REQUEST)); + return error_res; } - return models.proxy_request(req, method, name, true); // update last usage for POST request only + // a session request that waited for a load detaches from the client socket: the + // client may have dropped during the wait (page reload) and the session buffer must + // still receive the generation for a later resume + return models.proxy_request(req, method, name, true, waited && ticket != 0); // update last usage for POST request only }; this->post_router_models_load = [this](const server_http_req & req) { @@ -1779,7 +1792,7 @@ void server_models_routes::init_routes() { }; this->router_stream_get = [this](const server_http_req & req) { - // GET /v1/stream/?from=N. resolve the owning child from the conv_id -> model + // GET /v1/stream?conv_id=&from=N. resolve the owning child from the conv_id -> model // map, 404 when nothing maps auto res = std::make_unique(); std::string conv_id = req.get_param("conv_id"); @@ -1789,13 +1802,24 @@ void server_models_routes::init_routes() { } std::optional owner = resolve_child_for_conv(models, conv_id); if (!owner.has_value()) { - res_err(res, format_error_response("Stream not found or expired", ERROR_TYPE_NOT_FOUND)); + // a registered conv whose model is still loading earns a retry: the session appears + // once the load ends and the pending request reaches the child + auto tracked = models.conv_models.lookup(conv_id); + auto meta = tracked.has_value() ? models.get_meta(*tracked) : std::nullopt; + bool transient = meta.has_value() && (meta->status == SERVER_MODEL_STATUS_LOADING || + meta->status == SERVER_MODEL_STATUS_DOWNLOADING || + meta->status == SERVER_MODEL_STATUS_DOWNLOADED); + if (transient) { + res_err(res, format_error_response("Stream owner model is loading, retry later", ERROR_TYPE_UNAVAILABLE)); + } else { + res_err(res, format_error_response("Stream not found or expired", ERROR_TYPE_NOT_FOUND)); + } return res; } std::string from = req.get_param("from"); - std::string child_path = "/v1/stream/" + encode_qs(conv_id); + std::string child_path = "/v1/stream?conv_id=" + encode_qs(conv_id); if (!from.empty()) { - child_path += "?from=" + from; + child_path += "&from=" + from; } SRV_TRC("proxying stream resume to model %s on port %d, path=%s\n", owner->name.c_str(), owner->port, child_path.c_str()); @@ -1875,7 +1899,7 @@ void server_models_routes::init_routes() { }; this->router_stream_delete = [this](const server_http_req & req) { - // DELETE /v1/stream/. resolve the owning child via the map and forward only to + // DELETE /v1/stream?conv_id=. resolve the owning child via the map and forward only to // it, evict_and_cancel is idempotent on the child auto res = std::make_unique(); std::string conv_id = req.get_param("conv_id"); @@ -1883,7 +1907,7 @@ void server_models_routes::init_routes() { res_err(res, format_error_response("Missing conversation id in path", ERROR_TYPE_INVALID_REQUEST)); return res; } - std::string child_path = "/v1/stream/" + encode_qs(conv_id); + std::string child_path = "/v1/stream?conv_id=" + encode_qs(conv_id); auto owner = resolve_child_for_conv(models, conv_id); if (owner.has_value()) { httplib::Client cli(CHILD_ADDR, owner->port); @@ -1892,6 +1916,11 @@ void server_models_routes::init_routes() { cli.set_write_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000); auto resp = cli.Delete(child_path.c_str()); (void) resp; // the child logs its own miss when the session is unknown there + } else if (auto tracked = models.conv_models.lookup(conv_id); tracked.has_value()) { + // the entry exists but its model is still loading: the forget below erases it, + // which cancels the request parked in proxy_post before the generation starts + SRV_INF("router stop for conv_id=%s while model name=%s is loading, cancelling the pending request\n", + conv_id.c_str(), tracked->c_str()); } else { SRV_WRN("router stop for unknown conv_id=%s, no owning child in the conv map\n", conv_id.c_str()); diff --git a/tools/server/server-models.h b/tools/server/server-models.h index 62bed8725b..614798186c 100644 --- a/tools/server/server-models.h +++ b/tools/server/server-models.h @@ -134,12 +134,24 @@ private: // proxy_request forwards a POST carrying an X-Conversation-Id. best effort: a stale entry just // makes the child answer not found and the client recovers. owns its lock, one mutex per struct struct conv_model_tracker { - void remember(const std::string & conv_id, const std::string & model) { + // returns the ticket of this registration, 0 when nothing was registered. erasing or + // replacing the entry invalidates the ticket, which is how a stop cancels a request + // parked in the model load wait + uint64_t remember(const std::string & conv_id, const std::string & model) { if (conv_id.empty() || model.empty()) { - return; + return 0; } std::lock_guard lock(mu); - map[conv_id] = model; + uint64_t ticket = next_ticket++; + map[conv_id] = { model, ticket }; + return ticket; + } + + // false means a stop erased the entry or a newer request replaced it + bool alive(const std::string & conv_id, uint64_t ticket) { + std::lock_guard lock(mu); + auto it = map.find(conv_id); + return it != map.end() && it->second.ticket == ticket; } std::optional lookup(const std::string & conv_id) { @@ -151,7 +163,7 @@ private: if (it == map.end()) { return std::nullopt; } - return it->second; + return it->second.model; } void forget(const std::string & conv_id) { @@ -163,8 +175,13 @@ private: } private: - std::mutex mu; - std::unordered_map map; + struct entry_t { + std::string model; + uint64_t ticket; + }; + std::mutex mu; + uint64_t next_ticket = 1; + std::unordered_map map; }; common_preset_context ctx_preset; @@ -249,7 +266,7 @@ public: bool ensure_model_ready(const std::string & name); // proxy an HTTP request to the model instance - server_http_res_ptr proxy_request(const server_http_req & req, const std::string & method, const std::string & name, bool update_last_used); + server_http_res_ptr proxy_request(const server_http_req & req, const std::string & method, const std::string & name, bool update_last_used, bool detached = false); // handle message sent from server_child::notify_to_router() // raw input must starts with CMD_CHILD_TO_ROUTER_STATE, followed by a JSON string diff --git a/tools/server/server-stream.cpp b/tools/server/server-stream.cpp index f0a35b18e5..f6b9b8a9f4 100644 --- a/tools/server/server-stream.cpp +++ b/tools/server/server-stream.cpp @@ -453,7 +453,7 @@ static server_http_res_ptr make_error_response(int status, const std::string & m server_http_context::handler_t server_stream_make_get_handler() { return [](const server_http_req & req) -> server_http_res_ptr { - // GET /v1/stream/?from=N replays buffered SSE bytes then blocks for live + // GET /v1/stream?conv_id=&from=N replays buffered SSE bytes then blocks for live // bytes until the session finalizes, streamed as text/event-stream for EventSource std::string conv_id = req.get_param("conv_id"); if (conv_id.empty()) { @@ -560,13 +560,13 @@ server_http_context::handler_t server_stream_make_lookup_handler() { server_http_context::handler_t server_stream_make_delete_handler() { return [](const server_http_req & req) -> server_http_res_ptr { - // DELETE /v1/stream/ is the explicit user Stop, cancels the producer and evicts + // DELETE /v1/stream?conv_id= is the explicit user Stop, cancels the producer and evicts // the buffer. idempotent, returns 204 even if the session was already gone std::string conv_id = req.get_param("conv_id"); if (conv_id.empty()) { return make_error_response(400, "Missing conversation id in path", ERROR_TYPE_INVALID_REQUEST); } - SRV_TRC("DELETE /v1/stream/%s -> evict_and_cancel\n", conv_id.c_str()); + SRV_TRC("DELETE /v1/stream conv_id=%s -> evict_and_cancel\n", conv_id.c_str()); g_stream_sessions.evict_and_cancel(conv_id); auto res = std::make_unique(); res->status = 204; @@ -621,7 +621,7 @@ bool server_res_spipe::conn_alive() { bool server_res_spipe::should_stop() { if (spipe) { - // note: if DELETE /v1/stream/ is called, is_cancelled() will be true + // note: if DELETE /v1/stream is called for this conv, is_cancelled() will be true return spipe->is_cancelled(); } else { return !conn_alive(); diff --git a/tools/server/server-stream.h b/tools/server/server-stream.h index 9753140dd6..1e7461285f 100644 --- a/tools/server/server-stream.h +++ b/tools/server/server-stream.h @@ -45,7 +45,13 @@ void server_stream_session_manager_start(); void server_stream_session_manager_stop(); // route handler factories wired under /v1/stream/* by server.cpp +// child-side handlers for the resumable stream routes. the conv id travels in the conv_id +// query string because it can embed a model name containing slashes (org/repo), which the +// decoded path would split before the param is captured server_http_context::handler_t server_stream_make_get_handler(); +// POST /v1/streams/lookup with body {"conversation_ids": [...]}: only answers for ids the +// caller already owns (the WebUI passes the convs visible in its sidebar), the server never +// lists ids it has not been asked about, so a random caller cannot enumerate live sessions server_http_context::handler_t server_stream_make_lookup_handler(); server_http_context::handler_t server_stream_make_delete_handler(); diff --git a/tools/server/server.cpp b/tools/server/server.cpp index b6fef99e87..a3b2a8b0fe 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -272,10 +272,8 @@ int llama_server(common_params & params, int argc, char ** argv) { ctx_http.get ("/slots", ex_wrapper(routes.get_slots)); ctx_http.post("/slots/:id_slot", ex_wrapper(routes.post_slots)); - // resumable streaming, the conversation_id is the session identity end to end. router and - // child wire different handlers under the same paths: a child binds the local session - // factories, the router binds proxies that resolve the owning child through the - // conv_id -> model map + // resumable streaming: a child binds the local session factories, the router binds + // proxies that resolve the owning child, see server-stream.h server_http_context::handler_t stream_get_h; server_http_context::handler_t streams_lookup_h; server_http_context::handler_t stream_delete_h; @@ -288,12 +286,9 @@ int llama_server(common_params & params, int argc, char ** argv) { streams_lookup_h = server_stream_make_lookup_handler(); stream_delete_h = server_stream_make_delete_handler(); } - ctx_http.get ("/v1/stream/:conv_id", ex_wrapper(stream_get_h)); - // POST /v1/streams/lookup with body {"conversation_ids": [...]}. you can only ask for ids - // you already own (the WebUI passes the convs visible in its sidebar). the server never - // lists ids it has not been asked about, so a random caller cannot enumerate live sessions + ctx_http.get ("/v1/stream", ex_wrapper(stream_get_h)); ctx_http.post("/v1/streams/lookup", ex_wrapper(streams_lookup_h)); - ctx_http.del ("/v1/stream/:conv_id", ex_wrapper(stream_delete_h)); + ctx_http.del ("/v1/stream", ex_wrapper(stream_delete_h)); // Google Cloud Platform (Vertex AI) compat ctx_http.register_gcp_compat(); diff --git a/tools/server/tests/unit/test_stream.py b/tools/server/tests/unit/test_stream.py new file mode 100644 index 0000000000..a1ef55567b --- /dev/null +++ b/tools/server/tests/unit/test_stream.py @@ -0,0 +1,153 @@ +import json +import socket +import threading +import time +from urllib.parse import quote +import pytest +from utils import * + +server: ServerProcess + +# a model name with slashes exercises the query string routing of the stream routes: the id +# cannot travel as a path param because the decoded slash would split it before capture +MODEL = "ggml-org/tinygemma3-GGUF:Q8_0" +STREAM_ID = f"conv-stream-test::{MODEL}" +QS = "conv_id=" + quote(STREAM_ID, safe="") + + +@pytest.fixture(autouse=True) +def create_server(): + global server + server = ServerPreset.router() + + +def test_stream_resume_and_stop_with_slashed_model_name(): + global server + server.start() + + content = "" + for data in server.make_stream_request("POST", "/chat/completions", data={ + "model": MODEL, + "stream": True, + "max_tokens": 16, + "messages": [{"role": "user", "content": "hello"}], + }, headers={"X-Conversation-Id": STREAM_ID}): + if data["choices"]: + content += data["choices"][0]["delta"].get("content") or "" + assert len(content) > 0 + + # the finished session replays from the beginning through the router + res = server.make_request("GET", f"/v1/stream?{QS}&from=0") + assert res.status_code == 200 + assert "data: " in str(res.body) + + # the explicit stop reaches the owning child and evicts the session + res = server.make_request("DELETE", f"/v1/stream?{QS}") + assert res.status_code == 204 + res = server.make_request("GET", f"/v1/stream?{QS}&from=0") + assert res.status_code == 404 + + +def test_stream_stop_during_model_load(): + global server + server.start() + + thread_error: list[ServerError] = [] + thread_done = threading.Event() + + def fire_post(): + try: + for _ in server.make_stream_request("POST", "/chat/completions", data={ + "model": MODEL, + "stream": True, + "max_tokens": 512, + "messages": [{"role": "user", "content": "Count from 1 to 1000."}], + }, headers={"X-Conversation-Id": STREAM_ID}): + pass + except ServerError as e: + thread_error.append(e) + finally: + thread_done.set() + + t = threading.Thread(target=fire_post) + t.start() + + # catch the autoload window, tiny models load fast so poll aggressively + saw_loading = False + deadline = time.time() + 5.0 + while time.time() < deadline and not thread_done.is_set(): + res = server.make_request("GET", "/models") + status = next(m["status"]["value"] for m in res.body["data"] if m["id"] == MODEL) + if status == "loading": + saw_loading = True + break + time.sleep(0.002) + if not saw_loading: + t.join() + pytest.skip("load window too short to be observed on this machine") # ty: ignore[too-many-positional-arguments] + + # a stop during the load cancels the parked request instead of leaving an orphan + res = server.make_request("DELETE", f"/v1/stream?{QS}") + assert res.status_code == 204 + assert thread_done.wait(timeout=60) + t.join() + assert len(thread_error) == 1 + assert thread_error[0].code == 400 + assert "cancelled" in json.dumps(thread_error[0].body) + res = server.make_request("GET", f"/v1/stream?{QS}&from=0") + assert res.status_code == 404 + + +def test_stream_resumes_after_reload_during_model_load(): + global server + server.start() + + # raw socket client so the connection can be dropped mid load like a page reload + body = json.dumps({ + "model": MODEL, + "stream": True, + "max_tokens": 16, + "messages": [{"role": "user", "content": "hello"}], + }) + request = ( + f"POST /v1/chat/completions HTTP/1.1\r\n" + f"Host: {server.server_host}:{server.server_port}\r\n" + f"Content-Type: application/json\r\n" + f"X-Conversation-Id: {STREAM_ID}\r\n" + f"Content-Length: {len(body)}\r\n" + f"Connection: close\r\n\r\n{body}" + ) + sock = socket.create_connection((server.server_host, server.server_port)) + sock.sendall(request.encode()) + + # drop the client while the model loads, poll aggressively to catch the window + saw_loading = False + saw_503 = False + deadline = time.time() + 5.0 + while time.time() < deadline: + res = server.make_request("GET", "/models") + status = next(m["status"]["value"] for m in res.body["data"] if m["id"] == MODEL) + if status == "loading": + saw_loading = True + break + if status == "loaded": + break + time.sleep(0.002) + sock.close() + if not saw_loading: + pytest.skip("load window too short to be observed on this machine") # ty: ignore[too-many-positional-arguments] + + # while the model loads the resume route answers retry later, then the session appears, + # receives the whole generation despite the dead client, and replays from the beginning + deadline = time.time() + 60.0 + replay = None + while time.time() < deadline: + res = server.make_request("GET", f"/v1/stream?{QS}&from=0") + if res.status_code == 503: + saw_503 = True + elif res.status_code == 200 and "data: " in str(res.body): + replay = res + break + time.sleep(0.1) + assert saw_503, "resume during the load did not answer 503" + assert replay is not None, "session never became resumable after the client disconnect" diff --git a/tools/ui/src/lib/components/app/chat/ChatMessages/ChatMessage/ChatMessageAssistant/ChatMessageAssistant.svelte b/tools/ui/src/lib/components/app/chat/ChatMessages/ChatMessage/ChatMessageAssistant/ChatMessageAssistant.svelte index 00578fcf1a..199d75fcec 100644 --- a/tools/ui/src/lib/components/app/chat/ChatMessages/ChatMessage/ChatMessageAssistant/ChatMessageAssistant.svelte +++ b/tools/ui/src/lib/components/app/chat/ChatMessages/ChatMessage/ChatMessageAssistant/ChatMessageAssistant.svelte @@ -10,7 +10,7 @@ } from '$lib/components/app'; import { getMessageEditContext } from '$lib/contexts'; import { useProcessingState } from '$lib/hooks/use-processing-state.svelte'; - import { isLoading, isChatStreaming } from '$lib/stores/chat.svelte'; + import { chatStore, isLoading, isChatStreaming } from '$lib/stores/chat.svelte'; import { modelLoadProgressText } from '$lib/utils'; import { MessageRole } from '$lib/enums'; import { config } from '$lib/stores/settings.svelte'; @@ -82,8 +82,11 @@ let hasNoContent = $derived(!message?.content?.trim()); let isActivelyProcessing = $derived(isCurrentlyLoading || isStreaming); - // during a router auto-load the message has no model yet, so target the selected one - let loadTargetModel = $derived(message.model ?? modelsStore.selectedModelName); + // during a router auto-load the message has no model yet: target the model frozen in the + // persisted stream state (survives a reload), then fall back to the dropdown selection + let loadTargetModel = $derived( + message.model ?? chatStore.getResumeModel(message.convId) ?? modelsStore.selectedModelName + ); let modelLoadProgress = $derived( isRouter && loadTargetModel ? modelsStore.getLoadProgress(loadTargetModel) : null ); diff --git a/tools/ui/src/lib/constants/api-endpoints.ts b/tools/ui/src/lib/constants/api-endpoints.ts index 37137c1c76..ab35708a46 100644 --- a/tools/ui/src/lib/constants/api-endpoints.ts +++ b/tools/ui/src/lib/constants/api-endpoints.ts @@ -21,7 +21,11 @@ export const API_TOOLS = { EXECUTE: '/tools' }; -// resumable stream routes, the conv::model identity is appended as a path segment +// resumable stream routes, the conv::model identity travels as the conv_id query param +// because model names can contain slashes that a path segment cannot carry +// resume retry cadence while the owning model is still loading (server answers 503) +export const STREAM_RESUME_RETRY_MS = 2000; + export const API_STREAM = { BASE: './v1/stream', LOOKUP: './v1/streams/lookup' diff --git a/tools/ui/src/lib/services/chat.service.ts b/tools/ui/src/lib/services/chat.service.ts index d2455614fb..4ce396533d 100644 --- a/tools/ui/src/lib/services/chat.service.ts +++ b/tools/ui/src/lib/services/chat.service.ts @@ -343,6 +343,9 @@ export class ChatService { // model the ::model suffix keeps the per model session distinct if (stream && conversationId) { headers['X-Conversation-Id'] = streamIdentity(conversationId, options.model); + // persist the pending stream before the fetch: a reload during the model load or + // the prompt processing must still find its way back to the session once it exists + ChatService.saveStreamState(conversationId, 0, options.model ?? null); } const response = await fetch(API_CHAT.COMPLETIONS, { @@ -353,6 +356,11 @@ export class ChatService { }); if (!response.ok) { + // a rejected request (including one cancelled by a stop during the model load) + // leaves nothing to resume + if (conversationId) { + ChatService.clearStreamState(conversationId); + } const error = await ChatService.parseErrorResponse(response); if (onError) { @@ -512,7 +520,7 @@ export class ChatService { if (!conversationId) return; try { const id = streamIdentity(conversationId, model); - await fetch(`${API_STREAM.BASE}/${encodeURIComponent(id)}`, { + await fetch(`${API_STREAM.BASE}?conv_id=${encodeURIComponent(id)}`, { method: 'DELETE', headers: getAuthHeaders() }); @@ -605,6 +613,26 @@ export class ChatService { * existing SSE parser drains it like a fresh stream. The server returns 200 on success, 404 if * no session exists for the conv_id, and 400 if the offset is below the dropped prefix. */ + // probe the resume route status without consuming the stream: the SSE route has no HEAD, + // so issue the GET and abort it right after the status line. 0 on network error + static async probeResumeStatus(streamId: string): Promise { + if (!streamId) return 0; + const ac = new AbortController(); + try { + const resp = await fetch( + `${API_STREAM.BASE}?conv_id=${encodeURIComponent(streamId)}&from=0`, + { + headers: getAuthHeaders(), + signal: ac.signal + } + ); + ac.abort(); + return resp.status; + } catch { + return 0; + } + } + static async resumeStream( conversationId: string, signal?: AbortSignal, @@ -614,7 +642,7 @@ export class ChatService { const state = ChatService.getStreamState(conversationId); const from = state?.bytesReceived ?? 0; const id = streamIdentity(conversationId, model); - const url = `${API_STREAM.BASE}/${encodeURIComponent(id)}?from=${from}`; + const url = `${API_STREAM.BASE}?conv_id=${encodeURIComponent(id)}&from=${from}`; return await fetch(url, { method: 'GET', signal, headers: getAuthHeaders() }); } diff --git a/tools/ui/src/lib/stores/chat.svelte.ts b/tools/ui/src/lib/stores/chat.svelte.ts index 5cbfe213b1..222723ab10 100644 --- a/tools/ui/src/lib/stores/chat.svelte.ts +++ b/tools/ui/src/lib/stores/chat.svelte.ts @@ -14,6 +14,7 @@ import { SvelteMap, SvelteSet } from 'svelte/reactivity'; import { DatabaseService } from '$lib/services/database.service'; import { ChatService } from '$lib/services/chat.service'; +import { STREAM_RESUME_RETRY_MS } from '$lib/constants/api-endpoints'; import { streamIdentity } from '$lib/utils/stream-identity'; import { getAuthHeaders } from '$lib/utils/api-headers'; import { CONTENT_TYPE_HEADER } from '$lib/constants'; @@ -78,7 +79,7 @@ class ChatStore { // true while the active conversation streams reasoning content but no visible content yet isReasoning = $state(false); // resumable stream connection state for the active conversation - // streaming -> bytes flowing normally, resuming -> waiting on /v1/stream/:id reconnect, lost -> unrecoverable + // streaming -> bytes flowing normally, resuming -> waiting on /v1/stream reconnect, lost -> unrecoverable streamConnectionState = $state(StreamConnectionState.STREAMING); chatLoadingStates = new SvelteMap(); chatReasoningStates = new SvelteMap(); @@ -94,6 +95,11 @@ class ChatStore { // off when one conv finishes while another is still streaming. mirrors chatLoadingStates // in scope but tracks the attach + tee replay path specifically private attachingConvs = new SvelteSet(); + // pending resume retry timers while an owning model loads, one per conv + private resumeRetryTimers = new SvelteMap>(); + // convs whose resume waits on a model load: their loading state belongs to the retry loop, + // so discoverActiveStream must not treat it as a live send and bail + private resumePendingConvs = new SvelteSet(); // in-flight discoverActiveStream guard, keyed by conv id private discoveringConvs = new SvelteSet(); private abortControllers = new SvelteMap(); @@ -263,7 +269,7 @@ class ChatStore { const id = streamId || streamIdentity(convId, selectedModelName()); let response: Response; try { - response = await fetch(`./v1/stream/${encodeURIComponent(id)}?from=0`, { + response = await fetch(`./v1/stream?conv_id=${encodeURIComponent(id)}&from=0`, { headers: getAuthHeaders() }); } catch (e) { @@ -438,13 +444,22 @@ class ChatStore { } } + /** + * Model frozen at send time for a stream awaiting resume, from the persisted stream state. + * The load progress indicator targets it after a reload, when the message row has no model + * yet and the dropdown selection may not be restored. + */ + getResumeModel(convId: string): string | null { + return ChatService.getStreamState(convId)?.model ?? null; + } + async discoverActiveStream(convId: string): Promise { if (!convId) return; if (this.chatStreamingStates.has(convId)) return; - if (this.chatLoadingStates.get(convId)) return; + if (this.chatLoadingStates.get(convId) && !this.resumePendingConvs.has(convId)) return; // concurrency guard: another discover may already be running for this conv (typical race // between mount and visibilitychange on tab switch). a second concurrent fetch on the same - // /v1/stream/ would duplicate every byte into the DB message, this guard bounces it + // /v1/stream would duplicate every byte into the DB message, this guard bounces it if (this.discoveringConvs.has(convId)) return; this.discoveringConvs.add(convId); @@ -470,6 +485,38 @@ class ChatStore { if (!localState) { return; } + // quiet status probe first: a full attach flips the loading UI on every try, probing + // keeps the retry loop invisible while the owning model is still loading (503) + const status = await ChatService.probeResumeStatus(streamId); + if (status === 503) { + // make the wait visible: the empty assistant row persisted at send time renders + // the processing info, whose model load percentage flows from the models feed + this.resumePendingConvs.add(convId); + this.setChatLoading(convId, true); + if (!this.resumeRetryTimers.has(convId)) { + this.resumeRetryTimers.set( + convId, + setTimeout(() => { + this.resumeRetryTimers.delete(convId); + void this.discoverActiveStream(convId); + }, STREAM_RESUME_RETRY_MS) + ); + } + return; + } + if (this.resumePendingConvs.delete(convId) && status !== 200) { + // the wait is over without a session to attach, drop the visible loading state + this.setChatLoading(convId, false); + } + if (status === 0) { + // transient network failure, the next mount or visibility change retries + return; + } + if (status !== 200) { + // the session is gone (stopped, TTL expired), nothing to resume anymore + ChatService.clearStreamState(convId); + return; + } await this.attachServerStream(convId, streamId); // if attachServerStream failed (session gone, TTL expired), clear the local state to avoid retrying forever if (!this.chatStreamingStates.has(convId) && !this.chatLoadingStates.get(convId)) { @@ -1469,8 +1516,16 @@ class ChatStore { // detached drain keeps producing tokens until eos or max_tokens. use the frozen identity // captured when the session started, not the live dropdown const streamStateForStop = this.chatStreamingStates.get(convId); - const modelForStop = streamStateForStop?.model; + const modelForStop = streamStateForStop?.model ?? ChatService.getStreamState(convId)?.model; void ChatService.cancelServerStream(convId, modelForStop); + // an explicit stop leaves nothing to resume and kills a pending resume retry + ChatService.clearStreamState(convId); + const retryTimer = this.resumeRetryTimers.get(convId); + if (retryTimer !== undefined) { + clearTimeout(retryTimer); + this.resumeRetryTimers.delete(convId); + } + this.resumePendingConvs.delete(convId); this.abortRequest(convId); this.setChatLoading(convId, false); this.clearChatStreaming(convId);