From 6d2e722fa8f916a10f50f23027bd21e5382fdee6 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Wed, 13 May 2026 10:00:34 +0530 Subject: [PATCH] fix(threads): typed ThreadNotFound error + skip Sentry for stale-thread RPCs (OPENHUMAN-TAURI-4H, -60) (#1570) Co-authored-by: Claude Opus 4.7 (1M context) Co-authored-by: Steven Enamakel --- Cargo.toml | 2 +- app/src/pages/Conversations.tsx | 9 + .../services/__tests__/coreRpcClient.test.ts | 40 +++- app/src/services/coreRpcClient.ts | 42 +++- app/src/store/__tests__/threadSlice.test.ts | 111 +++++++++++ app/src/store/threadSlice.ts | 77 +++++++- src/core/jsonrpc.rs | 38 +++- src/core/jsonrpc_tests.rs | 139 ++++++++++++- src/openhuman/memory/conversations/store.rs | 6 +- src/openhuman/threads/error.rs | 183 ++++++++++++++++++ src/openhuman/threads/mod.rs | 2 + src/openhuman/threads/ops.rs | 13 +- src/openhuman/threads/ops_tests.rs | 78 ++++++++ src/rpc/mod.rs | 2 + src/rpc/structured_error.rs | 108 +++++++++++ tests/json_rpc_e2e.rs | 67 +++++++ 16 files changed, 893 insertions(+), 24 deletions(-) create mode 100644 src/openhuman/threads/error.rs create mode 100644 src/rpc/structured_error.rs diff --git a/Cargo.toml b/Cargo.toml index cfc5e25ed..b3c0fa403 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -110,7 +110,7 @@ tower = { version = "0.5", default-features = false } opentelemetry = { version = "0.31", default-features = false, features = ["trace", "metrics"] } opentelemetry_sdk = { version = "0.31", default-features = false, features = ["trace", "metrics"] } opentelemetry-otlp = { version = "0.31", default-features = false, features = ["trace", "metrics", "http-proto", "reqwest-client", "reqwest-rustls-webpki-roots"] } -sentry = { version = "0.47.0", default-features = false, features = ["backtrace", "contexts", "panic", "tracing", "debug-images", "reqwest", "rustls"] } +sentry = { version = "0.47.0", default-features = false, features = ["backtrace", "contexts", "panic", "tracing", "debug-images", "reqwest", "rustls", "test"] } tokio-stream = { version = "0.1.18", features = ["full"] } url = "2" socketioxide = { version = "0.15", features = ["extensions"] } diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx index 08e5486d5..843f77d5e 100644 --- a/app/src/pages/Conversations.tsx +++ b/app/src/pages/Conversations.tsx @@ -39,6 +39,7 @@ import { persistReaction, setActiveThread, setSelectedThread, + THREAD_NOT_FOUND_MESSAGE, } from '../store/threadSlice'; import type { ConfirmationModal as ConfirmationModalType } from '../types/intelligence'; import type { ThreadMessage } from '../types/thread'; @@ -581,6 +582,14 @@ const Conversations = ({ variant = 'page', composer = 'text' }: ConversationsPro try { await dispatch(addMessageLocal({ threadId: sendingThreadId, message: userMessage })).unwrap(); } catch (error) { + // RTK's unwrap() re-throws the rejectWithValue payload directly (a plain + // string, not an Error). Check for the stale-thread sentinel before + // coercing to a display string so this guard doesn't accidentally match + // unrelated errors whose `.toString()` happens to equal the sentinel. + if (error === THREAD_NOT_FOUND_MESSAGE) { + setSendError(null); + return; + } const msg = error instanceof Error ? error.message : String(error); setSendError(chatSendError('cloud_send_failed', msg)); return; diff --git a/app/src/services/__tests__/coreRpcClient.test.ts b/app/src/services/__tests__/coreRpcClient.test.ts index 513c3b2ae..84667bddd 100644 --- a/app/src/services/__tests__/coreRpcClient.test.ts +++ b/app/src/services/__tests__/coreRpcClient.test.ts @@ -4,7 +4,12 @@ import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; import { dispatchLocalAiMethod } from '../../lib/ai/localCoreAiMemory'; import { CORE_RPC_TIMEOUT_MS } from '../../utils/config'; import type { AccessibilityStatus, CommandResponse } from '../../utils/tauriCommands'; -import { callCoreRpc, classifyRpcError, CoreRpcError } from '../coreRpcClient'; +import { + callCoreRpc, + classifyRpcError, + CoreRpcError, + isThreadNotFoundCoreRpcError, +} from '../coreRpcClient'; function sampleAccessibilityStatus( overrides: Partial = {} @@ -526,6 +531,12 @@ describe('classifyRpcError', () => { test('http status 429 wins over message text', () => { expect(classifyRpcError('anything', 429)).toBe('rate_limited'); }); + + test('structured ThreadNotFound data wins over message text', () => { + expect( + classifyRpcError('thread thread-123 not found', undefined, { kind: 'ThreadNotFound' }) + ).toBe('thread_not_found'); + }); }); describe('coreRpcClient — typed errors + auth-expired event', () => { @@ -649,6 +660,33 @@ describe('coreRpcClient — typed errors + auth-expired event', () => { expect((err as Error).message).toBe('something weird'); expect(authExpiredHandler).not.toHaveBeenCalled(); }); + + test('classifies structured ThreadNotFound data without firing the auth-expired event', async () => { + const fetchMock = vi.mocked(fetch); + fetchMock.mockResolvedValueOnce({ + ok: true, + json: async () => ({ + jsonrpc: '2.0', + id: 1, + error: { + code: -32000, + message: 'thread thread-123 not found', + data: { + kind: 'ThreadNotFound', + thread_id: 'thread-123', + method: 'openhuman.threads_message_append', + }, + }, + }), + } as Response); + + const err = await callCoreRpc({ method: 'openhuman.threads_message_append' }).catch(e => e); + expect(err).toBeInstanceOf(CoreRpcError); + expect((err as CoreRpcError).kind).toBe('thread_not_found'); + expect(isThreadNotFoundCoreRpcError(err, 'thread-123')).toBe(true); + expect(isThreadNotFoundCoreRpcError(err, 'thread-other')).toBe(false); + expect(authExpiredHandler).not.toHaveBeenCalled(); + }); }); describe('getCoreRpcUrl', () => { diff --git a/app/src/services/coreRpcClient.ts b/app/src/services/coreRpcClient.ts index 2e4ae79c3..5870fc9ba 100644 --- a/app/src/services/coreRpcClient.ts +++ b/app/src/services/coreRpcClient.ts @@ -53,16 +53,19 @@ export type CoreRpcErrorKind = | 'transport' | 'rate_limited' | 'budget_exceeded' + | 'thread_not_found' | 'unknown'; export class CoreRpcError extends Error { readonly kind: CoreRpcErrorKind; readonly httpStatus?: number; - constructor(message: string, kind: CoreRpcErrorKind, httpStatus?: number) { + readonly data?: unknown; + constructor(message: string, kind: CoreRpcErrorKind, httpStatus?: number, data?: unknown) { super(message); this.name = 'CoreRpcError'; this.kind = kind; this.httpStatus = httpStatus; + this.data = data; } } @@ -74,7 +77,12 @@ const AUTH_EXPIRED_EVENT = 'core-rpc-auth-expired'; * produced by `src/openhuman/backend_api/*` (`authed_json`, rate limiter, * budget guard) and `reqwest::Error`'s connect/timeout variants. */ -export function classifyRpcError(message: string, httpStatus?: number): CoreRpcErrorKind { +export function classifyRpcError( + message: string, + httpStatus?: number, + data?: unknown +): CoreRpcErrorKind { + if (isThreadNotFoundRpcData(data)) return 'thread_not_found'; if (httpStatus === 401) return 'auth_expired'; if (httpStatus === 429) return 'rate_limited'; if (/\(401\b.*Unauthorized\)|Session expired/i.test(message)) return 'auth_expired'; @@ -92,6 +100,32 @@ export function classifyRpcError(message: string, httpStatus?: number): CoreRpcE return 'unknown'; } +function isThreadNotFoundRpcData(data: unknown): boolean { + if (!data || typeof data !== 'object') return false; + // The server only ever emits kind === 'ThreadNotFound' (see + // src/openhuman/threads/error.rs THREAD_NOT_FOUND_KIND). The snake_case + // variant is not produced anywhere; keep only the canonical form. + return (data as { kind?: unknown }).kind === 'ThreadNotFound'; +} + +function threadIdFromRpcData(data: unknown): string | null { + if (!data || typeof data !== 'object') return null; + const record = data as { thread_id?: unknown; threadId?: unknown }; + if (typeof record.thread_id === 'string') return record.thread_id; + if (typeof record.threadId === 'string') return record.threadId; + return null; +} + +export function isThreadNotFoundCoreRpcError( + error: unknown, + threadId?: string +): error is CoreRpcError { + if (!(error instanceof CoreRpcError) || error.kind !== 'thread_not_found') return false; + if (!threadId) return true; + const errorThreadId = threadIdFromRpcData(error.data); + return !errorThreadId || errorThreadId === threadId; +} + function dispatchAuthExpired(method: string): void { if (typeof window === 'undefined') return; try { @@ -368,9 +402,9 @@ export async function callCoreRpc({ error: json.error, }); const rawMessage = json.error.message || 'Core RPC returned an error'; - const kind = classifyRpcError(rawMessage); + const kind = classifyRpcError(rawMessage, undefined, json.error.data); if (kind === 'auth_expired') dispatchAuthExpired(payload.method); - throw new CoreRpcError(rawMessage, kind); + throw new CoreRpcError(rawMessage, kind, undefined, json.error.data); } if (!Object.prototype.hasOwnProperty.call(json, 'result')) { throw new Error('Core RPC response missing result'); diff --git a/app/src/store/__tests__/threadSlice.test.ts b/app/src/store/__tests__/threadSlice.test.ts index c0c10a310..85b3ffe08 100644 --- a/app/src/store/__tests__/threadSlice.test.ts +++ b/app/src/store/__tests__/threadSlice.test.ts @@ -2,17 +2,21 @@ import { configureStore } from '@reduxjs/toolkit'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { threadApi } from '../../services/api/threadApi'; +import { CoreRpcError } from '../../services/coreRpcClient'; import type { Thread, ThreadMessage } from '../../types/thread'; import threadReducer, { addInferenceResponse, addMessageLocal, clearAllThreads, clearSelectedThread, + clearStaleThread, + generateThreadTitleIfNeeded, loadThreadMessages, loadThreads, setActiveThread, setSelectedThread, setWelcomeThreadId, + THREAD_NOT_FOUND_MESSAGE, } from '../threadSlice'; vi.mock('../../services/api/threadApi', () => ({ @@ -158,6 +162,31 @@ describe('threadSlice synchronous reducers', () => { expect(state.activeThreadId).toBeNull(); expect(state.messages).toEqual([]); }); + + it('clearStaleThread removes stale selection, cache, and active id', async () => { + const store = createStore(); + mockedThreadApi.getThreads.mockResolvedValueOnce({ + threads: [makeThread({ id: 't-1' }), makeThread({ id: 't-2' })], + count: 2, + }); + await store.dispatch(loadThreads()); + mockedThreadApi.getThreadMessages.mockResolvedValueOnce({ + messages: [makeMessage()], + count: 1, + }); + await store.dispatch(loadThreadMessages('t-1')); + store.dispatch(setSelectedThread('t-1')); + store.dispatch(setActiveThread('t-1')); + + store.dispatch(clearStaleThread('t-1')); + + const state = store.getState().thread; + expect(state.threads.map(thread => thread.id)).toEqual(['t-2']); + expect(state.messagesByThreadId['t-1']).toBeUndefined(); + expect(state.selectedThreadId).toBeNull(); + expect(state.activeThreadId).toBeNull(); + expect(state.messages).toEqual([]); + }); }); describe('threadSlice loadThreads thunk', () => { @@ -285,6 +314,37 @@ describe('threadSlice addMessageLocal thunk', () => { expect(mockedThreadApi.generateTitleIfNeeded).not.toHaveBeenCalled(); }); + + it('clears stale thread state and does not retry append on ThreadNotFound', async () => { + const store = createStore(); + mockedThreadApi.getThreads.mockResolvedValueOnce({ + threads: [makeThread({ id: 't-1' })], + count: 1, + }); + await store.dispatch(loadThreads()); + store.dispatch(setSelectedThread('t-1')); + store.dispatch(setActiveThread('t-1')); + + mockedThreadApi.appendMessage.mockRejectedValueOnce( + new CoreRpcError('thread t-1 not found', 'thread_not_found', undefined, { + kind: 'ThreadNotFound', + thread_id: 't-1', + }) + ); + mockedThreadApi.getThreads.mockResolvedValueOnce({ threads: [], count: 0 }); + + const result = await store.dispatch( + addMessageLocal({ threadId: 't-1', message: makeMessage() }) + ); + + expect(result.type).toBe('thread/addMessageLocal/rejected'); + expect(result.payload).toBe(THREAD_NOT_FOUND_MESSAGE); + expect(mockedThreadApi.appendMessage).toHaveBeenCalledTimes(1); + expect(mockedThreadApi.generateTitleIfNeeded).not.toHaveBeenCalled(); + expect(mockedThreadApi.getThreads).toHaveBeenCalledTimes(2); + expect(store.getState().thread.selectedThreadId).toBeNull(); + expect(store.getState().thread.activeThreadId).toBeNull(); + }); }); describe('threadSlice addInferenceResponse thunk', () => { @@ -326,4 +386,55 @@ describe('threadSlice addInferenceResponse thunk', () => { expect(result.type).toBe('thread/addInferenceResponse/rejected'); expect(mockedThreadApi.appendMessage).not.toHaveBeenCalled(); }); + + it('clears stale active thread when assistant append returns ThreadNotFound', async () => { + const store = createStore(); + store.dispatch(setActiveThread('t-active')); + mockedThreadApi.appendMessage.mockRejectedValueOnce( + new CoreRpcError('thread t-active not found', 'thread_not_found', undefined, { + kind: 'ThreadNotFound', + thread_id: 't-active', + }) + ); + mockedThreadApi.getThreads.mockResolvedValueOnce({ threads: [], count: 0 }); + + const result = await store.dispatch(addInferenceResponse({ content: 'ack' })); + + expect(result.type).toBe('thread/addInferenceResponse/rejected'); + expect(result.payload).toBe(THREAD_NOT_FOUND_MESSAGE); + expect(mockedThreadApi.appendMessage).toHaveBeenCalledTimes(1); + expect(store.getState().thread.activeThreadId).toBeNull(); + }); +}); + +describe('threadSlice generateThreadTitleIfNeeded thunk', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('clears stale thread state and refreshes list on ThreadNotFound', async () => { + const store = createStore(); + mockedThreadApi.getThreads.mockResolvedValueOnce({ + threads: [makeThread({ id: 't-1' })], + count: 1, + }); + await store.dispatch(loadThreads()); + store.dispatch(setSelectedThread('t-1')); + + mockedThreadApi.generateTitleIfNeeded.mockRejectedValueOnce( + new CoreRpcError('thread t-1 not found', 'thread_not_found', undefined, { + kind: 'ThreadNotFound', + thread_id: 't-1', + }) + ); + mockedThreadApi.getThreads.mockResolvedValueOnce({ threads: [], count: 0 }); + + const result = await store.dispatch(generateThreadTitleIfNeeded({ threadId: 't-1' })); + + expect(result.type).toBe('thread/generateThreadTitleIfNeeded/rejected'); + expect(result.payload).toBe(THREAD_NOT_FOUND_MESSAGE); + expect(mockedThreadApi.generateTitleIfNeeded).toHaveBeenCalledTimes(1); + expect(mockedThreadApi.getThreads).toHaveBeenCalledTimes(2); + expect(store.getState().thread.selectedThreadId).toBeNull(); + }); }); diff --git a/app/src/store/threadSlice.ts b/app/src/store/threadSlice.ts index c14126199..07dafacec 100644 --- a/app/src/store/threadSlice.ts +++ b/app/src/store/threadSlice.ts @@ -1,10 +1,13 @@ -import { createAsyncThunk, createSlice } from '@reduxjs/toolkit'; +import { createAsyncThunk, createSlice, type PayloadAction } from '@reduxjs/toolkit'; import { threadApi } from '../services/api/threadApi'; +import { isThreadNotFoundCoreRpcError } from '../services/coreRpcClient'; import type { Thread, ThreadMessage } from '../types/thread'; import { IS_DEV } from '../utils/config'; import { resetUserScopedState } from './resetActions'; +export const THREAD_NOT_FOUND_MESSAGE = 'This thread is no longer available.'; + interface ThreadState { threads: Thread[]; selectedThreadId: string | null; @@ -117,6 +120,38 @@ export const loadThreadMessages = createAsyncThunk( } ); +/** + * Shared stale-thread handler used by write thunks. + * + * When `isThreadNotFoundCoreRpcError` is true the thunk should: + * 1. Evict the stale thread from Redux state immediately. + * 2. Reload the thread list so the sidebar reflects backend state. + * 3. Reject with `THREAD_NOT_FOUND_MESSAGE` so callers can branch on it + * without inspecting error message strings. + * + * The `loadThreads` failure is swallowed — a network hiccup on the refresh + * should not surface an additional error on top of the stale-thread UX. + */ +/** + * Side-effect half of stale-thread cleanup: evict the thread from Redux state + * and reload the thread list. Separated from the `rejectWithValue` call so the + * thunk return type is inferred purely from `rejectWithValue(THREAD_NOT_FOUND_MESSAGE)`. + */ +// eslint-disable-next-line @typescript-eslint/no-explicit-any +async function evictStaleThread(threadId: string, dispatch: (action: any) => any): Promise { + dispatch(clearStaleThread(threadId)); + try { + await dispatch(loadThreads()).unwrap(); + } catch (refreshError) { + if (IS_DEV) { + console.debug('[threadSlice] stale-thread list refresh failed', { + threadId, + error: refreshError, + }); + } + } +} + export const addMessageLocal = createAsyncThunk( 'thread/addMessageLocal', async (payload: { threadId: string; message: ThreadMessage }, { dispatch, rejectWithValue }) => { @@ -136,6 +171,10 @@ export const addMessageLocal = createAsyncThunk( } return { threadId: payload.threadId, message: persisted }; } catch (error) { + if (isThreadNotFoundCoreRpcError(error, payload.threadId)) { + await evictStaleThread(payload.threadId, dispatch); + return rejectWithValue(THREAD_NOT_FOUND_MESSAGE); + } return rejectWithValue(error instanceof Error ? error.message : 'Failed to save message'); } } @@ -151,7 +190,7 @@ export const addInferenceResponse = createAsyncThunk( type?: string; extraMetadata?: Record; }, - { getState, rejectWithValue } + { dispatch, getState, rejectWithValue } ) => { const state = getState() as { thread: ThreadState }; const targetThreadId = payload.threadId ?? state.thread.activeThreadId; @@ -172,6 +211,10 @@ export const addInferenceResponse = createAsyncThunk( const persisted = await threadApi.appendMessage(targetThreadId, message); return { threadId: targetThreadId, message: persisted }; } catch (error) { + if (isThreadNotFoundCoreRpcError(error, targetThreadId)) { + await evictStaleThread(targetThreadId, dispatch); + return rejectWithValue(THREAD_NOT_FOUND_MESSAGE); + } return rejectWithValue(error instanceof Error ? error.message : 'Failed to save response'); } } @@ -187,6 +230,10 @@ export const generateThreadTitleIfNeeded = createAsyncThunk( try { thread = await threadApi.generateTitleIfNeeded(payload.threadId, payload.assistantMessage); } catch (error) { + if (isThreadNotFoundCoreRpcError(error, payload.threadId)) { + await evictStaleThread(payload.threadId, dispatch); + return rejectWithValue(THREAD_NOT_FOUND_MESSAGE); + } return rejectWithValue( error instanceof Error ? error.message : 'Failed to generate thread title' ); @@ -283,6 +330,22 @@ const threadSlice = createSlice({ setActiveThread: (state, action: { payload: string | null }) => { state.activeThreadId = action.payload; }, + clearStaleThread: (state, action: PayloadAction) => { + const threadId = action.payload; + state.threads = state.threads.filter(thread => thread.id !== threadId); + delete state.messagesByThreadId[threadId]; + if (state.selectedThreadId === threadId) { + state.selectedThreadId = null; + state.messages = []; + state.messagesError = null; + } + if (state.activeThreadId === threadId) { + state.activeThreadId = null; + } + if (state.welcomeThreadId === threadId) { + state.welcomeThreadId = null; + } + }, clearAllThreads: state => { state.threads = []; state.messagesByThreadId = {}; @@ -319,6 +382,15 @@ const threadSlice = createSlice({ .addCase(loadThreads.fulfilled, (state, action) => { state.isLoadingThreads = false; state.threads = action.payload.threads; + const liveThreadIds = new Set(action.payload.threads.map(thread => thread.id)); + if (state.selectedThreadId && !liveThreadIds.has(state.selectedThreadId)) { + state.selectedThreadId = null; + state.messages = []; + state.messagesError = null; + } + if (state.activeThreadId && !liveThreadIds.has(state.activeThreadId)) { + state.activeThreadId = null; + } }) .addCase(loadThreads.rejected, state => { state.isLoadingThreads = false; @@ -374,6 +446,7 @@ export const { setSelectedThread, clearSelectedThread, setActiveThread, + clearStaleThread, clearAllThreads, resetThreadCachesPreservingSelection, setWelcomeThreadId, diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 4813ab7d0..40ff4cf8b 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -22,6 +22,7 @@ use tokio_util::sync::CancellationToken; use crate::core::all; use crate::core::types::{AppState, RpcError, RpcFailure, RpcRequest, RpcSuccess}; +use crate::rpc::StructuredRpcError; /// Axum handler for JSON-RPC POST requests. /// @@ -55,19 +56,42 @@ pub async fn rpc_handler(State(state): State, Json(req): Json { + Err(raw_message) => { + // Decode the controller-emitted structured envelope (if any) + // here at the transport boundary. Domains opt in by emitting a + // `StructuredRpcError` from their handlers — this layer never + // branches on the RPC method name to recover error semantics. + let structured = StructuredRpcError::decode(&raw_message); + let (display_message, error_data, expected_user_state) = match structured { + Some(envelope) => ( + envelope.message, + envelope.data, + envelope.expected_user_state, + ), + None => (raw_message, None, false), + }; + // Session-expired bubbles up as an "error" but is an expected // boundary condition (auth handler clears the local token and the - // UI re-auths). Don't spam Sentry with it. - if !is_session_expired_error(&message) { + // UI re-auths). Don't spam Sentry with it. Domains that surface + // their own expected-user-state errors (stale thread refs, etc.) + // set the `expected_user_state` flag on their structured envelope + // and skip Sentry here uniformly. + if expected_user_state { + tracing::info!( + method = %method, + "[rpc] expected-user-state error — skipping Sentry: {}", + display_message + ); + } else if !is_session_expired_error(&display_message) { crate::core::observability::report_error_or_expected( - message.as_str(), + display_message.as_str(), "rpc", "invoke_method", &[("method", method.as_str()), ("elapsed_ms", &ms.to_string())], ); } else { - tracing::info!("[rpc] {} -> err ({}ms): {}", method, ms, message); + tracing::info!("[rpc] {} -> err ({}ms): {}", method, ms, display_message); } ( StatusCode::OK, @@ -76,8 +100,8 @@ pub async fn rpc_handler(State(state): State, Json(req): Json sentry::integrations::tracing::EventFilter::Event, + Level::WARN | Level::INFO => sentry::integrations::tracing::EventFilter::Breadcrumb, + _ => sentry::integrations::tracing::EventFilter::Ignore, + }), + ); + let _subscriber_guard = tracing::subscriber::set_default(subscriber); + + let stale_thread_request = crate::core::types::RpcRequest { + jsonrpc: "2.0".to_string(), + id: json!(1), + method: "openhuman.threads_message_append".to_string(), + params: json!({ + "thread_id": "thread-missing", + "message": { + "id": "msg-1", + "content": "hello", + "type": "text", + "extraMetadata": {}, + "sender": "user", + "createdAt": "2026-01-01T00:00:00Z" + } + }), + }; + let response = rpc_handler(State(default_state()), Json(stale_thread_request)).await; + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("response body"); + let body: serde_json::Value = serde_json::from_slice(&body).expect("json response"); + assert_eq!(body["error"]["data"]["kind"], "ThreadNotFound"); + assert!( + transport.fetch_and_clear_events().is_empty(), + "ThreadNotFound should not reach Sentry" + ); + + let unrelated_error_request = crate::core::types::RpcRequest { + jsonrpc: "2.0".to_string(), + id: json!(2), + method: "core.not_a_real_method".to_string(), + params: json!({}), + }; + let response = rpc_handler(State(default_state()), Json(unrelated_error_request)).await; + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("response body"); + let body: serde_json::Value = serde_json::from_slice(&body).expect("json response"); + assert_eq!(body["error"]["data"], serde_json::Value::Null); + + let events = transport.fetch_and_clear_events(); + assert_eq!( + events.len(), + 1, + "unrelated RPC errors should still reach Sentry" + ); + assert_eq!( + events[0].tags.get("domain").map(String::as_str), + Some("rpc") + ); + assert_eq!( + events[0].tags.get("operation").map(String::as_str), + Some("invoke_method") + ); + assert_eq!( + events[0].tags.get("method").map(String::as_str), + Some("core.not_a_real_method") + ); +} + #[test] fn is_session_expired_error_matches_session_jwt_required() { // Regression: Sentry issue 7472592145. diff --git a/src/openhuman/memory/conversations/store.rs b/src/openhuman/memory/conversations/store.rs index b6efeb2a3..52fb61a02 100644 --- a/src/openhuman/memory/conversations/store.rs +++ b/src/openhuman/memory/conversations/store.rs @@ -140,7 +140,7 @@ impl ConversationStore { ) -> Result { let _guard = CONVERSATION_STORE_LOCK.lock(); if !self.thread_exists_unlocked(thread_id)? { - return Err(format!("thread {} does not exist", thread_id)); + return Err(format!("thread {} not found", thread_id)); } let path = self.thread_messages_path(thread_id); if let Some(parent) = path.parent() { @@ -179,7 +179,7 @@ impl ConversationStore { let index = self.thread_index_unlocked()?; let entry = index .get(thread_id) - .ok_or_else(|| format!("thread {} does not exist", thread_id))?; + .ok_or_else(|| format!("thread {} not found", thread_id))?; let threads_path = self.ensure_root()?.join(THREADS_FILENAME); append_jsonl( &threads_path, @@ -213,7 +213,7 @@ impl ConversationStore { let index = self.thread_index_unlocked()?; let entry = index .get(thread_id) - .ok_or_else(|| format!("thread {} does not exist", thread_id))?; + .ok_or_else(|| format!("thread {} not found", thread_id))?; let threads_path = self.ensure_root()?.join(THREADS_FILENAME); append_jsonl( &threads_path, diff --git a/src/openhuman/threads/error.rs b/src/openhuman/threads/error.rs new file mode 100644 index 000000000..4df350ff9 --- /dev/null +++ b/src/openhuman/threads/error.rs @@ -0,0 +1,183 @@ +//! Error taxonomy for the conversation threads RPC surface. + +use serde_json::json; + +use crate::rpc::StructuredRpcError; + +/// Stable JSON-RPC discriminator used by the frontend to handle stale thread +/// references without string matching or user-facing error toasts. +pub const THREAD_NOT_FOUND_KIND: &str = "ThreadNotFound"; + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum ThreadsError { + #[error("thread {thread_id} not found")] + NotFound { thread_id: String }, + #[error("{0}")] + Message(String), +} + +impl ThreadsError { + pub fn not_found(thread_id: impl Into) -> Self { + Self::NotFound { + thread_id: thread_id.into(), + } + } + + pub fn from_thread_scoped_store_error(thread_id: &str, err: String) -> Self { + // Only promote to `NotFound` when the parsed id matches the requested + // `thread_id`. If the store reports a *different* missing id we return + // `Message` so the caller sees the real store error rather than + // clearing the wrong stale thread on the frontend. + match parse_thread_not_found_message(&err) { + Some(parsed_id) if parsed_id == thread_id => Self::not_found(thread_id), + _ => Self::Message(err), + } + } + + /// Builds the structured RPC envelope that the JSON-RPC boundary will + /// surface to the frontend. The transport layer decodes this without + /// any domain-specific branching — it just sees a generic typed error. + fn to_structured_rpc_error(&self) -> Option { + match self { + Self::NotFound { thread_id } => Some(StructuredRpcError { + message: self.to_string(), + data: Some(json!({ + "kind": THREAD_NOT_FOUND_KIND, + "thread_id": thread_id, + })), + // `ThreadNotFound` is a routine user-state condition (the UI + // is holding a stale reference after a delete / purge). The + // boundary must NOT forward this to Sentry — it's noise, not + // an internal failure. + expected_user_state: true, + }), + Self::Message(_) => None, + } + } +} + +impl From for ThreadsError { + fn from(value: String) -> Self { + Self::Message(value) + } +} + +impl From<&str> for ThreadsError { + fn from(value: &str) -> Self { + Self::Message(value.to_string()) + } +} + +impl From for String { + /// Conversion at the controller-handler boundary: structured variants + /// (currently `NotFound`) emit a [`StructuredRpcError`] envelope encoded + /// as a sentinel-prefixed string; plain variants degrade to `Display`. + /// + /// This is the ONLY place where `ThreadNotFound` becomes a wire-shaped + /// RPC error. The JSON-RPC transport layer never inspects the method + /// name to recover that shape. + fn from(value: ThreadsError) -> Self { + if let Some(structured) = value.to_structured_rpc_error() { + structured.encode() + } else { + value.to_string() + } + } +} + +/// Parse the canonical display form (`thread not found`) and the legacy +/// store form (`thread does not exist`) so this module can classify +/// store-layer errors into a typed `NotFound` variant. Private — the JSON-RPC +/// transport layer no longer string-sniffs error messages. +fn parse_thread_not_found_message(message: &str) -> Option<&str> { + let thread_id = message + .strip_prefix("thread ") + .and_then(|rest| rest.strip_suffix(" not found")) + .or_else(|| { + message + .strip_prefix("thread ") + .and_then(|rest| rest.strip_suffix(" does not exist")) + })?; + if thread_id.trim().is_empty() { + None + } else { + Some(thread_id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::rpc::StructuredRpcError; + + #[test] + fn not_found_display_is_stable() { + let err = ThreadsError::not_found("thread-123"); + assert_eq!(err.to_string(), "thread thread-123 not found"); + } + + #[test] + fn parses_canonical_and_legacy_thread_not_found_messages() { + assert_eq!( + parse_thread_not_found_message("thread thread-123 not found"), + Some("thread-123") + ); + assert_eq!( + parse_thread_not_found_message("thread thread-123 does not exist"), + Some("thread-123") + ); + assert_eq!( + parse_thread_not_found_message("message msg-1 not found in thread thread-123"), + None + ); + } + + #[test] + fn not_found_serializes_to_structured_rpc_error_at_boundary() { + let raw: String = ThreadsError::not_found("thread-123").into(); + let structured = + StructuredRpcError::decode(&raw).expect("NotFound must emit a structured envelope"); + assert_eq!(structured.message, "thread thread-123 not found"); + assert!(structured.expected_user_state); + let data = structured.data.expect("structured error must carry data"); + assert_eq!(data["kind"], THREAD_NOT_FOUND_KIND); + assert_eq!(data["thread_id"], "thread-123"); + } + + #[test] + fn message_variant_stays_plain_at_boundary() { + let raw: String = ThreadsError::Message("kaboom".into()).into(); + assert_eq!(raw, "kaboom"); + assert!( + StructuredRpcError::decode(&raw).is_none(), + "plain messages must not carry the structured sentinel" + ); + } + + #[test] + fn from_thread_scoped_store_error_id_guard() { + // Matching id → NotFound + let matching = ThreadsError::from_thread_scoped_store_error( + "thread-123", + "thread thread-123 not found".to_string(), + ); + assert_eq!(matching, ThreadsError::not_found("thread-123")); + + // Mismatched id → Message (avoid clearing the wrong thread on the frontend) + let mismatch = ThreadsError::from_thread_scoped_store_error( + "thread-456", + "thread thread-123 not found".to_string(), + ); + assert!( + matches!(mismatch, ThreadsError::Message(_)), + "mismatched id must produce Message, not NotFound" + ); + + // Unrecognised format → Message + let unrecognised = ThreadsError::from_thread_scoped_store_error( + "thread-123", + "some other error".to_string(), + ); + assert!(matches!(unrecognised, ThreadsError::Message(_))); + } +} diff --git a/src/openhuman/threads/mod.rs b/src/openhuman/threads/mod.rs index 0384aca14..fd9cb01bd 100644 --- a/src/openhuman/threads/mod.rs +++ b/src/openhuman/threads/mod.rs @@ -4,11 +4,13 @@ //! CRUD. Storage delegates to `memory::conversations` JSONL files; this //! module owns the RPC surface and controller registry. +pub mod error; pub mod ops; pub mod schemas; pub mod title; pub mod turn_state; +pub use error::{ThreadsError, THREAD_NOT_FOUND_KIND}; pub use schemas::{ all_controller_schemas as all_threads_controller_schemas, all_registered_controllers as all_threads_registered_controllers, diff --git a/src/openhuman/threads/ops.rs b/src/openhuman/threads/ops.rs index e06311d98..72ba5dd4a 100644 --- a/src/openhuman/threads/ops.rs +++ b/src/openhuman/threads/ops.rs @@ -25,6 +25,7 @@ use crate::openhuman::threads::turn_state::{ self, ClearTurnStateRequest, ClearTurnStateResponse, GetTurnStateRequest, GetTurnStateResponse, ListTurnStatesResponse, }; +use crate::openhuman::threads::ThreadsError; use crate::rpc::RpcOutcome; use serde::Serialize; use std::collections::BTreeMap; @@ -230,10 +231,11 @@ pub async fn messages_list( /// Appends a message to a conversation thread. pub async fn message_append( request: AppendConversationMessageRequest, -) -> Result>, String> { +) -> Result>, ThreadsError> { let dir = workspace_dir().await?; let message = - conversations::append_message(dir, &request.thread_id, record_to_message(request.message))?; + conversations::append_message(dir, &request.thread_id, record_to_message(request.message)) + .map_err(|err| ThreadsError::from_thread_scoped_store_error(&request.thread_id, err))?; Ok(envelope( message_to_record(message), Some(counts([("num_messages", 1)])), @@ -244,7 +246,7 @@ pub async fn message_append( /// Generates a durable thread title from the first user message and assistant reply. pub async fn thread_generate_title( request: GenerateConversationThreadTitleRequest, -) -> Result>, String> { +) -> Result>, ThreadsError> { let config = Config::load_or_init() .await .map_err(|e| format!("load config: {e}"))?; @@ -253,7 +255,7 @@ pub async fn thread_generate_title( .into_iter() .find(|thread| thread.id == request.thread_id) else { - return Err(format!("thread {} not found", request.thread_id)); + return Err(ThreadsError::not_found(request.thread_id)); }; if !is_auto_generated_thread_title(&thread.title) { @@ -403,7 +405,8 @@ pub async fn thread_generate_title( &request.thread_id, &title, &chrono::Utc::now().to_rfc3339(), - )?; + ) + .map_err(|err| ThreadsError::from_thread_scoped_store_error(&request.thread_id, err))?; tracing::debug!( thread_id = %request.thread_id, diff --git a/src/openhuman/threads/ops_tests.rs b/src/openhuman/threads/ops_tests.rs index 6f33803e9..ae368d00d 100644 --- a/src/openhuman/threads/ops_tests.rs +++ b/src/openhuman/threads/ops_tests.rs @@ -4,7 +4,32 @@ //! the async `ops::*` entry points rely on. use super::*; use crate::openhuman::threads::title::collapse_whitespace; +use crate::openhuman::threads::ThreadsError; use serde_json::{json, Value}; +use std::ffi::OsString; +use std::path::Path; + +struct EnvVarGuard { + key: &'static str, + old: Option, +} + +impl EnvVarGuard { + fn set_to_path(key: &'static str, value: &Path) -> Self { + let old = std::env::var_os(key); + std::env::set_var(key, value.as_os_str()); + Self { key, old } + } +} + +impl Drop for EnvVarGuard { + fn drop(&mut self) { + match &self.old { + Some(value) => std::env::set_var(self.key, value), + None => std::env::remove_var(self.key), + } + } +} // ── request_id ──────────────────────────────────────────────── @@ -394,3 +419,56 @@ fn title_log_prefix_is_grep_friendly_and_stable() { // of the observable contract — lock it down. assert_eq!(THREAD_TITLE_LOG_PREFIX, "[threads:title]"); } + +#[tokio::test] +async fn message_append_returns_typed_not_found_for_stale_thread() { + let _env_lock = crate::openhuman::config::TEST_ENV_LOCK.lock().unwrap(); + let workspace = tempfile::tempdir().expect("workspace"); + let _workspace_guard = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", workspace.path()); + let thread_id = "thread-missing"; + + let err = message_append(AppendConversationMessageRequest { + thread_id: thread_id.to_string(), + message: ConversationMessageRecord { + id: "msg-1".to_string(), + content: "hello".to_string(), + message_type: "text".to_string(), + extra_metadata: Value::Null, + sender: "user".to_string(), + created_at: "2026-01-01T00:00:00Z".to_string(), + }, + }) + .await + .expect_err("missing thread must return a typed not-found error"); + + assert_eq!( + err, + ThreadsError::NotFound { + thread_id: thread_id.to_string() + } + ); + assert_eq!(err.to_string(), "thread thread-missing not found"); +} + +#[tokio::test] +async fn generate_title_returns_typed_not_found_for_stale_thread() { + let _env_lock = crate::openhuman::config::TEST_ENV_LOCK.lock().unwrap(); + let workspace = tempfile::tempdir().expect("workspace"); + let _workspace_guard = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", workspace.path()); + let thread_id = "thread-missing"; + + let err = thread_generate_title(GenerateConversationThreadTitleRequest { + thread_id: thread_id.to_string(), + assistant_message: None, + }) + .await + .expect_err("missing thread must return a typed not-found error"); + + assert_eq!( + err, + ThreadsError::NotFound { + thread_id: thread_id.to_string() + } + ); + assert_eq!(err.to_string(), "thread thread-missing not found"); +} diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index 4a80891a5..0dd9ee8c3 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -11,8 +11,10 @@ use serde::Serialize; use serde_json::json; mod dispatch; +mod structured_error; pub use dispatch::try_dispatch; +pub use structured_error::{StructuredRpcError, STRUCTURED_RPC_ERROR_SENTINEL}; /// Successful RPC handler result: serialized JSON value plus optional log lines. /// diff --git a/src/rpc/structured_error.rs b/src/rpc/structured_error.rs new file mode 100644 index 000000000..197486b0d --- /dev/null +++ b/src/rpc/structured_error.rs @@ -0,0 +1,108 @@ +//! Generic structured error envelope for the JSON-RPC controller boundary. +//! +//! Domains expose typed errors (e.g. `ThreadsError::NotFound`) and convert +//! them — at their own boundary — into [`StructuredRpcError`]s. The envelope +//! is then encoded as a sentinel-prefixed string so it can travel through +//! the existing `Result` channel that controller handlers +//! already use, without changing every handler signature. +//! +//! The JSON-RPC transport layer (`src/core/jsonrpc.rs`) decodes the envelope +//! transparently — it has zero knowledge of which domain produced the error, +//! and never branches on the RPC method name. New domains that want +//! structured RPC errors just emit a [`StructuredRpcError`] at their +//! controller boundary; the transport handles the rest. +//! +//! Wire shape is unchanged: the human-readable `message` populates +//! `RpcError.message` and the typed `data` populates `RpcError.data`. The +//! `expected_user_state` flag is consumed by the boundary itself to decide +//! whether to forward the error to Sentry; it does not appear on the wire. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// Sentinel prefix that marks a [`StructuredRpcError`] encoded as a `String`. +/// +/// The prefix is intentionally noisy and unlikely to collide with a real +/// human-readable error message. If a controller error string starts with +/// this prefix, the boundary decodes the rest as JSON. +pub const STRUCTURED_RPC_ERROR_SENTINEL: &str = "__OPENHUMAN_STRUCTURED_RPC_ERROR_V1__:"; + +/// Generic structured error emitted by a controller / domain boundary. +/// +/// The transport layer decodes this without inspecting the RPC method name +/// or the message contents, so new domains can adopt it without touching +/// `src/core/jsonrpc.rs`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct StructuredRpcError { + /// Human-readable error text for the JSON-RPC `error.message` field. + pub message: String, + /// Typed payload for the JSON-RPC `error.data` field. Must include a + /// stable `kind` discriminator so frontends can branch on it. + pub data: Option, + /// When `true`, the boundary skips Sentry reporting — this is an + /// expected user-visible state (stale thread, missing resource, etc.) + /// not an internal failure. + #[serde(default)] + pub expected_user_state: bool, +} + +impl StructuredRpcError { + /// Encode into a sentinel-prefixed string suitable for the controller + /// `Result<_, String>` error channel. + pub fn encode(&self) -> String { + // serde_json::to_string on a struct of String/Option/bool + // cannot fail in practice, so unwrap is acceptable. + let json = serde_json::to_string(self) + .expect("StructuredRpcError serialization cannot fail: struct contains only String, Option, and bool"); + format!("{STRUCTURED_RPC_ERROR_SENTINEL}{json}") + } + + /// Decode a controller error string if it carries the sentinel prefix. + /// Returns `None` for plain error strings. + pub fn decode(raw: &str) -> Option { + let body = raw.strip_prefix(STRUCTURED_RPC_ERROR_SENTINEL)?; + serde_json::from_str(body).ok() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn encode_then_decode_round_trips() { + let original = StructuredRpcError { + message: "thread thread-123 not found".to_string(), + data: Some(json!({ "kind": "ThreadNotFound", "thread_id": "thread-123" })), + expected_user_state: true, + }; + let encoded = original.encode(); + assert!( + encoded.starts_with(STRUCTURED_RPC_ERROR_SENTINEL), + "encoded string must carry the sentinel prefix" + ); + let decoded = StructuredRpcError::decode(&encoded).expect("decoded"); + assert_eq!(decoded, original); + } + + #[test] + fn decode_returns_none_for_plain_strings() { + assert!(StructuredRpcError::decode("plain error").is_none()); + assert!(StructuredRpcError::decode("").is_none()); + assert!(StructuredRpcError::decode("__OPENHUMAN_STRUCTURED_RPC_ERROR_V1__").is_none()); + } + + #[test] + fn decode_returns_none_for_corrupt_envelope() { + let bad = format!("{STRUCTURED_RPC_ERROR_SENTINEL}not-json"); + assert!(StructuredRpcError::decode(&bad).is_none()); + } + + #[test] + fn expected_user_state_defaults_to_false_when_absent() { + let raw = format!("{STRUCTURED_RPC_ERROR_SENTINEL}{{\"message\":\"x\",\"data\":null}}"); + let decoded = StructuredRpcError::decode(&raw).expect("decoded"); + assert!(!decoded.expected_user_state); + } +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index a2fb121d4..255e669d1 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -1019,6 +1019,73 @@ async fn json_rpc_thread_labels_create_and_update() { rpc_join.abort(); } +#[tokio::test] +async fn json_rpc_thread_not_found_errors_are_structured() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_url_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + let _api_url_guard = EnvVarGuard::unset("OPENHUMAN_API_URL"); + + let (api_addr, api_join) = serve_on_ephemeral(mock_upstream_router()).await; + let api_origin = format!("http://{api_addr}"); + write_min_config(openhuman_home.as_path(), &api_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + let thread_id = "thread-missing"; + + let append = post_json_rpc( + &rpc_base, + 9011, + "openhuman.threads_message_append", + json!({ + "thread_id": thread_id, + "message": { + "id": "msg-1", + "content": "hello", + "type": "text", + "extraMetadata": {}, + "sender": "user", + "createdAt": "2026-01-01T00:00:00Z" + } + }), + ) + .await; + let append_err = assert_jsonrpc_error(&append, "threads_message_append missing thread"); + assert_eq!(append_err["message"], "thread thread-missing not found"); + assert_eq!(append_err["data"]["kind"], "ThreadNotFound"); + assert_eq!(append_err["data"]["thread_id"], thread_id); + // The transport layer no longer stamps the RPC method into the structured + // error data — the domain controller emits a method-agnostic envelope and + // jsonrpc.rs surfaces it verbatim. The frontend keys on `kind` + + // `thread_id` (see `coreRpcClient.isThreadNotFoundRpcData`), not method. + assert!( + append_err["data"]["method"].is_null(), + "method must not appear in structured error data: the domain envelope is method-agnostic" + ); + + let title = post_json_rpc( + &rpc_base, + 9012, + "openhuman.threads_generate_title", + json!({ "thread_id": thread_id }), + ) + .await; + let title_err = assert_jsonrpc_error(&title, "threads_generate_title missing thread"); + assert_eq!(title_err["message"], "thread thread-missing not found"); + assert_eq!(title_err["data"]["kind"], "ThreadNotFound"); + assert_eq!(title_err["data"]["thread_id"], thread_id); + + api_join.abort(); + rpc_join.abort(); +} + #[tokio::test] async fn json_rpc_thread_turn_state_lifecycle() { let _env_lock = json_rpc_e2e_env_lock();