server + ui: fix stream routes for model names containing a slash (#26137)
* server + ui: refactor resumable stream routes to query string conv_id The conversation id can embed a model name containing slashes (ggml-org/...) in router mode, which the decoded path splits before the :conv_id param is captured, so stop and resume never matched the session. Move the id to the conv_id query string on the public routes and on the internal router -> child hop, where slashes survive encoding. Handlers are unchanged since query and path params land in the same map. Add a regression test with a slashed model name. * server: move stream route docs to server-stream.h Address review: ngxson wants the main server.cpp registration code kept clean and simple, with route-level explanations living in the header. Move the query string rationale and the lookup ownership note next to the handler declarations in server-stream.h, and shorten the wiring comment to a pointer. * server: cancel a pending request when its stream is stopped during model load The conversation was registered in the conv map only after the blocking autoload wait, so a stop issued while the model loaded found nothing to cancel and the request went on to generate an orphan once the load ended. Register the conversation before the wait and give the entry a ticket: a stop erases the entry, and the parked request checks its ticket after the wait and aborts with 400 instead of starting. A newer request on the same conversation replaces the entry, so only the stopped request is cancelled. Add a regression test that stops during the load window. * server + ui: resume a stream after a page reload during model load A pending request died with the client socket when the page was reloaded while its model was loading, so no session ever existed and the conversation had nothing to recover. A session request that waited for a load now detaches from the client socket and reaches the child regardless, the session buffer receives the generation, and the resume route answers 503 while the owner is loading so the client retries instead of dropping its state. The WebUI persists the pending stream at send time, quietly polls on 503, and attaches once the session exists. Add a regression test that drops the client during the load window. * ui: show the model load progress again after a page refresh The resume wait was invisible, so a conversation refreshed while its model was loading showed nothing until the first byte. On a 503 from the resume probe, mark the conversation as loading again so the assistant row persisted at send time renders the processing info, and target the model frozen in the persisted stream state for the progress, since the row has no model yet and the dropdown may not be restored. * fix CI * fix CI bis
This commit is contained in:
+6
-3
@@ -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
|
||||
);
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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<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,
|
||||
@@ -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() });
|
||||
}
|
||||
|
||||
|
||||
@@ -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>(StreamConnectionState.STREAMING);
|
||||
chatLoadingStates = new SvelteMap<string, boolean>();
|
||||
chatReasoningStates = new SvelteMap<string, boolean>();
|
||||
@@ -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<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>();
|
||||
@@ -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<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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user