diff --git a/.claude/memory.md b/.claude/memory.md index aeb5dca058..6e99e06faa 100644 --- a/.claude/memory.md +++ b/.claude/memory.md @@ -4,6 +4,8 @@ Quick reference for anyone starting with Claude on this project. Updated by the ## Fixes & Gotchas +- **Who persists a chat reply (issue #6034)** — the core writes an unsegmented reply to the conversation store under `agent:` **before** it publishes `chat_done` (`web_chat::presentation::deliver_response` → `web_chat::reply_persistence`), and the frontend's `addInferenceResponse` reuses that id so the store's idempotency collapses the two writes onto one row. Before this the frontend was the *only* persister of an interactive reply, so a failed `threads_message_append`, a socket reconnect (which issues a new `client_id`, leaving only the `thread:` room as a route) or a webview reload lost the answer from screen and disk while the agent's session history still held it. Two rules follow: any new `chat_done` append site must pass `deliveredReplyMessageId(event)` or the thread gets two copies of one answer (#5933 again), and a **segmented** delivery is the exception — the client owns one row per segment there, so the core stores nothing and the ids stay generated. +- **`deliver_response` does not segment any more** — `presentation.rs` sets `let segments = [full_response.to_string()]` unconditionally, so the multi-segment branch (and every `chat_segment` event) is dead on the interactive path; the segmentation helpers survive for channel callers and tests. Read that binding before reasoning about `segment_total`, which is never set in practice today. - **macOS close button does not dismiss window (issue #2049)** — `WebviewWindow::hide()` routes through CEF's `WindowMessage::Hide` → `cef::Window::hide()` which does NOT propagate to the visible NSWindow frame. Fix: use `AppHandle::hide()` which calls `[NSApp hide:]` via `set_application_visibility(false)`. This is macOS-only (`#[cfg(target_os = "macos")]`); the `CloseRequested` handlers are in `app/src-tauri/src/lib.rs` (grep `WindowEvent::CloseRequested` — line numbers drift). PR #2118. - **ServiceBlockingGate CORS errors** — The gate calls `openhumanServiceStatus()` and `openhumanAgentServerStatus()` at startup. These used `callCoreRpc()` which falls back to raw `fetch()` when socket isn't connected yet, causing CORS errors. Fix: route through `invoke('core_rpc_relay')` instead (Tauri IPC, no CORS). - **Socket not connected at startup** — `SocketProvider` only connects when a Redux `auth.token` is set. At fresh launch (no token), socket is null, so any `callCoreRpc()` call falls back to `fetch()`. Always use `invoke('core_rpc_relay')` for local sidecar RPC calls. diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index f91a78f11a..41fb2dfa35 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -6404,6 +6404,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'النموذج المحلي غير متاح', 'userErrors.localModelUnavailable.body': 'لا يمكن الوصول إلى Ollama على النقطة الطرفية المُهيأة، أو أن النموذج المطلوب غير مثبّت عليها. شغّل Ollama ونزّل النموذج على تلك النقطة الطرفية، أو حوّل هذا العمل إلى مزوّد سحابي.', + 'userErrors.replyDeliveryFailed.title': 'تعذّر عرض الرد', + 'userErrors.replyDeliveryFailed.body': + 'أنهى الوكيل هذه الجولة، لكن تعذّر حفظ ردّه أو قراءته مجددًا. اطلب منه تكرار الرد.', 'userErrors.memoryStoreCorrupt.title': 'تلف فهرس الذاكرة', 'userErrors.memoryStoreCorrupt.body': 'كانت قاعدة بيانات شجرة الذاكرة تالفة. تم الاحتفاظ بالملف التالف بجوار بيانات الذاكرة وأعيد إنشاء فهرس فارغ. أعد مزامنة مصادر الذاكرة لإعادة تعبئته.', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index fa73f7cf1c..df8d7a4b70 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -6554,6 +6554,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'লোকাল মডেল অনুপলব্ধ', 'userErrors.localModelUnavailable.body': 'কনফিগার করা এন্ডপয়েন্টে Ollama-তে পৌঁছানো যাচ্ছে না, অথবা সেখানে প্রয়োজনীয় মডেলটি ইনস্টল করা নেই। Ollama চালু করে সেই এন্ডপয়েন্টে মডেলটি পুল করুন, অথবা এই কাজটি কোনো ক্লাউড প্রোভাইডারে সরিয়ে নিন।', + 'userErrors.replyDeliveryFailed.title': 'উত্তরটি দেখানো যায়নি', + 'userErrors.replyDeliveryFailed.body': + 'এজেন্ট এই দফাটি শেষ করেছে, কিন্তু তার উত্তর সংরক্ষণ বা পুনরায় পড়া যায়নি। আবার জিজ্ঞাসা করলে সে উত্তরটি আবার দেবে।', 'userErrors.memoryStoreCorrupt.title': 'মেমোরি ইনডেক্স নষ্ট হয়ে গেছে', 'userErrors.memoryStoreCorrupt.body': 'আপনার মেমোরি ট্রির ডেটাবেস নষ্ট হয়ে গিয়েছিল। নষ্ট ফাইলটি মেমোরি ডেটার পাশে সংরক্ষিত আছে এবং একটি খালি ইনডেক্স নতুন করে তৈরি হয়েছে। আবার পূরণ করতে মেমোরি উৎসগুলি পুনরায় সিঙ্ক করুন।', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index 9fddcfc4b7..c3a8b0267e 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -6739,6 +6739,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Lokales Modell nicht verfügbar', 'userErrors.localModelUnavailable.body': 'Ollama ist unter dem konfigurierten Endpunkt nicht erreichbar, oder das benötigte Modell ist dort nicht installiert. Starte Ollama und lade das Modell auf diesem Endpunkt, oder verlagere diese Arbeit auf einen Cloud-Anbieter.', + 'userErrors.replyDeliveryFailed.title': 'Antwort konnte nicht angezeigt werden', + 'userErrors.replyDeliveryFailed.body': + 'Der Agent hat diese Runde beendet, seine Antwort ließ sich aber weder speichern noch erneut lesen. Frag noch einmal, damit er sie wiederholt.', 'userErrors.memoryStoreCorrupt.title': 'Gedächtnisindex war beschädigt', 'userErrors.memoryStoreCorrupt.body': 'Die Datenbank des Gedächtnisbaums war beschädigt. Die beschädigte Datei wurde neben deinen Gedächtnisdaten aufbewahrt und ein leerer Index neu aufgebaut. Synchronisiere deine Gedächtnisquellen erneut, um ihn wieder zu füllen.', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index bd411add04..1641176490 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -7069,6 +7069,9 @@ const en: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Local model unavailable', 'userErrors.localModelUnavailable.body': 'Ollama is not reachable at the configured endpoint, or the required model is not installed there. Start Ollama and pull the model at that endpoint, or switch this workload to a cloud provider.', + 'userErrors.replyDeliveryFailed.title': 'Reply could not be shown', + 'userErrors.replyDeliveryFailed.body': + 'The agent finished this turn, but its reply could not be saved or read back. Ask again to have it repeated.', 'userErrors.memoryStoreCorrupt.title': 'Memory index was corrupted', 'userErrors.memoryStoreCorrupt.body': 'The database behind your memory tree was damaged. The damaged file was preserved next to your memory data, and an empty index was rebuilt. Re-sync your memory sources to fill it again.', diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index bf53103b48..0364a571fd 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -6698,6 +6698,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Modelo local no disponible', 'userErrors.localModelUnavailable.body': 'No se puede acceder a Ollama en el punto de conexión configurado, o el modelo necesario no está instalado allí. Inicia Ollama y descarga el modelo en ese punto de conexión, o cambia este trabajo a un proveedor en la nube.', + 'userErrors.replyDeliveryFailed.title': 'No se pudo mostrar la respuesta', + 'userErrors.replyDeliveryFailed.body': + 'El agente terminó este turno, pero su respuesta no se pudo guardar ni volver a leer. Vuelve a preguntar para que la repita.', 'userErrors.memoryStoreCorrupt.title': 'El índice de memoria se dañó', 'userErrors.memoryStoreCorrupt.body': 'La base de datos del árbol de memoria estaba dañada. El archivo dañado se conservó junto a tus datos de memoria y se reconstruyó un índice vacío. Vuelve a sincronizar tus fuentes de memoria para rellenarlo.', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index b690c1ef8d..92d8f1e168 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -6726,6 +6726,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.body': "Ollama n'est pas joignable sur le point de terminaison configuré, ou le modèle requis n'y est pas installé. Lancez Ollama et téléchargez le modèle sur ce point de terminaison, ou basculez cette charge de travail vers un fournisseur cloud.", 'userErrors.scope.chat': 'Chat', + 'userErrors.replyDeliveryFailed.title': 'Impossible d’afficher la réponse', + 'userErrors.replyDeliveryFailed.body': + 'L’agent a terminé ce tour, mais sa réponse n’a pas pu être enregistrée ni récupérée. Redemande-lui de la répéter.', 'userErrors.memoryStoreCorrupt.title': 'L’index mémoire a été corrompu', 'userErrors.memoryStoreCorrupt.body': 'La base de données de l’arbre mémoire était endommagée. Le fichier endommagé a été conservé à côté de vos données mémoire et un index vide a été reconstruit. Resynchronisez vos sources mémoire pour le remplir à nouveau.', diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index 760e545108..ba7835b026 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -6554,6 +6554,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'लोकल मॉडल उपलब्ध नहीं है', 'userErrors.localModelUnavailable.body': 'कॉन्फ़िगर किए गए एंडपॉइंट पर Ollama तक पहुँच नहीं है, या ज़रूरी मॉडल वहाँ इंस्टॉल नहीं है। Ollama शुरू करके उसी एंडपॉइंट पर मॉडल पुल करें, या इस काम को किसी क्लाउड प्रोवाइडर पर ले जाएँ।', + 'userErrors.replyDeliveryFailed.title': 'उत्तर दिखाया नहीं जा सका', + 'userErrors.replyDeliveryFailed.body': + 'एजेंट ने यह बारी पूरी कर ली, लेकिन उसका उत्तर न सहेजा जा सका और न दोबारा पढ़ा जा सका। दोबारा पूछें ताकि वह उत्तर फिर से दे।', 'userErrors.memoryStoreCorrupt.title': 'मेमोरी इंडेक्स खराब हो गया', 'userErrors.memoryStoreCorrupt.body': 'आपकी मेमोरी ट्री का डेटाबेस खराब हो गया था। खराब फाइल आपके मेमोरी डेटा के पास सुरक्षित रखी गई है, और एक खाली इंडेक्स फिर से बनाया गया है। इसे दोबारा भरने के लिए अपने मेमोरी स्रोतों को फिर से सिंक करें।', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index ec5787c00a..9d4057a5a1 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -6590,6 +6590,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Model lokal tidak tersedia', 'userErrors.localModelUnavailable.body': 'Ollama tidak dapat dijangkau di endpoint yang dikonfigurasi, atau model yang dibutuhkan belum terpasang di sana. Jalankan Ollama dan unduh modelnya di endpoint tersebut, atau alihkan pekerjaan ini ke penyedia cloud.', + 'userErrors.replyDeliveryFailed.title': 'Balasan tidak dapat ditampilkan', + 'userErrors.replyDeliveryFailed.body': + 'Agen menyelesaikan giliran ini, tetapi balasannya tidak dapat disimpan atau dibaca ulang. Tanyakan lagi agar diulangi.', 'userErrors.memoryStoreCorrupt.title': 'Indeks memori rusak', 'userErrors.memoryStoreCorrupt.body': 'Basis data pohon memori mengalami kerusakan. Berkas yang rusak disimpan di samping data memori Anda, dan indeks kosong telah dibangun ulang. Sinkronkan ulang sumber memori untuk mengisinya kembali.', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index 06a36c2a7d..fde717d9f8 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -6680,6 +6680,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.body': "Ollama non è raggiungibile sull'endpoint configurato, oppure il modello necessario non è installato lì. Avvia Ollama e scarica il modello su quell'endpoint, oppure sposta questo lavoro su un provider cloud.", 'userErrors.scope.chat': 'Chat', + 'userErrors.replyDeliveryFailed.title': 'Impossibile mostrare la risposta', + 'userErrors.replyDeliveryFailed.body': + 'L’agente ha completato questo turno, ma la sua risposta non è stata salvata né riletta. Chiedi di nuovo per fartela ripetere.', 'userErrors.memoryStoreCorrupt.title': 'L’indice della memoria era corrotto', 'userErrors.memoryStoreCorrupt.body': 'Il database dell’albero della memoria era danneggiato. Il file danneggiato è stato conservato accanto ai tuoi dati di memoria ed è stato ricostruito un indice vuoto. Risincronizza le tue fonti di memoria per riempirlo di nuovo.', diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index d571709b0b..b471dfc1a9 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -6480,6 +6480,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': '로컬 모델을 사용할 수 없음', 'userErrors.localModelUnavailable.body': '구성된 엔드포인트에서 Ollama에 연결할 수 없거나 필요한 모델이 그곳에 설치되어 있지 않습니다. Ollama를 실행하고 해당 엔드포인트에 모델을 내려받거나, 이 작업을 클라우드 제공업체로 전환하세요.', + 'userErrors.replyDeliveryFailed.title': '답변을 표시하지 못했습니다', + 'userErrors.replyDeliveryFailed.body': + '에이전트가 이 턴을 마쳤지만 답변을 저장하거나 다시 읽어올 수 없었습니다. 다시 물어보면 답변을 되풀이합니다.', 'userErrors.memoryStoreCorrupt.title': '메모리 인덱스가 손상되었습니다', 'userErrors.memoryStoreCorrupt.body': '메모리 트리의 데이터베이스가 손상되었습니다. 손상된 파일은 메모리 데이터 옆에 보존되었고 빈 인덱스가 다시 생성되었습니다. 메모리 소스를 다시 동기화하여 채워 주세요.', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index 9773ad6732..b4573a5a6d 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -6656,6 +6656,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Model lokalny niedostępny', 'userErrors.localModelUnavailable.body': 'Ollama jest nieosiągalna pod skonfigurowanym punktem końcowym albo wymagany model nie jest tam zainstalowany. Uruchom Ollamę i pobierz model w tym punkcie końcowym lub przenieś tę pracę do dostawcy w chmurze.', + 'userErrors.replyDeliveryFailed.title': 'Nie udało się pokazać odpowiedzi', + 'userErrors.replyDeliveryFailed.body': + 'Agent zakończył tę turę, ale jego odpowiedzi nie udało się zapisać ani odczytać ponownie. Zapytaj jeszcze raz, aby ją powtórzył.', 'userErrors.memoryStoreCorrupt.title': 'Indeks pamięci był uszkodzony', 'userErrors.memoryStoreCorrupt.body': 'Baza danych drzewa pamięci była uszkodzona. Uszkodzony plik zachowano obok danych pamięci i odbudowano pusty indeks. Zsynchronizuj ponownie źródła pamięci, aby go wypełnić.', diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index b52244a773..8eb69772af 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -6667,6 +6667,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Modelo local indisponível', 'userErrors.localModelUnavailable.body': 'O Ollama não está acessível no endpoint configurado, ou o modelo necessário não está instalado nele. Inicie o Ollama e baixe o modelo nesse endpoint, ou mude este trabalho para um provedor na nuvem.', + 'userErrors.replyDeliveryFailed.title': 'Não foi possível mostrar a resposta', + 'userErrors.replyDeliveryFailed.body': + 'O agente concluiu este turno, mas a resposta dele não pôde ser salva nem lida novamente. Pergunte de novo para que ele repita.', 'userErrors.memoryStoreCorrupt.title': 'O índice de memória foi corrompido', 'userErrors.memoryStoreCorrupt.body': 'O banco de dados da árvore de memória estava danificado. O arquivo danificado foi preservado ao lado dos seus dados de memória e um índice vazio foi reconstruído. Sincronize novamente suas fontes de memória para preenchê-lo.', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index 2f5b85dc17..d834f52ecc 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -6629,6 +6629,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': 'Локальная модель недоступна', 'userErrors.localModelUnavailable.body': 'Ollama недоступен по настроенному адресу, либо нужная модель там не установлена. Запустите Ollama и загрузите модель по этому адресу или переведите эту работу на облачного провайдера.', + 'userErrors.replyDeliveryFailed.title': 'Не удалось показать ответ', + 'userErrors.replyDeliveryFailed.body': + 'Агент завершил этот ход, но его ответ не удалось сохранить или прочитать заново. Спросите ещё раз, чтобы он повторил.', 'userErrors.memoryStoreCorrupt.title': 'Индекс памяти был повреждён', 'userErrors.memoryStoreCorrupt.body': 'База данных дерева памяти была повреждена. Повреждённый файл сохранён рядом с данными памяти, а пустой индекс создан заново. Заново синхронизируйте источники памяти, чтобы заполнить его.', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index 75a2a9bcb3..d2c0e1ccbc 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -6191,6 +6191,9 @@ const messages: TranslationMap = { 'userErrors.localModelUnavailable.title': '本地模型不可用', 'userErrors.localModelUnavailable.body': '无法在配置的端点连接 Ollama，或所需模型未安装在该端点。请启动 Ollama 并在该端点拉取模型，或将此工作切换到云端提供商。', + 'userErrors.replyDeliveryFailed.title': '无法显示回复', + 'userErrors.replyDeliveryFailed.body': + '智能体已完成这一轮，但它的回复既没能保存也没能重新读取。再问一次即可让它重复回复。', 'userErrors.memoryStoreCorrupt.title': '记忆索引已损坏', 'userErrors.memoryStoreCorrupt.body': '记忆树使用的数据库已损坏。受损文件已保留在记忆数据旁边，并已重建一个空索引。请重新同步记忆来源以重新填充。', diff --git a/app/src/lib/userErrors/classify.ts b/app/src/lib/userErrors/classify.ts index b2dad07116..82f95a04a3 100644 --- a/app/src/lib/userErrors/classify.ts +++ b/app/src/lib/userErrors/classify.ts @@ -127,6 +127,30 @@ export function classifyIntegrationError( }; } +/** + * #6034: a completed reply that reached neither writer, so there is nothing to + * render and nothing to re-read. + * + * The caller already knows this happened — it just failed an append and a + * refetch — so, like the two constructors above, this takes the fact rather + * than sniffing prose. Scoped per thread so two different lost replies are two + * entries, and one thread failing repeatedly is one entry with a count. + */ +export function classifyReplyDeliveryFailure(threadId: string): UserErrorDescriptor | null { + const thread = threadId?.trim(); + if (!thread) return null; + return { + id: userErrorId('reply_delivery_failed', 'chat', thread), + kind: 'reply_delivery_failed', + severity: 'error', + scope: 'chat', + sourceDomain: 'chat', + titleKey: 'userErrors.replyDeliveryFailed.title', + bodyKey: 'userErrors.replyDeliveryFailed.body', + action: 'dismiss', + }; +} + /** Build the stable dedupe identity for an error. */ export function userErrorId( kind: UserErrorDescriptor['kind'], diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 7a21d1ca34..a1279bc9db 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -7,6 +7,7 @@ import { createSkillToolChainLatencyTracker, SKILL_TOOL_CHAIN_TARGET_MS, } from '../lib/ai/skillToolChainLatency'; +import { classifyReplyDeliveryFailure } from '../lib/userErrors/classify'; import { ingestRuntimeErrorSignal } from '../lib/userErrors/report'; import { maybeParseWorkflowProposalTool } from '../lib/workflows/workflowProposal'; import { @@ -28,6 +29,7 @@ import { segmentText, subscribeChatEvents, } from '../services/chatService'; +import { socketService } from '../services/socketService'; import { store } from '../store'; import { appendSubagentStreamDelta, @@ -74,9 +76,11 @@ import { clearThreadInferenceActive, createNewThread, generateThreadTitleIfNeeded, + loadThreadMessages, setActiveThread, setSelectedThread, } from '../store/threadSlice'; +import { reportUserError } from '../store/userErrorsSlice'; import { IS_PROD } from '../utils/config'; import { AssistantUiRuntimeProvider } from './AssistantUiRuntimeProvider'; import { isProactiveConversationSurface, proactiveThreadPins } from './proactiveThreadPins'; @@ -238,6 +242,28 @@ function corePersistedMessageId(event: { return event.client_id === 'system' && event.request_id ? `agent:${event.request_id}` : undefined; } +/** + * Message id for a delivered reply, shared with the core. + * + * Since #6034 the core stores an unsegmented reply before it announces it, on + * every surface that carries a workspace — interactive turns and forked lanes, + * not just the core-initiated ones {@link corePersistedMessageId} covers. Both + * writers therefore derive the id from the same `request_id`, which is what + * makes our append collapse onto the core's row rather than add a second copy + * of the answer (#5933). + * + * A segmented `chat_done` is excluded on purpose: the core leaves those rows to + * us (one per segment), so there is nothing to collapse onto and a shared id + * would make several segments fight over one row. + */ +function deliveredReplyMessageId(event: { + request_id?: string; + segment_total?: number | null; +}): string | undefined { + if (event.segment_total) return undefined; + return event.request_id ? `agent:${event.request_id}` : undefined; +} + /** * Map a `chat_done` event's holistic usage onto the `recordChatTurnUsage` * payload. Prefers the structured `usage` object (tokens + cost + context window @@ -310,6 +336,9 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { // (#4273, AC3). Single instance for the provider's lifetime; observability // only — it never gates or cancels a turn. const skillLatencyRef = useRef(createSkillToolChainLatencyTracker()); + // Threads that had a turn in flight when the socket dropped, held across the + // gap so the reconnect can rejoin their rooms and re-read them (#6034). + const interruptedThreadsRef = useRef>(new Set()); useEffect(() => { toolTimelineRef.current = toolTimelineByThread; @@ -474,6 +503,71 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { } }; + /** + * Tell the user a finished reply could not be shown. + * + * The turn ran and was paid for, and by this point neither writer left a + * row — so a console line is the wrong place for it. An in-thread message + * is not an option either: writing one needs the very append that just + * failed. The shell's notice panel is the surface that does not depend on + * the thread store. + */ + const reportLostReply = (threadId: string) => { + const descriptor = classifyReplyDeliveryFailure(threadId); + if (descriptor) dispatch(reportUserError({ descriptor, at: Date.now() })); + }; + + /** + * Put a delivered reply back on screen after our own append failed. + * + * Since #6034 the core stores an unsegmented reply before it announces it, + * so a failed `threads_message_append` is almost always a lost *render* + * rather than a lost answer: re-reading the thread brings the core's row + * into the cache and the reply appears without the user re-asking. + * + * The residual case — neither writer stored it — is reported at error + * level rather than through the dev-only debug channel. It cannot be + * surfaced as an in-thread message, because writing one needs the same + * append that just failed. + */ + const recoverDeliveredReply = async (event: ChatDoneEvent, cause: unknown) => { + const expectedId = deliveredReplyMessageId(event); + try { + await dispatch(loadThreadMessages(event.thread_id)).unwrap(); + } catch (refetchError) { + console.error( + '[chat-runtime] a delivered reply could not be appended or re-read; it may be missing from this thread', + { + threadId: event.thread_id, + requestId: event.request_id, + appendError: cause instanceof Error ? cause.message : String(cause), + refetchError: + refetchError instanceof Error ? refetchError.message : String(refetchError), + } + ); + reportLostReply(event.thread_id); + return; + } + const recovered = expectedId + ? (store.getState().thread.messagesByThreadId[event.thread_id] ?? []).some( + m => m.id === expectedId + ) + : false; + if (recovered) { + rtLog('chat_done_recovered_from_store', { + thread: event.thread_id, + request: event.request_id, + }); + return; + } + console.error('[chat-runtime] a delivered reply is absent from the thread after a refetch', { + threadId: event.thread_id, + requestId: event.request_id, + appendError: cause instanceof Error ? cause.message : String(cause), + }); + reportLostReply(event.thread_id); + }; + const finishChatDoneTurn = async (event: ChatDoneEvent, path: string) => { rtLog('refresh_usage_counter', { thread: event.thread_id, @@ -1179,6 +1273,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { addInferenceResponse({ content: event.full_response, threadId: event.thread_id, + messageId: deliveredReplyMessageId(event), extraMetadata: chatDoneExtraMetadata(event), }) ).unwrap(); @@ -1194,6 +1289,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { request: event.request_id, error: error instanceof Error ? error.message : String(error), }); + await recoverDeliveredReply(event, error); } })(); } @@ -1226,7 +1322,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { addInferenceResponse({ content: event.full_response, threadId: event.thread_id, - messageId: corePersistedMessageId(event), + messageId: deliveredReplyMessageId(event), extraMetadata: chatDoneExtraMetadata(event), }) ).unwrap(); @@ -1242,6 +1338,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { request: event.request_id, error: error instanceof Error ? error.message : String(error), }); + await recoverDeliveredReply(event, error); } await finishChatDoneTurn(event, 'proactive'); })(); @@ -1477,6 +1574,13 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { const threadIds = Object.keys(lifecycles); const activeThreadIds = Object.keys(state.thread.activeThreadIds); if (threadIds.length === 0 && activeThreadIds.length === 0) return; + // Remember what was in flight BEFORE the markers are cleared below. The + // reconnect handler in `socketService` re-subscribes from + // `activeThreadIds`, which this effect is about to empty — so without this + // snapshot the new socket rejoins only the selected thread's room, and a + // turn finishing on any other thread announces itself to a `client_id` + // that no longer exists (#6034). + interruptedThreadsRef.current = new Set([...threadIds, ...activeThreadIds]); // Abandon any in-flight tool-chain latency windows: a disconnect tears down // these turns without an onDone/onError, so without this the next tool call // on a reused thread would attribute stale elapsed/tool counts (#4288). @@ -1502,6 +1606,39 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { } }, [socketStatus, dispatch]); + // Heal the threads a disconnect orphaned, once the socket is back. + // + // Two things are wrong at this moment and neither fixes itself. The new + // socket has a new `client_id`, so the only route left to an in-flight turn + // is its `thread:` room — and the reconnect handler rejoined just the + // selected thread. And a turn that finished while we were away announced its + // `chat_done` to nobody: the reply is on disk (the core stores it before + // announcing it) but nothing re-reads the thread while the user stays put. + // Re-subscribe and re-read, so a reply that landed during the gap appears + // instead of looking lost until the thread is reselected (#6034). + useEffect(() => { + if (socketStatus !== 'connected') return; + const interrupted = interruptedThreadsRef.current; + if (interrupted.size === 0) return; + rtLog('socket_reconnect_heal', { threads: interrupted.size }); + for (const threadId of [...interrupted]) { + // Read only after the room join is acknowledged. Firing both at once + // leaves a window where the read misses a reply that lands a moment + // later and the turn's `chat_done` goes to a room we have not joined + // yet — the reply would then stay invisible until a manual reload. + // `subscribeThread` resolves `false` on its own timeout, or immediately + // if the socket has already dropped again, so the read is never skipped. + void socketService.subscribeThread(threadId).then(joined => { + // Forget the thread only once it is provably back in its room. A + // socket that dropped again between `connected` and this effect never + // emitted, and dropping the id here would strand that thread until the + // user reselected it; keeping it means the next connection retries. + if (joined) interruptedThreadsRef.current.delete(threadId); + return dispatch(loadThreadMessages(threadId)); + }); + } + }, [socketStatus, dispatch]); + // assistant-ui's runtime is mounted here, INSIDE the subscription provider, // so it sits above every chat surface and below the Redux store this file // already feeds. It is additive: it publishes the runtime context without diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index 4e90adde69..eb7ddff5b6 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import * as chatService from '../../services/chatService'; import { threadApi } from '../../services/api/threadApi'; +import { socketService } from '../../services/socketService'; import { store } from '../../store'; import { clearAllChatRuntime, @@ -15,7 +16,12 @@ import { setPendingPlanReviewForThread, } from '../../store/chatRuntimeSlice'; import { setStatusForUser } from '../../store/socketSlice'; -import { clearAllThreads, loadThreads, setSelectedThread } from '../../store/threadSlice'; +import { + clearAllThreads, + loadThreads, + setActiveThread, + setSelectedThread, +} from '../../store/threadSlice'; import ChatRuntimeProvider from '../ChatRuntimeProvider'; import { clearAllProactiveThreadPins } from '../proactiveThreadPins'; @@ -39,6 +45,10 @@ vi.mock('../../services/api/threadApi', () => ({ }, })); +vi.mock('../../services/socketService', () => ({ + socketService: { subscribeThread: vi.fn(() => Promise.resolve(true)) }, +})); + vi.mock('../../hooks/usageRefresh', () => ({ requestUsageRefresh: vi.fn() })); const mockRefetchSnapshot = vi.fn(); @@ -605,7 +615,7 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria expect(threadApi.appendMessage).toHaveBeenCalledTimes(1); }); - it('keeps a generated id for an interactive chat_done (nothing else persisted it)', async () => { + it('mirrors the core id on an interactive chat_done so the two writers collapse (#6034)', async () => { const listeners = renderProvider(); act(() => { @@ -617,12 +627,103 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria }); }); + // The core stores an unsegmented reply before announcing it, so this + // append must carry the same `agent:` id — otherwise the + // thread ends up with the core's row AND ours, which is #5933 again. await waitFor(() => expect(threadApi.appendMessage).toHaveBeenCalledTimes(1)); const [, persisted] = vi.mocked(threadApi.appendMessage).mock.calls[0]; - expect(persisted.id).not.toBe('agent:r-user'); + expect(persisted.id).toBe('agent:r-user'); expect(persisted.sender).toBe('agent'); }); + it('surfaces a user-visible error when neither writer stored the reply (#6034)', async () => { + vi.mocked(threadApi.appendMessage).mockRejectedValueOnce(new Error('rpc timeout')); + // The refetch succeeds but the row is genuinely not there — the case + // where the core write failed too. A console line is not enough here: + // the user is looking at a finished turn with no answer. + vi.mocked(threadApi.getThreadMessages).mockResolvedValueOnce({ + messages: [], + } as unknown as Awaited>); + + const listeners = renderProvider(); + act(() => { + listeners.onDone?.({ + thread_id: 't-gone', + request_id: 'r-gone', + full_response: 'an answer nobody stored', + rounds_used: 1, + }); + }); + + await waitFor(() => + expect( + Object.values(store.getState().userErrors.byId).some( + e => e.kind === 'reply_delivery_failed' + ) + ).toBe(true) + ); + }); + + it('keeps a generated id for a segmented chat_done (the core left those rows to us)', async () => { + const listeners = renderProvider(); + + act(() => { + listeners.onDone?.({ + thread_id: 't-seg', + request_id: 'r-seg', + full_response: 'one two', + segment_total: 2, + rounds_used: 1, + }); + }); + + // Segments are the client's rows to write, several of them under one + // request id, so a shared deterministic id would make them collapse onto + // each other and lose all but the first. + await waitFor(() => expect(threadApi.appendMessage).toHaveBeenCalledTimes(1)); + const [, persisted] = vi.mocked(threadApi.appendMessage).mock.calls[0]; + expect(persisted.id).not.toBe('agent:r-seg'); + }); + + it('re-reads the thread when the append fails, so the core row still renders (#6034)', async () => { + vi.mocked(threadApi.appendMessage).mockRejectedValueOnce(new Error('rpc timeout')); + vi.mocked(threadApi.getThreadMessages).mockResolvedValueOnce({ + messages: [ + { + id: 'agent:r-lost', + content: 'the answer the core stored', + type: 'text', + sender: 'agent', + createdAt: new Date().toISOString(), + extraMetadata: {}, + }, + ], + } as unknown as Awaited>); + + const listeners = renderProvider(); + + act(() => { + listeners.onDone?.({ + thread_id: 't-lost', + request_id: 'r-lost', + full_response: 'the answer the core stored', + rounds_used: 1, + }); + }); + + // The failed append used to end the story with a debug log and no + // assistant row. Re-reading the thread is what brings the core's copy + // into the cache without the user having to ask again. + await waitFor(() => expect(threadApi.getThreadMessages).toHaveBeenCalledWith('t-lost')); + await waitFor(() => + expect( + (store.getState().thread.messagesByThreadId['t-lost'] ?? []).some( + m => m.id === 'agent:r-lost' + ) + ).toBe(true) + ); + }); + it('stores a parked plan review from the plan_review_request event', () => { const listeners = renderProvider(); act(() => { @@ -1074,6 +1175,91 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria }); }); + describe('socket reconnect recovery (#6034)', () => { + it('rejoins the rooms of interrupted threads and re-reads them once the socket returns', async () => { + vi.mocked(threadApi.getThreadMessages).mockResolvedValue({ + messages: [], + } as unknown as Awaited>); + + renderProvider(); + + // A turn is in flight on a thread the user is not looking at. + act(() => { + store.dispatch(setActiveThread('t-away')); + }); + vi.mocked(threadApi.getThreadMessages).mockClear(); + vi.mocked(socketService.subscribeThread).mockClear(); + + // The socket drops. The provider clears every active marker so the + // composer unlocks, which is also what erases the list the reconnect + // handler would have re-subscribed from. + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'disconnected' })); + }); + expect(store.getState().thread.activeThreadIds).toEqual({}); + + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'connected' })); + }); + + // Rejoin the room, so a turn still running can still reach us under the + // new client_id, and re-read the thread, so a turn that finished during + // the gap is not invisible until the thread is reselected. + await waitFor(() => expect(socketService.subscribeThread).toHaveBeenCalledWith('t-away')); + await waitFor(() => expect(threadApi.getThreadMessages).toHaveBeenCalledWith('t-away')); + }); + + it('retries a thread whose room join never emitted, on the next connection', async () => { + vi.mocked(threadApi.getThreadMessages).mockResolvedValue({ + messages: [], + } as unknown as Awaited>); + // The socket dropped again between `connected` and this effect, so the + // emit never went out. Forgetting the thread here would strand it until + // the user reselected it. + vi.mocked(socketService.subscribeThread).mockResolvedValueOnce(false); + + renderProvider(); + act(() => { + store.dispatch(setActiveThread('t-flaky')); + }); + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'disconnected' })); + }); + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'connected' })); + }); + await waitFor(() => expect(socketService.subscribeThread).toHaveBeenCalledWith('t-flaky')); + + vi.mocked(socketService.subscribeThread).mockClear(); + vi.mocked(socketService.subscribeThread).mockResolvedValue(true); + + // Second reconnect: the thread is still pending, so it is retried. + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'disconnected' })); + }); + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'connected' })); + }); + await waitFor(() => expect(socketService.subscribeThread).toHaveBeenCalledWith('t-flaky')); + }); + + it('does nothing on a connect with no interrupted threads', async () => { + renderProvider(); + vi.mocked(socketService.subscribeThread).mockClear(); + + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'disconnected' })); + }); + act(() => { + store.dispatch(setStatusForUser({ userId: '__pending__', status: 'connected' })); + }); + + // A blip with nothing in flight must not re-read every thread the user + // has ever opened. + expect(socketService.subscribeThread).not.toHaveBeenCalled(); + }); + }); + describe('mid-turn streaming invariants', () => { it('reconciles missing segment events from chat_done.full_response', async () => { const listeners = renderProvider(); diff --git a/app/src/services/__tests__/socketService.subscribeThread.test.ts b/app/src/services/__tests__/socketService.subscribeThread.test.ts new file mode 100644 index 0000000000..c85173f7c6 --- /dev/null +++ b/app/src/services/__tests__/socketService.subscribeThread.test.ts @@ -0,0 +1,84 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { socketService } from '../socketService'; + +/** + * `subscribeThread` is what orders the reconnect recovery in #6034: the chat + * runtime re-reads a thread only after the room join is acknowledged, and it + * keeps a thread queued for the next connection when the join never happened. + * Both of those decisions read this function's resolved value, so the three + * outcomes are pinned here rather than through the provider (which mocks this + * module out entirely). + */ + +type FakeSocket = { connected: boolean; emit: ReturnType }; + +/** Install a stand-in for the private socket the singleton holds. */ +function withSocket(socket: FakeSocket | null) { + (socketService as unknown as { socket: FakeSocket | null }).socket = socket; +} + +afterEach(() => { + withSocket(null); + vi.useRealTimers(); +}); + +describe('socketService.subscribeThread (#6034)', () => { + it('resolves true once the server acknowledges the room join', async () => { + const emit = vi.fn((_event: string, _payload: unknown, ack: () => void) => ack()); + withSocket({ connected: true, emit }); + + await expect(socketService.subscribeThread('t-1')).resolves.toBe(true); + expect(emit).toHaveBeenCalledWith( + 'thread:subscribe', + { thread_id: 't-1' }, + expect.any(Function) + ); + }); + + it('resolves false without emitting when the socket is not connected', async () => { + const emit = vi.fn(); + withSocket({ connected: false, emit }); + + // The caller keeps the thread queued for the next connection on a false — + // clearing it here is what stranded a thread whose socket dropped again. + await expect(socketService.subscribeThread('t-2')).resolves.toBe(false); + expect(emit).not.toHaveBeenCalled(); + }); + + it('resolves false when the acknowledgement never arrives', async () => { + vi.useFakeTimers(); + // A core without the ack handler never calls back. Recovery must still + // proceed rather than hang, so the wait is bounded. + const emit = vi.fn(); + withSocket({ connected: true, emit }); + + const pending = socketService.subscribeThread('t-3', 50); + await vi.advanceTimersByTimeAsync(60); + await expect(pending).resolves.toBe(false); + expect(emit).toHaveBeenCalledTimes(1); + }); + + it('resolves false for an empty thread id and sends nothing', async () => { + const emit = vi.fn(); + withSocket({ connected: true, emit }); + + await expect(socketService.subscribeThread('')).resolves.toBe(false); + expect(emit).not.toHaveBeenCalled(); + }); + + it('does not resolve twice when the ack lands after the timeout', async () => { + vi.useFakeTimers(); + let late: (() => void) | undefined; + const emit = vi.fn((_event: string, _payload: unknown, ack: () => void) => { + late = ack; + }); + withSocket({ connected: true, emit }); + + const pending = socketService.subscribeThread('t-4', 50); + await vi.advanceTimersByTimeAsync(60); + late?.(); + + await expect(pending).resolves.toBe(false); + }); +}); diff --git a/app/src/services/socketService.ts b/app/src/services/socketService.ts index 06673892f4..7bceae779e 100644 --- a/app/src/services/socketService.ts +++ b/app/src/services/socketService.ts @@ -475,6 +475,46 @@ class SocketService { } } + /** + * Join one thread's event room. + * + * The reconnect handler re-subscribes from `activeThreadIds`, which the chat + * runtime clears when the socket drops — so a turn left in flight on a thread + * the user had navigated away from has no room to be delivered into once a new + * `client_id` is issued, and its `chat_done` reaches nobody. `ChatRuntimeProvider` + * calls this for the threads it remembers across that gap (#6034). + * + * Emitting the room join directly rather than through {@link emit} keeps a + * disconnected call quiet: re-subscription is what the `connect` handler + * already does, so a warning here would only be noise. + */ + subscribeThread(threadId: string, timeoutMs = 3000): Promise { + if (!threadId || !this.socket?.connected) return Promise.resolve(false); + socketLog('Subscribing to thread room', { threadId }); + const socket = this.socket; + return new Promise(resolve => { + let settled = false; + const finish = (joined: boolean) => { + if (settled) return; + settled = true; + resolve(joined); + }; + // A caller that reads the thread after this resolves cannot race the + // join: the server acknowledges only once the socket is in the room. + // The timeout keeps a server that never acks (an older core) from + // stalling recovery — the read still happens, just without the ordering + // guarantee, which is exactly the pre-ack behaviour. + const timer = setTimeout(() => { + socketWarn('Thread room subscription not acknowledged', { threadId }); + finish(false); + }, timeoutMs); + socket.emit('thread:subscribe', { thread_id: threadId }, () => { + clearTimeout(timer); + finish(true); + }); + }); + } + /** * Listen to an event from the server */ diff --git a/app/src/types/userError.ts b/app/src/types/userError.ts index 49b7293651..6b1e113d6d 100644 --- a/app/src/types/userError.ts +++ b/app/src/types/userError.ts @@ -51,7 +51,16 @@ export type UserErrorKind = * rebuilt tree repopulates by re-syncing sources, which is why the action * deep-links to Brain's sync tab rather than any settings screen. */ - | 'memory_store_corrupt'; + | 'memory_store_corrupt' + /** + * A reply the agent finished could not be shown: neither the core's write + * nor the client's append left a row, and re-reading the thread did not + * find one (#6034). Distinct from every entry above because nothing is + * misconfigured — the turn ran and was paid for, and the only useful next + * step is to ask again, so the action is `dismiss` rather than a settings + * deep link. + */ + | 'reply_delivery_failed'; /** Where the failure originated, for grouping/labelling (privacy-safe). */ export type UserErrorScope = diff --git a/docs/TEST-COVERAGE-MATRIX.md b/docs/TEST-COVERAGE-MATRIX.md index 902b1f0021..e3f19ccbcc 100644 --- a/docs/TEST-COVERAGE-MATRIX.md +++ b/docs/TEST-COVERAGE-MATRIX.md @@ -191,6 +191,7 @@ Canonical mapping of every product feature to its test source(s). Drives gap-fil | 4.2.8 | Composer attachments (image / video / document; drag-drop + paste) | VU | `app/src/lib/attachments.test.ts`, `app/src/components/chat/__tests__/ChatComposer.test.tsx`, `app/src/pages/__tests__/Conversations.attachments.test.tsx` | 🟡 | Attach affordance gated on the resolved vision tier (images/video need vision; documents flow on any model); video is sampled into still frames client-side and forwarded through the existing `[IMAGE:]` vision path; drag-drop + clipboard-paste reuse the picker ingest. VU covers MIME/kind/limits/marker building + drag-drop + paste; real video decode and the frames→vision round-trip are manual-smoke only (jsdom has no video codec). WD E2E is a follow-up | | 4.2.9 | Share cards ("Look what my agent did" post to X / LinkedIn) | VU | `app/src/features/share/shareContent.test.ts`, `app/src/features/share/shareTargets.test.ts`, `app/src/features/share/shareCaption.test.ts`, `app/src/features/share/shareCard.test.ts`, `app/src/features/share/ShareMessageButton.test.tsx`, `app/src/features/share/ShareCardModal.test.tsx` | ✅ | Hover Share button under completed agent messages opens a Canvas 2D branded card; LLM-drafted headline/caption via `inference_agent_chat_simple` with deterministic offline fallback; X/LinkedIn intent URLs + copy/save PNG; secrets/paths/emails redacted before render. MVP wires only the completed-task trigger; workflow-run + Human-tab triggers are follow-ups. WD E2E deferred | +| 4.2.10 | Durable agent reply (core stores it before announcing `chat_done`) | RU+VU | `src/openhuman/web_chat/reply_persistence_tests.rs`, `src/openhuman/web_chat/presentation_tests.rs`, `app/src/providers/__tests__/ChatRuntimeProvider.test.tsx` | ✅ | `deliver_response` persists an unsegmented reply under `agent:` before publishing the terminal event, so a dropped socket, a failed `threads_message_append` or a reloaded webview costs a repaint rather than the answer; the client's append collapses onto that row by id. RU covers the write, its idempotency, the empty-reply and missing-thread paths, and that delivery still announces when the store refuses. VU covers the mirrored id, the segmented exception, the refetch that recovers a failed append, and the reconnect that rejoins orphaned thread rooms. WD E2E (kill the append mid-turn) is a follow-up | ### 4.3 Tool Invocation diff --git a/gitbooks/developing/architecture/agent-harness.md b/gitbooks/developing/architecture/agent-harness.md index 78a27cc951..86842db435 100644 --- a/gitbooks/developing/architecture/agent-harness.md +++ b/gitbooks/developing/architecture/agent-harness.md @@ -288,7 +288,7 @@ The child run itself still uses the same runner: `wait_subagent` and `steer_subagent` accept either the durable `subagent_session_id` or the transient `task_id`; durable ids are preferred across turns. `list_subagents` shows reusable children for the current parent thread, and `close_subagent` marks a worker non-reusable and cancels it if it is still running. Inline blocking is explicit via `blocking: true`; it is no longer the default. -The synthesized archetype delegations (`delegate_*`, `build_workflow`, and the other `delegate_name` tools) follow the same contract: they route through the durable async path by default, returning an `[async_subagent_ref]` (with `subagent_session_id` + `task_id`) immediately, and the finished result is inserted into the parent chat as a new system turn via `background_completions`/`background_delivery`. That delivery turn (`task_dispatcher::run_system_turn_on_thread`, the same runner autonomous task sessions use) persists its own closing message — `sender: "agent"`, id `agent:`, `extraMetadata.requestId = run_id` — **before** it emits `chat_done` as `client_id: "system"`; the frontend reuses that id for system turns, so its usual `chat_done` append collapses onto the same row (the conversation store is idempotent for these deterministic `agent:`-prefixed ids; every other id is UUID-fresh and keeps the constant-time append path) instead of persisting the delivered result a second time (#5933). They fall back to inline blocking automatically when there is no parent agent turn or no chat thread to deliver into (cron/CLI), or when `blocking: true` is passed. Cross-turn continuity comes from three pieces: the per-turn `[active_subagents]` roster merges the live in-memory registry with the durable `subagent_sessions` store (so a cold-booted orchestrator still sees earlier workers); `continue_subagent` falls back from pause checkpoints to the durable store, resuming an idle worker with its persisted history; and a `workflow_proposal` payload found in a finished child's history is persisted as a parent-thread message (`extraMetadata.scope = "workflow_proposal"`) that the frontend rehydrates into the proposal card on thread load. +The synthesized archetype delegations (`delegate_*`, `build_workflow`, and the other `delegate_name` tools) follow the same contract: they route through the durable async path by default, returning an `[async_subagent_ref]` (with `subagent_session_id` + `task_id`) immediately, and the finished result is inserted into the parent chat as a new system turn via `background_completions`/`background_delivery`. That delivery turn (`task_dispatcher::run_system_turn_on_thread`, the same runner autonomous task sessions use) persists its own closing message — `sender: "agent"`, id `agent:`, `extraMetadata.requestId = run_id` — **before** it emits `chat_done` as `client_id: "system"`; the frontend reuses that id, so its usual `chat_done` append collapses onto the same row (the conversation store is idempotent for these deterministic `agent:`-prefixed ids; every other id is UUID-fresh and keeps the constant-time append path) instead of persisting the delivered result a second time (#5933). **Interactive turns follow the same contract since #6034**: `web_chat::presentation::deliver_response` stores the reply under `agent:` before publishing `chat_done`, so an answer the core produced exists on disk whether or not a client is there to receive the announcement — a dropped socket, a failed append or a reloaded webview costs a repaint, not the reply. The exception is a segmented delivery, where the client owns one row per segment and the core stores none; the frontend keeps generated ids there for exactly that reason. They fall back to inline blocking automatically when there is no parent agent turn or no chat thread to deliver into (cron/CLI), or when `blocking: true` is passed. Cross-turn continuity comes from three pieces: the per-turn `[active_subagents]` roster merges the live in-memory registry with the durable `subagent_sessions` store (so a cold-booted orchestrator still sees earlier workers); `continue_subagent` falls back from pause checkpoints to the durable store, resuming an idle worker with its persisted history; and a `workflow_proposal` payload found in a finished child's history is persisted as a parent-thread message (`extraMetadata.scope = "workflow_proposal"`) that the frontend rehydrates into the proposal card on thread load. ### Spawn hierarchy and tiers diff --git a/src/core/socketio.rs b/src/core/socketio.rs index 7b453d9816..2efa67f297 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -12,7 +12,7 @@ use serde_json::Value; #[cfg(feature = "http-server")] use serde_json::json; #[cfg(feature = "http-server")] -use socketioxide::extract::{Data, SocketRef, TryData}; +use socketioxide::extract::{AckSender, Data, SocketRef, TryData}; #[cfg(feature = "http-server")] use socketioxide::SocketIo; @@ -428,6 +428,13 @@ struct ThreadSubscribePayload { thread_id: String, } +/// Reply to `thread:subscribe`, so a client can order a read after the join. +#[cfg(feature = "http-server")] +#[derive(Debug, Serialize)] +struct ThreadSubscribeAck { + joined: bool, +} + /// Attaches the Socket.IO layer to the Axum router and sets up event handlers. /// /// It configures: @@ -485,7 +492,7 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) { log::info!("[socketio] client connected id={client_id} (authenticated)"); // Join a room named after the client ID for targeted event delivery. - join_room_logged(&socket, &client_id, &client_id); + let _ = join_room_logged(&socket, &client_id, &client_id); // Also auto-join the "system" room so every connected client // receives broadcast-style events that aren't tied to a // specific chat thread. Today this covers proactive messages @@ -494,7 +501,7 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) { // emits with `client_id = "system"` — see `emit_web_channel_event`. // If this join fails the welcome message silently disappears, // so we log both success and failure for diagnosability. - join_room_logged(&socket, "system", &client_id); + let _ = join_room_logged(&socket, "system", &client_id); let ready_payload = json!({ "sid": client_id }); log::debug!("[socketio] emit event=ready to_client={}", socket.id); let _ = socket.emit("ready", &ready_payload); @@ -668,19 +675,36 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) { // frontend emits this on connect/reconnect for the active thread, so // the new socket re-joins the thread room and keeps receiving the // stream. Membership is dropped automatically on disconnect. + // + // The join is acknowledged so a client can *order* work against it. + // A reconnecting client re-reads the thread to pick up a reply that + // landed while it was away (#6034); firing that read before the join + // is processed leaves a window where the read misses the row and the + // turn's `chat_done` is emitted to a room this socket has not joined + // yet, so the reply stays invisible until a manual reload. The ack + // closes it. Clients that ignore the ack are unaffected — an unused + // acknowledgement is inert. socket.on( "thread:subscribe", - |socket: SocketRef, Data(payload): Data| async move { + |socket: SocketRef, Data(payload): Data, ack: AckSender| async move { if !socket_is_authed(&socket) { drop_unauthed(&socket, "thread:subscribe from unauthenticated socket"); return; } let thread_id = payload.thread_id.trim(); if thread_id.is_empty() { + // Still acknowledge: a client awaiting this must not be + // left hanging on its own malformed payload. + ack.send(&ThreadSubscribeAck { joined: false }).ok(); return; } let room = format!("thread:{thread_id}"); - join_room_logged(&socket, &room, &socket.id.to_string()); + // Report what actually happened. Acknowledging a join that + // failed is worse than not acknowledging at all: the client + // stops queueing the thread for retry and reads on the + // strength of a room it is not in. + let joined = join_room_logged(&socket, &room, &socket.id.to_string()); + ack.send(&ThreadSubscribeAck { joined }).ok(); }, ); }, @@ -1365,11 +1389,23 @@ pub(crate) fn channel_connection_update_payload( /// so both the happy and error paths are logged with enough context /// (room name + client id) to diagnose missing welcome messages from /// logs alone. +/// +/// Returns whether the socket is actually in the room. Callers that only log +/// may ignore it; a caller that *tells the client* it joined must not — a +/// client told it is in a room it never joined reads the thread, waits for +/// events that will never be routed to it, and reproduces the invisible-reply +/// bug this room exists to prevent (#6034). #[cfg(feature = "http-server")] -fn join_room_logged(socket: &SocketRef, room: &str, client_id: &str) { +fn join_room_logged(socket: &SocketRef, room: &str, client_id: &str) -> bool { match socket.join(room.to_string()) { - Ok(()) => log::debug!("[socketio] joined room '{room}' for client {client_id}"), - Err(e) => log::warn!("[socketio] failed to join room '{room}' for client {client_id}: {e}"), + Ok(()) => { + log::debug!("[socketio] joined room '{room}' for client {client_id}"); + true + } + Err(e) => { + log::warn!("[socketio] failed to join room '{room}' for client {client_id}: {e}"); + false + } } } diff --git a/src/openhuman/flows/ops_part_10.rs b/src/openhuman/flows/ops_part_10.rs index 0921e89a56..7baf4b3202 100644 --- a/src/openhuman/flows/ops_part_10.rs +++ b/src/openhuman/flows/ops_part_10.rs @@ -23,6 +23,10 @@ async fn finalize_flow_stream( // Builder/scout turns don't surface in the chat footer; their // token/cost spend is still captured by the global cost tracker. None, + // No workspace in scope on this path, so the viewing client + // stays the only persister of a flow turn's reply — unchanged + // from before #6034, which covered the chat surfaces. + None, ) .await; } diff --git a/src/openhuman/web_chat/mod.rs b/src/openhuman/web_chat/mod.rs index b751915362..7e140267ff 100644 --- a/src/openhuman/web_chat/mod.rs +++ b/src/openhuman/web_chat/mod.rs @@ -4,6 +4,7 @@ mod ops; // standalone `presentation` provider — it is the web channel's delivery formatter). pub mod presentation; mod progress_bridge; +mod reply_persistence; mod run_task; mod schemas; mod session; diff --git a/src/openhuman/web_chat/ops_part_02.rs b/src/openhuman/web_chat/ops_part_02.rs index cc9fb96179..cf06c85501 100644 --- a/src/openhuman/web_chat/ops_part_02.rs +++ b/src/openhuman/web_chat/ops_part_02.rs @@ -385,6 +385,9 @@ pub async fn start_chat( &user_message, &chat_result.citations, chat_result.usage.as_ref(), + // The workspace the turn ran in, so the reply is stored + // there before it is announced (#6034). + Some(chat_result.workspace_dir.as_path()), ) .await; } diff --git a/src/openhuman/web_chat/ops_part_03.rs b/src/openhuman/web_chat/ops_part_03.rs index 0b659db093..38abaa1d7d 100644 --- a/src/openhuman/web_chat/ops_part_03.rs +++ b/src/openhuman/web_chat/ops_part_03.rs @@ -68,6 +68,9 @@ async fn spawn_parallel_turn( &user_message, &chat_result.citations, chat_result.usage.as_ref(), + // The workspace the turn ran in, so the reply is stored + // there before it is announced (#6034). + Some(chat_result.workspace_dir.as_path()), ) .await; } diff --git a/src/openhuman/web_chat/presentation.rs b/src/openhuman/web_chat/presentation.rs index b0921856eb..d95e117d38 100644 --- a/src/openhuman/web_chat/presentation.rs +++ b/src/openhuman/web_chat/presentation.rs @@ -50,6 +50,12 @@ fn usage_payload(usage: Option<&LastTurnUsage>) -> Option { /// paragraphs into `chat_segment` messages duplicates tool/reasoning parts and /// turns one answer into several bubbles, so this path always emits exactly one /// `chat_done` with the model's original text. +/// +/// `workspace_dir` is where the reply is stored **before** it is announced, so a +/// client that never receives the `chat_done` — or receives it and fails to +/// append — is not the difference between the answer existing and not existing +/// (#6034). Pass `None` from a caller that has no workspace in scope; delivery +/// then behaves exactly as it did when the renderer was the only writer. pub(crate) async fn deliver_response( client_id: &str, thread_id: &str, @@ -58,6 +64,7 @@ pub(crate) async fn deliver_response( user_message: &str, citations: &[crate::openhuman::memory::agent::memory_loader::MemoryCitation], usage: Option<&LastTurnUsage>, + workspace_dir: Option<&std::path::Path>, ) { let usage_payload = usage_payload(usage); @@ -75,6 +82,53 @@ pub(crate) async fn deliver_response( let reaction_emoji = reaction_handle.await.unwrap_or(None); if segments.len() <= 1 { + // Store the answer before announcing it. Ordering is the whole point: + // once the row is on disk, a `chat_done` that is never delivered, never + // painted, or never persisted by the client costs the user a repaint, + // not the reply (#6034). Only this single-bubble branch persists — the + // segmented branch below hands the client one row per segment to write, + // and a full-text row beside those would read as a duplicate answer. + if let Some(dir) = workspace_dir { + // The store appends under a process-wide lock and fsyncs, and its + // existence check folds the whole threads log (#5156) — blocking + // work that has no business holding a runtime worker while a turn + // is settling. Hand it to the blocking pool and await the handle, + // which keeps the ordering this whole change rests on. + let (dir, thread, request, reply, cites) = ( + dir.to_path_buf(), + thread_id.to_string(), + request_id.to_string(), + full_response.to_string(), + citations.to_vec(), + ); + let persisted = tokio::task::spawn_blocking(move || { + super::reply_persistence::persist_delivered_reply( + &dir, &thread, &request, &reply, &cites, + ) + }) + .await; + match persisted { + Ok(Ok(stored)) => { + if stored { + log::debug!( + "[web-channel] persisted reply before announcing it thread_id={thread_id} request_id={request_id}" + ); + } + } + // Deliberately non-fatal: announce anyway. The client's own + // append still persists the reply in the common case, and a + // storage failure must not also cost the user the delivery. + Ok(Err(err)) => log::warn!( + "[web-channel] could not persist reply before announcing it \ + thread_id={thread_id} request_id={request_id} error={err}" + ), + Err(err) => log::warn!( + "[web-channel] reply persistence task did not run \ + thread_id={thread_id} request_id={request_id} error={err}" + ), + } + } + // Single bubble — emit chat_done directly. publish_chat_done( client_id, diff --git a/src/openhuman/web_chat/presentation_test_support_tests.rs b/src/openhuman/web_chat/presentation_test_support_tests.rs index aee2b717ce..6c45ac43af 100644 --- a/src/openhuman/web_chat/presentation_test_support_tests.rs +++ b/src/openhuman/web_chat/presentation_test_support_tests.rs @@ -19,6 +19,29 @@ pub async fn deliver_response_for_test( full_response: &str, user_message: &str, citations: &[MemoryCitation], +) { + deliver_response_in_workspace_for_test( + client_id, + thread_id, + request_id, + full_response, + user_message, + citations, + None, + ) + .await; +} + +/// `deliver_response` with an explicit workspace, so a test can assert the +/// reply reached disk before the turn was announced (#6034). +pub async fn deliver_response_in_workspace_for_test( + client_id: &str, + thread_id: &str, + request_id: &str, + full_response: &str, + user_message: &str, + citations: &[MemoryCitation], + workspace_dir: Option<&std::path::Path>, ) { super::deliver_response( client_id, @@ -28,6 +51,7 @@ pub async fn deliver_response_for_test( user_message, citations, None, + workspace_dir, ) .await; } diff --git a/src/openhuman/web_chat/presentation_tests.rs b/src/openhuman/web_chat/presentation_tests.rs index 270f6bd42f..031f197786 100644 --- a/src/openhuman/web_chat/presentation_tests.rs +++ b/src/openhuman/web_chat/presentation_tests.rs @@ -309,3 +309,107 @@ fn single_bubble_delivery_emits_one_unsegmented_chat_done_without_reaction() { assert_eq!(done.reaction_emoji, None); assert!(done.usage.is_none()); } + +// ── Delivery persists before it announces (#6034) ─────────────────────── + +#[tokio::test] +async fn delivery_stores_the_reply_before_announcing_it() { + use crate::openhuman::memory::conversations::{self, CreateConversationThread}; + + let ws = std::env::temp_dir().join(format!("deliver-persist-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&ws).unwrap(); + conversations::ensure_thread( + ws.clone(), + CreateConversationThread { + id: "t-deliver".to_string(), + title: "Chat".to_string(), + created_at: chrono::Utc::now().to_rfc3339(), + parent_thread_id: None, + labels: None, + personality_id: None, + }, + ) + .expect("thread created"); + + let citation = crate::openhuman::memory::agent::memory_loader::MemoryCitation { + id: "mem-deliver".to_string(), + key: "summary-source".to_string(), + namespace: None, + score: Some(0.8), + timestamp: "2026-09-04T00:00:00Z".to_string(), + snippet: "source snippet".to_string(), + }; + test_support::deliver_response_in_workspace_for_test( + "client-1", + "t-deliver", + "req-deliver", + "Here is the summary you asked for.", + "summarise this", + &[citation], + Some(ws.as_path()), + ) + .await; + + // `deliver_response` returns only after the terminal event is published, so + // a row present here proves the write happened no later than the announce. + // A client that never receives that event, or fails to append it, no longer + // decides whether the reply exists. + let messages = conversations::get_messages(ws.clone(), "t-deliver").expect("messages"); + assert_eq!(messages.len(), 1, "delivery must leave exactly one row"); + assert_eq!(messages[0].id, "agent:req-deliver"); + assert_eq!(messages[0].content, "Here is the summary you asked for."); + assert_eq!(messages[0].sender, "agent"); + // The client's append is deduped onto this row, so the citations it would + // have written must already be here or the chips render empty. + assert_eq!( + messages[0].extra_metadata["citations"][0]["id"], + "mem-deliver" + ); +} + +#[tokio::test] +async fn delivery_still_announces_when_the_reply_cannot_be_stored() { + // The thread does not exist, so the store refuses the write. Delivery must + // continue regardless: a storage failure that also swallowed the + // announcement would turn a recoverable problem into a dead turn, and the + // client's own append is still a working fallback. + let ws = std::env::temp_dir().join(format!("deliver-nothread-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&ws).unwrap(); + + test_support::deliver_response_in_workspace_for_test( + "client-3", + "absent-thread", + "req-absent", + "an answer with nowhere to go", + "hi", + &[], + Some(ws.as_path()), + ) + .await; + + // `get_messages` answers `Ok(vec![])` for a thread it has never seen — only + // `append_message` refuses one — so absence is what proves the write was + // rejected and swallowed rather than silently creating a thread. + let messages = + crate::openhuman::memory::conversations::get_messages(ws.clone(), "absent-thread") + .expect("reading an unknown thread is not an error"); + assert!( + messages.is_empty(), + "the thread was never created, so nothing should have been written" + ); +} + +#[tokio::test] +async fn delivery_without_a_workspace_persists_nothing() { + // Callers with no workspace in scope (the flows stream finalizer) keep the + // pre-#6034 behaviour: the viewing client stays the only persister. + test_support::deliver_response_for_test( + "client-2", + "t-none", + "req-none", + "nothing to store", + "hi", + &[], + ) + .await; +} diff --git a/src/openhuman/web_chat/reply_persistence.rs b/src/openhuman/web_chat/reply_persistence.rs new file mode 100644 index 0000000000..b449c4cf7f --- /dev/null +++ b/src/openhuman/web_chat/reply_persistence.rs @@ -0,0 +1,89 @@ +//! Durable storage for the reply a web-channel turn is about to announce. +//! +//! An interactive turn's answer used to reach disk only if the viewing client +//! persisted the `chat_done` it received. That made the renderer the single +//! writer of a reply the core had already produced: one failed +//! `threads_message_append`, one socket reconnect that moved the client to a new +//! `client_id`, or one webview reload during a long turn, and the answer was +//! gone from the thread while the agent's own session history still held it +//! (#6034). +//! +//! Core-initiated turns never had that exposure — `task_session::append_final` +//! writes their closing row *before* the run announces it, and the client's +//! append collapses onto that row because both derive the same deterministic id +//! (#5933). This module gives interactive turns the same guarantee, so a reply +//! survives whatever the renderer was doing at the moment it was delivered. +//! +//! The id is [`run_reply_message_id`] over the turn's `request_id`, which is +//! what makes the second write a no-op rather than a duplicate bubble: the +//! conversation store is idempotent for exactly this id shape. + +use std::path::Path; + +use serde_json::json; + +use crate::openhuman::memory::agent::memory_loader::MemoryCitation; +use crate::openhuman::memory::conversations::{self, run_reply_message_id, ConversationMessage}; + +/// Metadata scope stamped on a reply persisted by the web-channel delivery path. +/// +/// Distinguishes it from `autonomous_task_result` (the same shape written by +/// `task_session::append_final`) when reading a thread back. +const REPLY_SCOPE: &str = "web_chat_reply"; + +/// Persist a delivered reply under the id the announcing client will reuse. +/// +/// Returns `Ok(false)` when there was nothing to store (an empty or +/// whitespace-only response — the same guard `task_session::append_final` +/// applies), `Ok(true)` when the row is on disk, and `Err` when the store +/// refused the write (most commonly a thread that does not exist yet). +/// +/// Callers must treat an `Err` as non-fatal and announce the reply anyway: the +/// client's own append is still a working fallback, and losing the delivery on +/// top of losing the row would turn a storage problem into a visibly dead turn. +/// +/// **This row's metadata is what the reader ends up with, so it must carry +/// everything the client would have written.** Because the store is idempotent +/// for this id, the client's later append returns *this* row rather than its +/// own, and whatever is missing here is missing from the rendered message — +/// which is how citations vanished in review. `citations` therefore mirrors the +/// shape `chatDoneExtraMetadata` builds, and any field added to that helper +/// belongs here too. +pub(crate) fn persist_delivered_reply( + workspace_dir: &Path, + thread_id: &str, + request_id: &str, + full_response: &str, + citations: &[MemoryCitation], +) -> Result { + let content = full_response.trim(); + if content.is_empty() { + return Ok(false); + } + let mut extra_metadata = json!({ + "scope": REPLY_SCOPE, + "requestId": request_id, + }); + if !citations.is_empty() { + // Same key and payload the client stamps, so a row read back from disk + // renders identical chips to one the client had appended itself. + extra_metadata["citations"] = json!(citations); + } + conversations::append_message( + workspace_dir.to_path_buf(), + thread_id, + ConversationMessage { + id: run_reply_message_id(request_id), + content: content.to_string(), + message_type: "text".to_string(), + extra_metadata, + sender: "agent".to_string(), + created_at: chrono::Utc::now().to_rfc3339(), + }, + )?; + Ok(true) +} + +#[cfg(test)] +#[path = "reply_persistence_tests.rs"] +mod tests; diff --git a/src/openhuman/web_chat/reply_persistence_tests.rs b/src/openhuman/web_chat/reply_persistence_tests.rs new file mode 100644 index 0000000000..b1c0ff5f0e --- /dev/null +++ b/src/openhuman/web_chat/reply_persistence_tests.rs @@ -0,0 +1,135 @@ +use std::path::PathBuf; + +use super::persist_delivered_reply; +use crate::openhuman::memory::agent::memory_loader::MemoryCitation; +use crate::openhuman::memory::conversations::{self, CreateConversationThread}; + +fn temp_ws() -> PathBuf { + let dir = std::env::temp_dir().join(format!("web-chat-reply-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +fn seed_thread(ws: &PathBuf, thread_id: &str) { + conversations::ensure_thread( + ws.clone(), + CreateConversationThread { + id: thread_id.to_string(), + title: "Chat".to_string(), + created_at: chrono::Utc::now().to_rfc3339(), + parent_thread_id: None, + labels: None, + personality_id: None, + }, + ) + .expect("thread created"); +} + +#[test] +fn persists_the_reply_under_the_id_the_client_will_reuse() { + let ws = temp_ws(); + seed_thread(&ws, "t-1"); + + let stored = persist_delivered_reply(&ws, "t-1", "req-1", "Done, the draft is updated.", &[]) + .expect("append succeeds"); + assert!(stored, "a non-empty reply must report that it was stored"); + + let messages = conversations::get_messages(ws.clone(), "t-1").expect("messages"); + assert_eq!(messages.len(), 1); + // The id is what makes the client's own append collapse onto this row + // instead of adding a duplicate bubble (#5933 / #6034). + assert_eq!(messages[0].id, "agent:req-1"); + assert_eq!(messages[0].sender, "agent"); + assert_eq!(messages[0].content, "Done, the draft is updated."); + assert_eq!(messages[0].extra_metadata["scope"], "web_chat_reply"); + assert_eq!(messages[0].extra_metadata["requestId"], "req-1"); +} + +#[test] +fn a_second_write_of_the_same_turn_does_not_add_a_row() { + let ws = temp_ws(); + seed_thread(&ws, "t-2"); + + persist_delivered_reply(&ws, "t-2", "req-2", "First", &[]).expect("first append"); + // The client persists the same reply from the `chat_done` it received. The + // store's idempotency for deterministic ids is what keeps the thread at one + // row; assert the second write here so a change to the id shape (which + // would silently opt out of that lookup) fails loudly. + persist_delivered_reply(&ws, "t-2", "req-2", "First", &[]).expect("second append"); + + let messages = conversations::get_messages(ws.clone(), "t-2").expect("messages"); + assert_eq!(messages.len(), 1, "one turn must never leave two rows"); +} + +#[test] +fn an_empty_reply_is_not_stored() { + let ws = temp_ws(); + seed_thread(&ws, "t-3"); + + let stored = persist_delivered_reply(&ws, "t-3", "req-3", " \n ", &[]).expect("no error"); + assert!(!stored, "an empty reply reports that nothing was stored"); + assert!(conversations::get_messages(ws.clone(), "t-3") + .expect("messages") + .is_empty()); +} + +#[test] +fn content_is_trimmed_the_way_the_autonomous_path_trims_it() { + let ws = temp_ws(); + seed_thread(&ws, "t-4"); + + persist_delivered_reply(&ws, "t-4", "req-4", " padded reply\n", &[]).expect("append"); + + let messages = conversations::get_messages(ws.clone(), "t-4").expect("messages"); + assert_eq!(messages[0].content, "padded reply"); +} + +#[test] +fn a_missing_thread_is_reported_rather_than_silently_dropped() { + let ws = temp_ws(); + + let err = persist_delivered_reply(&ws, "nope", "req-5", "text", &[]) + .expect_err("a missing thread must not look like a successful store"); + assert!(err.contains("nope"), "error names the thread: {err}"); +} + +#[test] +fn citations_ride_on_the_authoritative_row() { + let ws = temp_ws(); + seed_thread(&ws, "t-6"); + + let citation = MemoryCitation { + id: "mem-1".to_string(), + key: "draft-location".to_string(), + namespace: Some("notes".to_string()), + score: Some(0.91), + timestamp: "2026-09-04T00:00:00Z".to_string(), + snippet: "The draft lives in Notion.".to_string(), + }; + persist_delivered_reply(&ws, "t-6", "req-6", "Updated the draft.", &[citation]) + .expect("append"); + + // The client's append is deduped onto this row, so whatever is missing here + // is missing from the rendered message — citation chips included. Losing + // them was a real regression caught in review of #6034. + let messages = conversations::get_messages(ws.clone(), "t-6").expect("messages"); + let cites = messages[0].extra_metadata["citations"] + .as_array() + .expect("citations are stored as an array"); + assert_eq!(cites.len(), 1); + assert_eq!(cites[0]["id"], "mem-1"); +} + +#[test] +fn a_reply_without_citations_stores_no_citations_key() { + let ws = temp_ws(); + seed_thread(&ws, "t-7"); + + persist_delivered_reply(&ws, "t-7", "req-7", "No sources for this one.", &[]).expect("append"); + + let messages = conversations::get_messages(ws.clone(), "t-7").expect("messages"); + assert!( + messages[0].extra_metadata.get("citations").is_none(), + "an empty citation list must not add an empty array the client never wrote" + ); +} diff --git a/src/openhuman/web_chat/run_task.rs b/src/openhuman/web_chat/run_task.rs index 03f3b11ceb..75886ead75 100644 --- a/src/openhuman/web_chat/run_task.rs +++ b/src/openhuman/web_chat/run_task.rs @@ -276,6 +276,7 @@ pub(crate) async fn run_chat_task( full_response: response, citations, usage, + workspace_dir: config.workspace_dir.clone(), }) } Err(err) => { @@ -306,6 +307,7 @@ pub(crate) async fn run_chat_task( full_response: inference_budget_exceeded_user_message().to_string(), citations: Vec::new(), usage: None, + workspace_dir: config.workspace_dir.clone(), }) } BudgetCorrelation::UpgradeEmptyToBudget => { @@ -324,6 +326,7 @@ pub(crate) async fn run_chat_task( full_response: inference_budget_exceeded_user_message().to_string(), citations: Vec::new(), usage: None, + workspace_dir: config.workspace_dir.clone(), }) } BudgetCorrelation::PassThrough => Err(err_message), diff --git a/src/openhuman/web_chat/run_task_tests.rs b/src/openhuman/web_chat/run_task_tests.rs index 8d15aee0b4..685767aa6c 100644 --- a/src/openhuman/web_chat/run_task_tests.rs +++ b/src/openhuman/web_chat/run_task_tests.rs @@ -5,6 +5,7 @@ fn ok() -> Result { full_response: "hello".to_string(), citations: Vec::new(), usage: None, + workspace_dir: std::path::PathBuf::from("/tmp/ws"), }) } diff --git a/src/openhuman/web_chat/types.rs b/src/openhuman/web_chat/types.rs index 7301d34b68..77c672ddff 100644 --- a/src/openhuman/web_chat/types.rs +++ b/src/openhuman/web_chat/types.rs @@ -74,6 +74,14 @@ pub(super) struct WebChatTaskResult { /// forwarded to the frontend on `chat_done`. `None` for synthetic results /// (e.g. budget-exhausted placeholders) that never ran a real turn. pub(super) usage: Option, + /// The workspace this turn actually ran in, carried to delivery so the + /// reply is stored there before it is announced (#6034). + /// + /// Taken from the config the turn resolved rather than re-read at delivery + /// time: a sign-out or account switch moves `workspace_dir`, and a reply + /// re-resolved afterwards would be filed under whoever is signed in when + /// the turn happens to finish. + pub(super) workspace_dir: std::path::PathBuf, } /// Per-request metadata carried alongside a chat send. Currently used by the