From a1b7c29338c372bfeed8d7c019641fdac2df85b7 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 17 Jun 2026 13:14:03 +0530 Subject: [PATCH] fix(chat): classify resume-on-reopen errors + de-poison thread (#3714) Classifies resume-on-reopen failures (session_expired / network / poisoned-history) instead of the generic catch-all, and evicts a 400-poisoned warm session so the thread self-heals. Closes #3714. --- app/src/services/chatService.ts | 2 + .../channels/providers/web/run_task.rs | 154 +++++++++++++++++- .../channels/providers/web_errors.rs | 142 +++++++++++++++- src/openhuman/channels/providers/web_tests.rs | 135 ++++++++++++++- ...els_provider_leftovers_raw_coverage_e2e.rs | 4 +- 5 files changed, 420 insertions(+), 17 deletions(-) diff --git a/app/src/services/chatService.ts b/app/src/services/chatService.ts index dfaed7ef0..8b067d67f 100644 --- a/app/src/services/chatService.ts +++ b/app/src/services/chatService.ts @@ -99,10 +99,12 @@ export interface ChatErrorEvent { | 'cancelled' | 'rate_limited' | 'auth_error' + | 'session_expired' | 'provider_error' | 'context_overflow' | 'model_unavailable' | 'payload_too_large' + | 'provider_request_rejected' | 'budget_exhausted'; round: number | null; } diff --git a/src/openhuman/channels/providers/web/run_task.rs b/src/openhuman/channels/providers/web/run_task.rs index 0d65f39da..47deee5b0 100644 --- a/src/openhuman/channels/providers/web/run_task.rs +++ b/src/openhuman/channels/providers/web/run_task.rs @@ -13,7 +13,8 @@ use super::session::{ use super::types::SessionEntry; use super::types::{ChatRequestMetadata, WebChatTaskResult}; use super::web_errors::{ - inference_budget_exceeded_user_message, is_inference_budget_exceeded_error, + classify_inference_error, inference_budget_exceeded_user_message, + is_inference_budget_exceeded_error, }; #[cfg(any(test, debug_assertions))] @@ -301,15 +302,150 @@ pub(crate) async fn run_chat_task( // Only the primary (non-fork) turn writes its agent back to the shared // cache; a fork is fully isolated and lets its agent drop here. if !fork { - let mut sessions = THREAD_SESSIONS.lock().await; - sessions.insert( - map_key, - SessionEntry { - agent, - fingerprint: current_fp, - }, - ); + // De-poison guard. A `provider_request_rejected` outcome means the + // provider could not parse THIS turn's request — an orphaned + // `tool_calls` round-trip, an empty `tool_call_id`, or a reasoning + // echo it rejects. For the managed backend that rejection arrives as an + // in-stream `event: error` SSE frame carrying `errorCode:"BAD_REQUEST"` + // (the response already flushed HTTP 200), NOT an HTTP 400 — so we key + // off the classified type, not a status code. Re-caching this agent + // would replay the identical malformed history on every later turn, + // dead-ending the thread. Drop it instead (the entry was already + // removed from the map at the top of this fn): the next turn cold-boots + // and reseeds from the plain-text conversation log, which is + // structurally incapable of carrying tool malformation + // (`seed_resume_from_messages` rebuilds only system/user/assistant + // text). Transient failures (rate-limit / timeout / 5xx) keep the warm + // session so the user can retry this turn with context intact. + if turn_result_poisoned_session(&result) { + log::warn!( + "[web-channel] dropping session agent after provider_request_rejected — \ + next turn cold-boots from the conversation log (de-poison) \ + client={} thread={} request_id={}", + client_id, + thread_id, + request_id + ); + } else { + let mut sessions = THREAD_SESSIONS.lock().await; + sessions.insert( + map_key, + SessionEntry { + agent, + fingerprint: current_fp, + }, + ); + } } result } + +/// Whether a completed turn's session agent must be **dropped** rather than +/// cached back, because its in-memory history would replay a provider request +/// rejection on every subsequent turn. +/// +/// True only for a *retryable* `provider_request_rejected` — i.e. the +/// poisoned-history case. The copy-split in `web_errors.rs` marks a tool-ordering +/// rejection (orphaned / mismatched `tool_call_id` — for the managed backend an +/// in-stream SSE `event: error` frame stamped `errorCode:"BAD_REQUEST"`) +/// `retryable: true` because the de-poison makes "send it again" true, while a +/// genuine model/parameter 400 stays `retryable: false`. Gating on `&& retryable` +/// therefore evicts ONLY the poisoned session: a non-retryable param 400 keeps +/// its warm session (no needless reseed), exactly like successes and transient +/// failures (rate-limit, timeout, 5xx, session-expiry). +fn turn_result_poisoned_session(result: &Result) -> bool { + matches!( + result, + Err(err) if { + let classified = classify_inference_error(err); + classified.error_type == "provider_request_rejected" && classified.retryable + } + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ok() -> Result { + Ok(WebChatTaskResult { + full_response: "hello".to_string(), + citations: Vec::new(), + }) + } + + #[test] + fn poisoned_on_managed_sse_bad_request_frame() { + // Managed backend 400: flushed HTTP 200, then an in-stream SSE error + // frame stamped errorCode:"BAD_REQUEST" — the exact shape the de-poison + // guard must catch (no HTTP 400 status anywhere in the string). Payload + // mirrors the real backend frame verified against tinyhumansai/backend + // upstream/develop `routes/inference.ts::writeInferenceSSE` + // ({error:{message,type:"stream_error",errorCode}}), wrapped by the + // client's `sse_error_frame_bail_message` as + // "OpenHuman streaming API error: ". `validateToolMessageOrdering` + // throws BadRequestError (errorCode=BAD_REQUEST) for an orphaned tool_call_id. + let err: Result = Err( + "OpenHuman streaming API error: {\"error\":{\"message\":\"Message has tool role, \ + but there was no previous assistant message with a tool call!\",\ + \"type\":\"stream_error\",\"errorCode\":\"BAD_REQUEST\"}}" + .to_string(), + ); + assert!(turn_result_poisoned_session(&err)); + } + + #[test] + fn poisoned_on_byo_provider_tool_ordering_400() { + // BYO/direct provider tool-ordering rejection — classifies as a + // *retryable* provider_request_rejected (poisoned history), so it evicts. + let err: Result = Err( + "OpenAI API error (400 Bad Request): {\"error\":{\"message\":\"Invalid parameter: \ + messages with role 'tool' must be a response to a preceding message with \ + 'tool_calls'.\"}}" + .to_string(), + ); + assert!(turn_result_poisoned_session(&err)); + } + + #[test] + fn genuine_param_400_keeps_warm_session() { + // A non-poisoning model/parameter 400 is a *non-retryable* + // provider_request_rejected — narrowing on `&& retryable` must keep its + // warm session (resending the same params won't help; no reseed needed). + let err: Result = Err( + "custom_openai API error (400 Bad Request): {\"error\":{\"message\":\ + \"Unsupported value: 'temperature' must be 1 for this model\"}}" + .to_string(), + ); + assert!( + !turn_result_poisoned_session(&err), + "non-retryable param 400 is not poisoned history — keep warm session" + ); + } + + #[test] + fn transient_failures_keep_warm_session() { + for raw in [ + // rate limit / 429 — history is fine, user should retry warm + "OpenAI API error (429 Too Many Requests): slow down", + // timeout + "request timed out while reading response", + // upstream 5xx + "OpenAI API error (503 Service Unavailable): no healthy upstream", + // session expiry — not a payload problem + "SESSION_EXPIRED: backend session not active — sign in to resume LLM work", + ] { + let err: Result = Err(raw.to_string()); + assert!( + !turn_result_poisoned_session(&err), + "transient/non-payload error must keep warm session: {raw}" + ); + } + } + + #[test] + fn success_keeps_warm_session() { + assert!(!turn_result_poisoned_session(&ok())); + } +} diff --git a/src/openhuman/channels/providers/web_errors.rs b/src/openhuman/channels/providers/web_errors.rs index 4eb2b28dc..5a9bd0bc7 100644 --- a/src/openhuman/channels/providers/web_errors.rs +++ b/src/openhuman/channels/providers/web_errors.rs @@ -406,13 +406,30 @@ fn classify_by_backend_error_code( fallback_available: None, }, BackendErrorCode::BadRequest => { - // Same code, two shapes (B8/F8): a backend-flagged *malformed* + // Same code, three shapes. FIRST: a tool-ordering rejection + // (`validateToolMessageOrdering` — an orphaned `role:'tool'` message + // with no matching assistant `tool_call`) is *poisoned history*, not + // a model/param problem. The de-poison guard in `run_task.rs` has + // already evicted the offending warm session by the time this copy + // is built, so the next turn cold-boots clean — tell the user + // exactly that (and mark retryable, because resending now works). + if is_malformed_tool_history_text(&err.to_lowercase()) { + ClassifiedError { + error_type: "provider_request_rejected", + message: malformed_history_user_message().to_string(), + source: "provider", + retryable: true, + retry_after_ms: None, + provider, + fallback_available: None, + } + // Else two shapes (B8/F8): a backend-flagged *malformed* // payload is a client bug (the request was built wrong — it pages // Sentry at the FE layer, gated elsewhere), while a plain // user-parameter rejection is a model/param mismatch the user can // fix. The copy differs: don't tell the user to abandon the thread // for a one-off malformation (only this turn failed). - if body_flags_malformed(err) { + } else if body_flags_malformed(err) { ClassifiedError { error_type: "provider_request_rejected", message: "Something went wrong with this message. Try rephrasing it — \ @@ -488,7 +505,32 @@ pub(crate) fn classify_inference_error(err: &str) -> ClassifiedError { // before the generic provider-429 branch — otherwise users see // a confusing "your AI provider is rate-limiting you" message // for limits OpenHuman itself enforced (issue #2364). - let classified = if is_action_budget_exhausted(&lower) { + let classified = if crate::core::observability::is_session_expired_message(err) { + // The OpenHuman app-session JWT expired (or the scheduler gate flagged + // signed-out / `SESSION_EXPIRED` sentinel). There is NO client-side + // refresh — recovery is an interactive re-auth only — so this is + // non-retryable and must route the user to sign-in. Checked FIRST so the + // `auth_error` arm below can't claim the backend's `401 "Invalid token"` + // envelope (it contains "401") and mislead managed-backend users with + // "check your API key". `is_session_expired_message` is conjunctively + // scoped to the OpenHuman/Embedding "Invalid token" envelopes + the + // `SESSION_EXPIRED` / "no backend session" / "session jwt required" + // sentinels, so a BYO provider's own 401 still falls through to + // `auth_error`. + ClassifiedError { + error_type: "session_expired", + message: "Your OpenHuman session expired while the app was idle. \ + Please sign in again to resume." + .to_string(), + source: "auth", + retryable: false, + retry_after_ms: None, + // OpenHuman's own session — provider name (if any leaked into the + // surrounding chain) is irrelevant to a sign-in prompt. + provider: None, + fallback_available: None, + } + } else if is_action_budget_exhausted(&lower) { ClassifiedError { error_type: "action_budget_exceeded", message: with_provider_detail( @@ -725,6 +767,21 @@ pub(crate) fn classify_inference_error(err: &str) -> ClassifiedError { provider: None, fallback_available: None, } + } else if is_provider_request_rejected_text(&lower) && is_malformed_tool_history_text(&lower) { + // Same poisoned-history rejection as the managed `BAD_REQUEST` branch, + // but on a BYO/direct provider (e.g. OpenAI "messages with role 'tool' + // must be a response to a preceding message with 'tool_calls'"). The + // de-poison guard already evicted the warm session, so resending works. + // Checked BEFORE the generic 4xx arm so the actionable copy wins. + ClassifiedError { + error_type: "provider_request_rejected", + message: malformed_history_user_message().to_string(), + source: "provider", + retryable: true, + retry_after_ms: None, + provider, + fallback_available, + } } else if is_provider_request_rejected_text(&lower) { // A provider rejected the request with a 4xx that none of the // specific arms above claimed (generic 400 Bad Request, 404, 422). @@ -749,6 +806,32 @@ pub(crate) fn classify_inference_error(err: &str) -> ClassifiedError { provider, fallback_available, } + } else if is_connection_dropped_text(&lower) { + // A transport-level drop with no provider status and no managed + // `errorCode`: a stale keep-alive socket reused after sleep/wake, a + // network change, or a RAW mid-stream SSE drop — the managed backend + // intentionally omits `errorCode` for raw upstream/network drops + // (backend `routes/inference.ts`), so those reach here as + // `"OpenHuman streaming API error: "` with nothing to branch on. + // These previously fell to the generic catch-all ("Something went + // wrong"). The turn's history is NOT poisoned — the agent loop bails + // before committing the failed iteration (`engine/core.rs`) — so this is + // cleanly retryable and the warm session is kept. Placed LAST so every + // specific provider-status / 4xx arm claims its shape first; only an + // otherwise-unclassified transport error lands here. + ClassifiedError { + error_type: "network", + message: with_provider_detail( + "The connection to the AI service dropped mid-response — usually a \ + sleep/wake or network change. Please try again.", + err, + ), + source: "transport", + retryable: true, + retry_after_ms: None, + provider, + fallback_available, + } } else { ClassifiedError { error_type: "inference", @@ -804,6 +887,59 @@ pub(crate) fn is_empty_provider_response_text(lower: &str) -> bool { /// reach this predicate. /// /// Caller passes the already-lowercased error string. +/// User-facing copy for a poisoned-history 400 (orphaned tool message). The +/// de-poison guard (`run_task.rs`) has already evicted the offending warm +/// session by the time this is shown, so "send it again" is literally true. +pub(crate) fn malformed_history_user_message() -> &'static str { + "We hit a temporary glitch in this conversation — we've cleared it. \ + Please send your message again." +} + +/// Detect a malformed tool-history rejection (orphaned / mismatched +/// `role:'tool'` message). This is the *poisoned history* shape the de-poison +/// guard recovers from — NOT a model/parameter mismatch — so it earns the +/// "we cleared it, resend" copy instead of "try a different model". +/// +/// Anchored on the managed backend's `validateToolMessageOrdering` strings +/// (verified against tinyhumansai/backend `chatCompletions.ts` — "role 'tool' … +/// matching tool_call", "does not match any tool_call from the preceding +/// assistant message"), the raw upstream jinja variant ("tool role … no +/// previous assistant message with a tool call"), and the equivalent BYO +/// provider phrasings. Caller passes the already-lowercased error string. +pub(crate) fn is_malformed_tool_history_text(lower: &str) -> bool { + let tool_role = lower.contains("role 'tool'") || lower.contains("tool role"); + let about_tool_call = lower.contains("tool call") || lower.contains("tool_call"); + (tool_role && about_tool_call) + || lower.contains("does not match any tool_call from the preceding assistant message") +} + +/// Detect a transport-level connection drop with no provider status / managed +/// `errorCode` — the residue that otherwise falls to the generic `inference` +/// catch-all (issue #3714 bucket #1). +/// +/// Anchored on the canonical reqwest/hyper shapes for a severed or never-opened +/// connection (stale keep-alive reused after sleep/wake, network change, raw +/// mid-stream SSE drop). Intentionally does NOT match `"timed out"` (the +/// dedicated `timeout` arm owns that) nor any `4xx/5xx` status (those arms claim +/// their shapes earlier). Caller passes the already-lowercased error string. +pub(crate) fn is_connection_dropped_text(lower: &str) -> bool { + const DROP_MARKERS: &[&str] = &[ + "connection closed before message completed", // hyper IncompleteMessage + "error reading a body from connection", + "connection reset", + "connection refused", + "connection aborted", + "broken pipe", + "unexpected end of file", + "unexpected eof", + "error sending request", + "tcp connect error", + "dns error", + "failed to lookup address", + ]; + DROP_MARKERS.iter().any(|marker| lower.contains(marker)) +} + pub(crate) fn is_provider_request_rejected_text(lower: &str) -> bool { // Match only when the 4xx status appears inside a provider error envelope // (` API error (4xx …)`, emitted by diff --git a/src/openhuman/channels/providers/web_tests.rs b/src/openhuman/channels/providers/web_tests.rs index 99f6ce9bb..34310f32d 100644 --- a/src/openhuman/channels/providers/web_tests.rs +++ b/src/openhuman/channels/providers/web_tests.rs @@ -135,7 +135,6 @@ async fn start_chat_emits_sanitized_chat_error_on_inference_failure() { .await .expect("start_chat should accept valid request"); - let expected = generic_inference_error_user_message().to_string(); let recv = timeout(Duration::from_secs(20), async move { loop { let event = rx.recv().await.expect("event stream should stay open"); @@ -151,11 +150,16 @@ async fn start_chat_emits_sanitized_chat_error_on_inference_failure() { .await .expect("expected chat_error event for started chat request"); + // #3714: "error sending request for url …" is a transport drop, now classified + // by the dedicated `network` arm (was the generic catch-all). The key property + // this test guards — raw transport details must never leak into the + // user-facing copy — still holds for the new arm. + assert_eq!(recv.error_type.as_deref(), Some("network")); let message = recv.message.unwrap_or_default(); - assert_eq!(message, expected); assert!( - !message.contains("error sending request for url"), - "chat error payload must not expose raw transport details" + !message.contains("error sending request for url") + && !message.contains("internal-api.example.invalid"), + "chat error payload must not expose raw transport details: {message}" ); // Reset the test-only forced error slot while still holding @@ -1925,3 +1929,126 @@ async fn parallel_turn_runs_concurrently_with_primary_on_same_thread() { set_test_run_chat_task_block(None).await; } + +// ── #3714: session-expired arm (must precede `auth_error`) ────────────── +#[test] +fn classify_session_expired_sentinel_routes_to_signin_not_generic() { + for raw in [ + "SESSION_EXPIRED: backend session not active — sign in to resume LLM work", + "SESSION_EXPIRED: backend session token expired locally — re-authentication required", + "no backend session token; run auth_store_session first", + ] { + let c = classify_inference_error(raw); + assert_eq!(c.error_type, "session_expired", "raw={raw:?}"); + assert!(!c.retryable, "session-expiry is not retryable: {raw:?}"); + assert_ne!( + c.message, + generic_inference_error_user_message(), + "must not be the generic catch-all: {raw:?}" + ); + } +} + +#[test] +fn classify_session_expired_claims_managed_backend_401_invalid_token_before_auth_error() { + // The OpenHuman backend 401 "Invalid token" envelope contains "401", which + // the `auth_error` arm would otherwise claim ("check your API key") — wrong + // for managed-backend users. The session arm must win. + let c = classify_inference_error( + "OpenHuman API error (401 Unauthorized): {\"error\":\"Invalid token\"}", + ); + assert_eq!(c.error_type, "session_expired"); +} + +#[test] +fn classify_byo_provider_401_stays_auth_error_not_session_expired() { + // A BYO provider's own 401 (user's API key) must NOT be swallowed by the + // session arm — it stays actionable as `auth_error`. + let c = classify_inference_error( + "OpenAI API error (401 Unauthorized): {\"error\":{\"message\":\"invalid_api_key\"}}", + ); + assert_eq!(c.error_type, "auth_error"); +} + +// ── #3714: transport-drop arm (bucket #1, was the generic catch-all) ───── +#[test] +fn classify_connection_drop_routes_to_network_retryable() { + for raw in [ + "error sending request for url (https://api.tinyhumans.ai/openai/v1/chat/completions): \ + connection closed before message completed", + "request or response body error: unexpected end of file", + // Raw mid-stream SSE drop: managed backend leaves OFF the errorCode, so + // it reaches the ladder as a streaming error with a transport body. + "OpenHuman streaming API error: error reading a body from connection: \ + end of file before message length reached", + ] { + let c = classify_inference_error(raw); + assert_eq!(c.error_type, "network", "raw={raw:?}"); + assert!(c.retryable, "transport drop is retryable: {raw:?}"); + assert_ne!( + c.message, + generic_inference_error_user_message(), + "raw={raw:?}" + ); + } +} + +#[test] +fn classify_managed_sse_badrequest_not_misread_as_network() { + // A managed 400 frame carries errorCode → must stay provider_request_rejected + // (claimed by the errorCode short-circuit before the transport arm). + let c = classify_inference_error( + "OpenHuman streaming API error: {\"error\":{\"message\":\"Message has tool role, \ + but there was no previous assistant message with a tool call!\",\ + \"type\":\"stream_error\",\"errorCode\":\"BAD_REQUEST\"}}", + ); + assert_eq!(c.error_type, "provider_request_rejected"); +} + +#[test] +fn classify_timeout_not_shadowed_by_network_arm() { + let c = classify_inference_error("request timed out while reading response"); + assert_eq!(c.error_type, "timeout"); +} + +// ── #3714: poisoned-history 400 gets the "we cleared it, resend" copy ──── +#[test] +fn classify_managed_tool_ordering_400_gets_cleared_resend_copy() { + // Managed backend `validateToolMessageOrdering` rejection (orphaned tool + // message) arrives as a BAD_REQUEST SSE frame — must read "we cleared it, + // send again" (not "try a different model") and be retryable, since the + // de-poison guard already evicted the bad warm session. + let c = classify_inference_error( + "OpenHuman streaming API error: {\"error\":{\"message\":\"Message at index 3 has role \ + 'tool' but is not preceded by an assistant message with a matching tool_call\",\ + \"type\":\"stream_error\",\"errorCode\":\"BAD_REQUEST\"}}", + ); + assert_eq!(c.error_type, "provider_request_rejected"); + assert!(c.retryable, "post-eviction resend works → retryable"); + assert!(c.message.contains("cleared it"), "got: {}", c.message); + assert!(!c.message.contains("different model"), "got: {}", c.message); +} + +#[test] +fn classify_byo_tool_ordering_400_gets_cleared_resend_copy() { + let c = classify_inference_error( + "OpenAI API error (400 Bad Request): {\"error\":{\"message\":\"Invalid parameter: \ + messages with role 'tool' must be a response to a preceding message with 'tool_calls'.\"}}", + ); + assert_eq!(c.error_type, "provider_request_rejected"); + assert!(c.retryable); + assert!(c.message.contains("cleared it"), "got: {}", c.message); +} + +#[test] +fn classify_genuine_param_400_keeps_model_mismatch_copy_not_glitch() { + // A real model/param 400 (no tool-ordering signature) must NOT get the + // "we cleared it" copy — resending the same params fails again. + let c = classify_inference_error( + "custom_openai API error (400 Bad Request): {\"error\":{\"message\":\ + \"Unsupported value: 'temperature' must be 1 for this model\"}}", + ); + assert_eq!(c.error_type, "provider_request_rejected"); + assert!(!c.retryable, "param mismatch is not retryable"); + assert!(!c.message.contains("cleared it"), "got: {}", c.message); +} diff --git a/tests/channels_provider_leftovers_raw_coverage_e2e.rs b/tests/channels_provider_leftovers_raw_coverage_e2e.rs index 192eafafb..9c6a963ab 100644 --- a/tests/channels_provider_leftovers_raw_coverage_e2e.rs +++ b/tests/channels_provider_leftovers_raw_coverage_e2e.rs @@ -328,10 +328,12 @@ async fn web_round19_covers_classifier_variants_and_cancel_cleanup() { assert_eq!(budget.error_type, "budget_exhausted"); assert_eq!(budget.source, "openhuman_billing"); + // #3714: a DNS / transport drop now classifies as the dedicated `network` + // arm (was the generic `inference` catch-all), still retryable. let network = web_test_support::classify_error_for_test( "request error: dns error while trying to connect", ); - assert_eq!(network.error_type, "inference"); + assert_eq!(network.error_type, "network"); assert!(network.retryable); web_test_support::set_forced_run_chat_task_error_for_test(Some(