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/<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.
+Consumer side: `GET /v1/stream?conv_id=<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=<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=<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.
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.
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");
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<bool()>([]() { return false; }) : req.should_stop,
base_params.timeout_read,
base_params.timeout_write
);
}
// 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;
}
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);
};
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) {
};
this->router_stream_get = [this](const server_http_req & req) {
- // GET /v1/stream/<conv_id>?from=N. resolve the owning child from the conv_id -> model
+ // GET /v1/stream?conv_id=<id>&from=N. resolve the owning child from the conv_id -> model
// map, 404 when nothing maps
auto res = std::make_unique<server_http_res>();
std::string conv_id = req.get_param("conv_id");
}
std::optional<server_model_meta> 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());
};
this->router_stream_delete = [this](const server_http_req & req) {
- // DELETE /v1/stream/<conv_id>. resolve the owning child via the map and forward only to
+ // DELETE /v1/stream?conv_id=<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<server_http_res>();
std::string conv_id = req.get_param("conv_id");
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);
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());
// 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<std::mutex> 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<std::mutex> lock(mu);
+ auto it = map.find(conv_id);
+ return it != map.end() && it->second.ticket == ticket;
}
std::optional<std::string> lookup(const std::string & conv_id) {
if (it == map.end()) {
return std::nullopt;
}
- return it->second;
+ return it->second.model;
}
void forget(const std::string & conv_id) {
}
private:
- std::mutex mu;
- std::unordered_map<std::string, std::string> map;
+ struct entry_t {
+ std::string model;
+ uint64_t ticket;
+ };
+ std::mutex mu;
+ uint64_t next_ticket = 1;
+ std::unordered_map<std::string, entry_t> map;
};
common_preset_context ctx_preset;
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
server_http_context::handler_t server_stream_make_get_handler() {
return [](const server_http_req & req) -> server_http_res_ptr {
- // GET /v1/stream/<conv_id>?from=N replays buffered SSE bytes then blocks for live
+ // GET /v1/stream?conv_id=<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()) {
server_http_context::handler_t server_stream_make_delete_handler() {
return [](const server_http_req & req) -> server_http_res_ptr {
- // DELETE /v1/stream/<conv_id> is the explicit user Stop, cancels the producer and evicts
+ // DELETE /v1/stream?conv_id=<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<server_http_res>();
res->status = 204;
bool server_res_spipe::should_stop() {
if (spipe) {
- // note: if DELETE /v1/stream/<conv_id> 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();
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();
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;
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();
--- /dev/null
+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"
} 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';
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
);
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'
// 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, {
});
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) {
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()
});
* 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<number> {
+ 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,
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() });
}
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';
// 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>(StreamConnectionState.STREAMING);
chatLoadingStates = new SvelteMap<string, boolean>();
chatReasoningStates = new SvelteMap<string, boolean>();
// 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<string>();
+ // pending resume retry timers while an owning model loads, one per conv
+ private resumeRetryTimers = new SvelteMap<string, ReturnType<typeof setTimeout>>();
+ // 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<string>();
// in-flight discoverActiveStream guard, keyed by conv id
private discoveringConvs = new SvelteSet<string>();
private abortControllers = new SvelteMap<string, AbortController>();
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) {
}
}
+ /**
+ * 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<void> {
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/<id> 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);
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)) {
// 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);
overview / pkg / ggml / sources / llama.cpp / commitdiff