fix(quick-1): prevent 'internal error' from polluting DB and chat history
- 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
This commit is contained in:
1 parent
d9198795bf
commit
44875aa03e
2 files changed
+84
-17
No files matched your search
@@ -7,6 +7,8 @@ use axum::{
|
|||||||
};
|
};
|
||||||
use std::{sync::Arc, time::SystemTime};
|
use std::{sync::Arc, time::SystemTime};
|
||||||
use tokio::sync::Mutex;
|
use tokio::sync::Mutex;
|
||||||
|
use tower::ServiceBuilder;
|
||||||
|
use tower_http::trace::{DefaultOnFailure, TraceLayer};
|
||||||
|
|
||||||
use api_contracts::{
|
use api_contracts::{
|
||||||
chat::{
|
chat::{
|
||||||
@@ -37,11 +39,49 @@ pub struct AppState {
|
|||||||
|
|
||||||
pub fn build_router(state: AppState) -> Router {
|
pub fn build_router(state: AppState) -> Router {
|
||||||
let cors = state.server_config.cors_layer();
|
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()
|
Router::new()
|
||||||
.route("/health", get(health_handler))
|
.route("/health", get(health_handler))
|
||||||
.route("/v1/models", get(models_handler))
|
.route("/v1/models", get(models_handler))
|
||||||
.route("/v1/chat/completions", post(chat_completions_handler))
|
.route("/v1/chat/completions", post(chat_completions_handler))
|
||||||
.layer(cors)
|
.layer(ServiceBuilder::new().layer(trace_layer).layer(cors))
|
||||||
.with_state(state)
|
.with_state(state)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -56,6 +96,20 @@ fn error_response(status: StatusCode, error: ApiError) -> Response {
|
|||||||
(status, Json(ErrorResponse { error })).into_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, R>(f: F) -> tokio::task::JoinHandle<R>
|
||||||
|
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<AppState>) -> impl IntoResponse {
|
async fn health_handler(State(state): State<AppState>) -> impl IntoResponse {
|
||||||
Json(HealthResponse {
|
Json(HealthResponse {
|
||||||
status: "ok",
|
status: "ok",
|
||||||
@@ -67,14 +121,13 @@ async fn health_handler(State(state): State<AppState>) -> impl IntoResponse {
|
|||||||
|
|
||||||
async fn models_handler(State(state): State<AppState>, headers: HeaderMap) -> Response {
|
async fn models_handler(State(state): State<AppState>, headers: HeaderMap) -> Response {
|
||||||
if let Err((status, error)) = authorize_request(&headers, &state.server_config) {
|
if let Err((status, error)) = authorize_request(&headers, &state.server_config) {
|
||||||
|
tracing::warn!(status = %status, "models: authorization failed");
|
||||||
return error_response(status, error);
|
return error_response(status, error);
|
||||||
}
|
}
|
||||||
|
|
||||||
Json(ModelList {
|
let model_list = state.server_config.model_list(unix_now());
|
||||||
object: "list",
|
tracing::info!(model_count = model_list.len(), "models.listed");
|
||||||
data: state.server_config.model_list(unix_now()),
|
Json(ModelList { object: "list", data: model_list }).into_response()
|
||||||
})
|
|
||||||
.into_response()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn chat_completions_handler(
|
async fn chat_completions_handler(
|
||||||
@@ -87,7 +140,13 @@ async fn chat_completions_handler(
|
|||||||
stream = req.stream,
|
stream = req.stream,
|
||||||
messages = req.messages.len(),
|
messages = req.messages.len(),
|
||||||
max_tokens = ?req.max_tokens,
|
max_tokens = ?req.max_tokens,
|
||||||
"chat_completions: request received"
|
temperature = ?req.temperature,
|
||||||
|
auth_present = headers.contains_key(
|
||||||
|
state.server_config.auth_header
|
||||||
|
.parse::<axum::http::HeaderName>()
|
||||||
|
.unwrap_or(axum::http::header::AUTHORIZATION)
|
||||||
|
),
|
||||||
|
"chat_completions.received"
|
||||||
);
|
);
|
||||||
|
|
||||||
if let Err((status, error)) = authorize_request(&headers, &state.server_config) {
|
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::<StreamItem>(256);
|
let (tx, mut rx) = tokio::sync::mpsc::channel::<StreamItem>(256);
|
||||||
let prompt = prompt.clone();
|
let prompt = prompt.clone();
|
||||||
|
|
||||||
tokio::task::spawn_blocking(move || {
|
spawn_blocking_in_span(move || {
|
||||||
let rt = tokio::runtime::Handle::current();
|
let rt = tokio::runtime::Handle::current();
|
||||||
tracing::debug!("stream: acquiring engine lock");
|
tracing::debug!("stream: acquiring engine lock");
|
||||||
let mut engine = rt.block_on(engine_arc.lock());
|
let mut engine = rt.block_on(engine_arc.lock());
|
||||||
@@ -202,8 +261,10 @@ async fn stream_response(
|
|||||||
tracing::info!(
|
tracing::info!(
|
||||||
prompt_tokens = usage.prompt_tokens,
|
prompt_tokens = usage.prompt_tokens,
|
||||||
completion_tokens = usage.completion_tokens,
|
completion_tokens = usage.completion_tokens,
|
||||||
|
total_tokens = usage.total_tokens,
|
||||||
tps = ?usage.tokens_per_second,
|
tps = ?usage.tokens_per_second,
|
||||||
"stream: generation complete"
|
finish_reason = "stop",
|
||||||
|
"stream.generation_complete"
|
||||||
);
|
);
|
||||||
counters_arc.inc_completed();
|
counters_arc.inc_completed();
|
||||||
let _ = rt.block_on(tx.send(StreamItem::Done {
|
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()));
|
yield Ok(Event::default().data(serde_json::to_string(&chunk).unwrap()));
|
||||||
}
|
}
|
||||||
StreamItem::Done { finish_reason, usage } => {
|
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 {
|
let final_chunk = ChatCompletionChunk {
|
||||||
id: id.clone(),
|
id: id.clone(),
|
||||||
object: "chat.completion.chunk",
|
object: "chat.completion.chunk",
|
||||||
@@ -253,7 +316,7 @@ async fn stream_response(
|
|||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
StreamItem::Error => {
|
StreamItem::Error => {
|
||||||
tracing::error!("stream: sending error chunk");
|
tracing::error!(streamed_tokens, "stream.error: session failed — sending empty finish chunk");
|
||||||
let chunk = ChatCompletionChunk {
|
let chunk = ChatCompletionChunk {
|
||||||
id: id.clone(),
|
id: id.clone(),
|
||||||
object: "chat.completion.chunk",
|
object: "chat.completion.chunk",
|
||||||
@@ -261,7 +324,7 @@ async fn stream_response(
|
|||||||
model: resolved_model.clone(),
|
model: resolved_model.clone(),
|
||||||
choices: vec![ChunkChoice {
|
choices: vec![ChunkChoice {
|
||||||
index: 0,
|
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),
|
finish_reason: Some(FinishReason::Stop),
|
||||||
}],
|
}],
|
||||||
usage: None,
|
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(serde_json::to_string(&final_chunk).unwrap()));
|
||||||
yield Ok(Event::default().data("[DONE]"));
|
yield Ok(Event::default().data("[DONE]"));
|
||||||
}
|
}
|
||||||
Err(response) => {
|
Err(_response) => {
|
||||||
|
tracing::error!("stream.buffered: generation failed");
|
||||||
yield Ok(Event::default().data("[DONE]"));
|
yield Ok(Event::default().data("[DONE]"));
|
||||||
let _ = response;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -392,7 +455,7 @@ async fn generate_full_completion(
|
|||||||
req: ChatCompletionRequest,
|
req: ChatCompletionRequest,
|
||||||
) -> Result<GenerationResult, Response> {
|
) -> Result<GenerationResult, Response> {
|
||||||
tracing::debug!(prompt_len = prompt.len(), "generate_full_completion: starting");
|
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();
|
let rt = tokio::runtime::Handle::current();
|
||||||
tracing::debug!("generate_full_completion: acquiring engine lock");
|
tracing::debug!("generate_full_completion: acquiring engine lock");
|
||||||
let mut engine = rt.block_on(engine_arc.lock());
|
let mut engine = rt.block_on(engine_arc.lock());
|
||||||
|
|||||||
@@ -126,7 +126,7 @@ export default function ChatView({ db, chatId, server, apiKey, model, onChatCrea
|
|||||||
const buildApiMessages = useCallback((cid: number) => {
|
const buildApiMessages = useCallback((cid: number) => {
|
||||||
const limit = useMemory ? MAX_MEMORY : 1;
|
const limit = useMemory ? MAX_MEMORY : 1;
|
||||||
return db.query<{ role: string; content: string }>(
|
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]
|
[cid, limit]
|
||||||
).reverse();
|
).reverse();
|
||||||
}, [db, useMemory]);
|
}, [db, useMemory]);
|
||||||
@@ -234,9 +234,12 @@ export default function ChatView({ db, chatId, server, apiKey, model, onChatCrea
|
|||||||
if (parsed.reasoning) setShowLiveReasoning(true);
|
if (parsed.reasoning) setShowLiveReasoning(true);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Save assistant message
|
// Save assistant message — only if there is real content to persist
|
||||||
const parsed = parseAssistantResponse(fullContent);
|
const parsed = parseAssistantResponse(fullContent);
|
||||||
const finalContent = parsed.content.trim() || fullContent.trim();
|
const finalContent = parsed.content.trim() || fullContent.trim();
|
||||||
|
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);
|
console.log(`[ChatView] send: saving assistant message — finalContent.length=${finalContent.length}, reasoning.length=${parsed.reasoning?.length ?? 0}, usage=`, usage);
|
||||||
db.run(
|
db.run(
|
||||||
`INSERT INTO messages (
|
`INSERT INTO messages (
|
||||||
@@ -252,6 +255,7 @@ export default function ChatView({ db, chatId, server, apiKey, model, onChatCrea
|
|||||||
usage?.tokens_per_second ?? null,
|
usage?.tokens_per_second ?? null,
|
||||||
],
|
],
|
||||||
);
|
);
|
||||||
|
}
|
||||||
db.run("UPDATE chats SET updated_at = strftime('%s','now') WHERE id = ?", [cid]);
|
db.run("UPDATE chats SET updated_at = strftime('%s','now') WHERE id = ?", [cid]);
|
||||||
if (usage?.tokens_per_second) setTps(usage.tokens_per_second);
|
if (usage?.tokens_per_second) setTps(usage.tokens_per_second);
|
||||||
onChatUpdated();
|
onChatUpdated();
|
||||||
|
|||||||
Reference in new issue
Block a user