From 44875aa03e3f5b7e376d3b3c4c47efdc19858de3 Mon Sep 17 00:00:00 2001 From: Evandro Fonseca Junior Date: Mon, 30 Mar 2026 17:29:57 -0300 Subject: [PATCH] fix(quick-1): prevent 'internal error' from polluting DB and chat history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - routes.rs: StreamItem::Error now emits content:None so fullContent stays empty and the frontend does not accumulate 'internal error' as a response - ChatView.tsx: buildApiMessages filters out empty/error messages from the history sent to the backend, preventing corrupted context on retries - ChatView.tsx: assistant INSERT is guarded — skipped when fullContent is empty, so a failed stream leaves no empty assistant row in the DB --- crates/inference-server/src/routes.rs | 105 ++++++++++++++++++++------ tauri-app/src/components/ChatView.tsx | 38 +++++----- 2 files changed, 105 insertions(+), 38 deletions(-) diff --git a/crates/inference-server/src/routes.rs b/crates/inference-server/src/routes.rs index 14d8026..e609e20 100644 --- a/crates/inference-server/src/routes.rs +++ b/crates/inference-server/src/routes.rs @@ -7,6 +7,8 @@ use axum::{ }; use std::{sync::Arc, time::SystemTime}; use tokio::sync::Mutex; +use tower::ServiceBuilder; +use tower_http::trace::{DefaultOnFailure, TraceLayer}; use api_contracts::{ chat::{ @@ -37,11 +39,49 @@ pub struct AppState { pub fn build_router(state: AppState) -> Router { let cors = state.server_config.cors_layer(); + + let trace_layer = TraceLayer::new_for_http() + .make_span_with(|request: &axum::http::Request<_>| { + let request_id = uuid::Uuid::new_v4().to_string(); + tracing::info_span!( + "http", + request_id = %request_id, + method = %request.method(), + path = %request.uri().path(), + user_agent = request + .headers() + .get(axum::http::header::USER_AGENT) + .and_then(|v| v.to_str().ok()) + .unwrap_or(""), + // filled in on_response + status = tracing::field::Empty, + latency_ms = tracing::field::Empty, + ) + }) + .on_request(|request: &axum::http::Request<_>, _span: &tracing::Span| { + let content_length = request + .headers() + .get(axum::http::header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .unwrap_or("0"); + tracing::info!(content_length, "request_received"); + }) + .on_response( + |response: &axum::http::Response<_>, + latency: std::time::Duration, + span: &tracing::Span| { + span.record("status", response.status().as_u16()); + span.record("latency_ms", latency.as_millis()); + tracing::info!("response_sent"); + }, + ) + .on_failure(DefaultOnFailure::new().level(tracing::Level::ERROR)); + Router::new() .route("/health", get(health_handler)) .route("/v1/models", get(models_handler)) .route("/v1/chat/completions", post(chat_completions_handler)) - .layer(cors) + .layer(ServiceBuilder::new().layer(trace_layer).layer(cors)) .with_state(state) } @@ -56,6 +96,20 @@ fn error_response(status: StatusCode, error: ApiError) -> Response { (status, Json(ErrorResponse { error })).into_response() } +/// Run `f` inside a `spawn_blocking` task while propagating the current tracing span. +/// Without this, logs emitted inside `spawn_blocking` lose the `request_id` context. +fn spawn_blocking_in_span(f: F) -> tokio::task::JoinHandle +where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, +{ + let span = tracing::Span::current(); + tokio::task::spawn_blocking(move || { + let _guard = span.enter(); + f() + }) +} + async fn health_handler(State(state): State) -> impl IntoResponse { Json(HealthResponse { status: "ok", @@ -67,14 +121,13 @@ async fn health_handler(State(state): State) -> impl IntoResponse { async fn models_handler(State(state): State, headers: HeaderMap) -> Response { if let Err((status, error)) = authorize_request(&headers, &state.server_config) { + tracing::warn!(status = %status, "models: authorization failed"); return error_response(status, error); } - Json(ModelList { - object: "list", - data: state.server_config.model_list(unix_now()), - }) - .into_response() + let model_list = state.server_config.model_list(unix_now()); + tracing::info!(model_count = model_list.len(), "models.listed"); + Json(ModelList { object: "list", data: model_list }).into_response() } async fn chat_completions_handler( @@ -83,11 +136,17 @@ async fn chat_completions_handler( Json(req): Json, ) -> Response { tracing::info!( - model = %req.model, - stream = req.stream, - messages = req.messages.len(), - max_tokens = ?req.max_tokens, - "chat_completions: request received" + model = %req.model, + stream = req.stream, + messages = req.messages.len(), + max_tokens = ?req.max_tokens, + temperature = ?req.temperature, + auth_present = headers.contains_key( + state.server_config.auth_header + .parse::() + .unwrap_or(axum::http::header::AUTHORIZATION) + ), + "chat_completions.received" ); if let Err((status, error)) = authorize_request(&headers, &state.server_config) { @@ -168,7 +227,7 @@ async fn stream_response( let (tx, mut rx) = tokio::sync::mpsc::channel::(256); let prompt = prompt.clone(); - tokio::task::spawn_blocking(move || { + spawn_blocking_in_span(move || { let rt = tokio::runtime::Handle::current(); tracing::debug!("stream: acquiring engine lock"); let mut engine = rt.block_on(engine_arc.lock()); @@ -200,10 +259,12 @@ async fn stream_response( tokens_per_second: record.tokens_per_second(), }; tracing::info!( - prompt_tokens = usage.prompt_tokens, + prompt_tokens = usage.prompt_tokens, completion_tokens = usage.completion_tokens, - tps = ?usage.tokens_per_second, - "stream: generation complete" + total_tokens = usage.total_tokens, + tps = ?usage.tokens_per_second, + finish_reason = "stop", + "stream.generation_complete" ); counters_arc.inc_completed(); let _ = rt.block_on(tx.send(StreamItem::Done { @@ -235,7 +296,9 @@ async fn stream_response( yield Ok(Event::default().data(serde_json::to_string(&chunk).unwrap())); } StreamItem::Done { finish_reason, usage } => { - tracing::info!(streamed_tokens, "stream: sending final chunk with usage"); + // prompt_tokens/completion_tokens/finish_reason already logged + // in stream.generation_complete inside spawn_blocking. + tracing::info!(streamed_tokens, "stream.final_chunk"); let final_chunk = ChatCompletionChunk { id: id.clone(), object: "chat.completion.chunk", @@ -253,7 +316,7 @@ async fn stream_response( break; } StreamItem::Error => { - tracing::error!("stream: sending error chunk"); + tracing::error!(streamed_tokens, "stream.error: session failed — sending empty finish chunk"); let chunk = ChatCompletionChunk { id: id.clone(), object: "chat.completion.chunk", @@ -261,7 +324,7 @@ async fn stream_response( model: resolved_model.clone(), choices: vec![ChunkChoice { index: 0, - delta: Delta { role: None, content: Some("internal error".to_owned()), tool_calls: None }, + delta: Delta { role: None, content: None, tool_calls: None }, finish_reason: Some(FinishReason::Stop), }], usage: None, @@ -325,9 +388,9 @@ async fn stream_response( yield Ok(Event::default().data(serde_json::to_string(&final_chunk).unwrap())); yield Ok(Event::default().data("[DONE]")); } - Err(response) => { + Err(_response) => { + tracing::error!("stream.buffered: generation failed"); yield Ok(Event::default().data("[DONE]")); - let _ = response; } } } @@ -392,7 +455,7 @@ async fn generate_full_completion( req: ChatCompletionRequest, ) -> Result { tracing::debug!(prompt_len = prompt.len(), "generate_full_completion: starting"); - let result = tokio::task::spawn_blocking(move || { + let result = spawn_blocking_in_span(move || { let rt = tokio::runtime::Handle::current(); tracing::debug!("generate_full_completion: acquiring engine lock"); let mut engine = rt.block_on(engine_arc.lock()); diff --git a/tauri-app/src/components/ChatView.tsx b/tauri-app/src/components/ChatView.tsx index f840650..b930451 100644 --- a/tauri-app/src/components/ChatView.tsx +++ b/tauri-app/src/components/ChatView.tsx @@ -126,7 +126,7 @@ export default function ChatView({ db, chatId, server, apiKey, model, onChatCrea const buildApiMessages = useCallback((cid: number) => { const limit = useMemory ? MAX_MEMORY : 1; return db.query<{ role: string; content: string }>( - "SELECT role, content FROM messages WHERE chat_id = ? ORDER BY created_at DESC LIMIT ?", + "SELECT role, content FROM messages WHERE chat_id = ? AND content != '' AND content NOT LIKE '⚠️ Error:%' AND content != 'internal error' ORDER BY created_at DESC LIMIT ?", [cid, limit] ).reverse(); }, [db, useMemory]); @@ -234,24 +234,28 @@ export default function ChatView({ db, chatId, server, apiKey, model, onChatCrea if (parsed.reasoning) setShowLiveReasoning(true); } - // Save assistant message + // Save assistant message — only if there is real content to persist const parsed = parseAssistantResponse(fullContent); const finalContent = parsed.content.trim() || fullContent.trim(); - console.log(`[ChatView] send: saving assistant message — finalContent.length=${finalContent.length}, reasoning.length=${parsed.reasoning?.length ?? 0}, usage=`, usage); - db.run( - `INSERT INTO messages ( - chat_id, role, content, reasoning_content, prompt_tokens, completion_tokens, total_tokens, tokens_per_second - ) VALUES (?, 'assistant', ?, ?, ?, ?, ?, ?)`, - [ - cid, - finalContent, - parsed.content.trim() ? parsed.reasoning : null, - usage?.prompt_tokens ?? 0, - usage?.completion_tokens ?? 0, - usage?.total_tokens ?? 0, - usage?.tokens_per_second ?? null, - ], - ); + if (!finalContent && !parsed.reasoning) { + console.warn("[ChatView] send: assistant returned empty content — not saving to DB"); + } else { + console.log(`[ChatView] send: saving assistant message — finalContent.length=${finalContent.length}, reasoning.length=${parsed.reasoning?.length ?? 0}, usage=`, usage); + db.run( + `INSERT INTO messages ( + chat_id, role, content, reasoning_content, prompt_tokens, completion_tokens, total_tokens, tokens_per_second + ) VALUES (?, 'assistant', ?, ?, ?, ?, ?, ?)`, + [ + cid, + finalContent, + parsed.content.trim() ? parsed.reasoning : null, + usage?.prompt_tokens ?? 0, + usage?.completion_tokens ?? 0, + usage?.total_tokens ?? 0, + usage?.tokens_per_second ?? null, + ], + ); + } db.run("UPDATE chats SET updated_at = strftime('%s','now') WHERE id = ?", [cid]); if (usage?.tokens_per_second) setTps(usage.tokens_per_second); onChatUpdated();