From 355684233702eadd10a3dd075bb887f4097c9bc2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sat, 30 May 2026 01:04:23 -0700 Subject: [PATCH] feat(agent): live streaming subagent view with reopenable sub-threads (#3007) --- CLAUDE.md | 4 +- app/src/lib/i18n/ar.ts | 11 + app/src/lib/i18n/bn.ts | 11 + app/src/lib/i18n/de.ts | 11 + app/src/lib/i18n/en.ts | 11 + app/src/lib/i18n/es.ts | 11 + app/src/lib/i18n/fr.ts | 11 + app/src/lib/i18n/hi.ts | 11 + app/src/lib/i18n/id.ts | 11 + app/src/lib/i18n/it.ts | 11 + app/src/lib/i18n/ko.ts | 11 + app/src/lib/i18n/pl.ts | 11 + app/src/lib/i18n/pt.ts | 11 + app/src/lib/i18n/ru.ts | 11 + app/src/lib/i18n/zh-CN.ts | 11 + app/src/pages/Conversations.tsx | 25 +- .../components/SubagentDrawer.tsx | 347 ++++++++++++++++++ .../components/ToolTimelineBlock.tsx | 64 +++- .../__tests__/SubagentDrawer.test.tsx | 176 +++++++++ .../__tests__/ToolTimelineBlock.test.tsx | 59 ++- app/src/providers/ChatRuntimeProvider.tsx | 70 +++- .../__tests__/ChatRuntimeProvider.test.tsx | 108 ++++++ .../chatService.subagentDelta.test.ts | 87 +++++ app/src/services/chatService.ts | 71 ++++ .../chatRuntimeSlice.subagentStream.test.ts | 192 ++++++++++ app/src/store/chatRuntimeSlice.ts | 157 ++++++++ app/src/types/turnState.ts | 2 + src/core/socketio.rs | 5 + .../agent/harness/subagent_runner/ops.rs | 70 +++- .../harness/subagent_runner/ops_tests.rs | 143 +++++++- src/openhuman/agent/progress.rs | 33 ++ src/openhuman/agent_orchestration/ops.rs | 1 + src/openhuman/agent_orchestration/tools.rs | 2 + .../agent_orchestration/tools/dispatch.rs | 1 + .../tools/spawn_parallel_agents.rs | 1 + .../tools/spawn_subagent.rs | 22 +- .../tools/spawn_worker_thread.rs | 51 +-- .../tools/worker_thread.rs | 114 ++++++ src/openhuman/channels/providers/web.rs | 50 +++ src/openhuman/skills/run_log.rs | 2 + src/openhuman/threads/turn_state/mirror.rs | 13 + .../threads/turn_state/mirror_tests.rs | 1 + src/openhuman/threads/turn_state/types.rs | 5 + 43 files changed, 1963 insertions(+), 67 deletions(-) create mode 100644 app/src/pages/conversations/components/SubagentDrawer.tsx create mode 100644 app/src/pages/conversations/components/__tests__/SubagentDrawer.test.tsx create mode 100644 app/src/services/__tests__/chatService.subagentDelta.test.ts create mode 100644 app/src/store/__tests__/chatRuntimeSlice.subagentStream.test.ts create mode 100644 src/openhuman/agent_orchestration/tools/worker_thread.rs diff --git a/CLAUDE.md b/CLAUDE.md index 9c000748e..a4a6e507f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -365,8 +365,8 @@ Specify → prove in Rust → prove over RPC → surface in the UI → test. - **Pre-merge** (code changes): Prettier, ESLint, `tsc --noEmit` in `app/`; `cargo fmt` + `cargo check` for changed Rust. - **No dynamic imports** in production `app/src` code — static `import` / `import type` only. No `import()`, `React.lazy(() => import(...))`, `await import(...)`. For heavy optional paths, use a static import and guard the call site with `try/catch` or a runtime check. *Exceptions*: Vitest harness patterns in `*.test.ts` / `__tests__` / `test/setup.ts`; ambient `typeof import('…')` in `.d.ts`; config files (e.g. `tailwind.config.js` JSDoc). - **Dual socket sync**: when changing the realtime protocol, keep `socketService` / MCP transport aligned with core socket behavior (see `gitbooks/developing/architecture.md` dual-socket section). -- **i18n for all UI text**: every user-visible string in `app/src/**` (headings, labels, button text, placeholders, status chips, toasts, error messages, dialog copy) must go through `useT()` from `app/src/lib/i18n/I18nContext`. Hard-coded literals in JSX or `label=`/`placeholder=`/`aria-label=` props are not allowed. Add the key to [`app/src/lib/i18n/en.ts`](app/src/lib/i18n/en.ts) in the same PR — other locales fall back to English. Exceptions: developer-only debug logs, code identifiers, and non-display data (URLs, slugs, technical sentinel values). -- **i18n locale files — update ALL locales**: each locale is a **single flat file** at `app/src/lib/i18n/.ts` (`en.ts` is the source of truth; the chunked `chunks/-N.ts` layout was retired). When adding or changing keys in `en.ts`, you **must also** add the same key to every non-English locale file (use the English value as a placeholder — translators fill in later). CI enforces parity via `pnpm i18n:check`; a missing or extra key in any locale will fail the i18n coverage gate. Locales: `ar`, `bn`, `de`, `es`, `fr`, `hi`, `id`, `it`, `ko`, `pl`, `pt`, `ru`, `zh-CN`. **`pnpm i18n:english:check`** ([`scripts/i18n-find-english.ts`](scripts/i18n-find-english.ts)) is a second gate that catches values still rendering English — including *stale* English (translated from an older en string that since changed), which `i18n:check` cannot see because it only compares against the current en value. It uses script-coverage for non-Latin locales and English-only function words for Latin locales, with a reviewed `INTENTIONAL_ENGLISH` allowlist for brand names / commands / paths / units / cognates. Add genuinely-English literals to that allowlist; never use it to silence an untranslated string. +- **i18n for all UI text**: every user-visible string in `app/src/**` (headings, labels, button text, placeholders, status chips, toasts, error messages, dialog copy) must go through `useT()` from `app/src/lib/i18n/I18nContext`. Hard-coded literals in JSX or `label=`/`placeholder=`/`aria-label=` props are not allowed. Add the key to [`app/src/lib/i18n/en.ts`](app/src/lib/i18n/en.ts) **and a real translation to every other locale file** in the same PR (see the next bullet — English-only fallback is the runtime safety net for *missing* keys, not an excuse to ship untranslated values). Exceptions: developer-only debug logs, code identifiers, and non-display data (URLs, slugs, technical sentinel values). +- **i18n locale files — update ALL locales with REAL translations**: each locale is a **single flat file** at `app/src/lib/i18n/.ts` (`en.ts` is the source of truth; the chunked `chunks/-N.ts` layout was retired). When adding or changing keys in `en.ts`, you **must also** add the same key to every non-English locale file **with an actual, correct translation in that language — NOT an English placeholder**. Do not copy the English string into `ar.ts`/`de.ts`/… and "leave it for translators"; translate it (the short UI strings here are well within reach — translate them, and only fall back to English for genuine brand names / commands / paths / units, which belong in the `INTENTIONAL_ENGLISH` allowlist). CI enforces this two ways: `pnpm i18n:check` fails on a missing or extra key in any locale (parity), and **`pnpm i18n:english:check`** ([`scripts/i18n-find-english.ts`](scripts/i18n-find-english.ts)) fails on any value still rendering English — including *stale* English (translated from an older en string that since changed). The English-detection gate uses script-coverage for non-Latin locales and English-only function words for Latin locales; the `INTENTIONAL_ENGLISH` allowlist is **only** for genuinely-English literals (brand names / commands / paths / units / cognates) — never use it to silence an untranslated string. Locales: `ar`, `bn`, `de`, `es`, `fr`, `hi`, `id`, `it`, `ko`, `pl`, `pt`, `ru`, `zh-CN`. --- diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index 0436a2bc1..365517f4a 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -2310,6 +2310,17 @@ const messages: TranslationMap = { 'conversations.taskKanban.updateFailed': 'ولم يتمكن من استكمال المهمة؛ ولم يتم توفير التغييرات.', 'conversations.toolTimeline.turn': 'دور', 'conversations.toolTimeline.workerThread': 'محادثة عامل', + 'conversations.subagent.viewProcessing': 'عرض المعالجة الكاملة', + 'conversations.subagent.parent': 'الأصل', + 'conversations.subagent.thinking': 'التفكير', + 'conversations.subagent.response': 'الرد', + 'conversations.subagent.toolCalls': 'استدعاءات الأدوات', + 'conversations.subagent.working': 'يعمل…', + 'conversations.subagent.noOutputYet': 'لا يوجد ناتج بعد', + 'conversations.subagent.close': 'إغلاق', + 'conversations.subagent.statusRunning': 'قيد التشغيل', + 'conversations.subagent.statusCompleted': 'اكتمل', + 'conversations.subagent.statusFailed': 'فشل', 'daemon.serviceBlockingGate.body': 'المحتوى', 'daemon.serviceBlockingGate.downloadHint': 'تلميح التنزيل', 'daemon.serviceBlockingGate.downloadLatest': 'تنزيل أحدث إصدار', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index 602a36e9b..bd45e4b69 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -2355,6 +2355,17 @@ const messages: TranslationMap = { 'conversations.taskKanban.updateFailed': 'কাজটি সংরক্ষণ করা যায়নি।', 'conversations.toolTimeline.turn': 'টার্ন', 'conversations.toolTimeline.workerThread': 'ওয়ার্কার থ্রেড', + 'conversations.subagent.viewProcessing': 'সম্পূর্ণ প্রক্রিয়া দেখুন', + 'conversations.subagent.parent': 'মূল', + 'conversations.subagent.thinking': 'ভাবছে', + 'conversations.subagent.response': 'প্রতিক্রিয়া', + 'conversations.subagent.toolCalls': 'টুল কল', + 'conversations.subagent.working': 'কাজ করছে…', + 'conversations.subagent.noOutputYet': 'এখনও কোনো আউটপুট নেই', + 'conversations.subagent.close': 'বন্ধ করুন', + 'conversations.subagent.statusRunning': 'চলছে', + 'conversations.subagent.statusCompleted': 'সম্পন্ন', + 'conversations.subagent.statusFailed': 'ব্যর্থ', 'daemon.serviceBlockingGate.body': 'বডি', 'daemon.serviceBlockingGate.downloadHint': 'ডাউনলোড হিন্ট', 'daemon.serviceBlockingGate.downloadLatest': 'সর্বশেষ সংস্করণ ডাউনলোড করুন', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index a46f98efb..205508098 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -2417,6 +2417,17 @@ const messages: TranslationMap = { 'Aufgabe konnte nicht aktualisiert werden; Änderungen wurden nicht gespeichert.', 'conversations.toolTimeline.turn': 'drehen', 'conversations.toolTimeline.workerThread': 'Worker-Thread', + 'conversations.subagent.viewProcessing': 'Vollständige Verarbeitung anzeigen', + 'conversations.subagent.parent': 'Übergeordnet', + 'conversations.subagent.thinking': 'Denkt nach', + 'conversations.subagent.response': 'Antwort', + 'conversations.subagent.toolCalls': 'Tool-Aufrufe', + 'conversations.subagent.working': 'Arbeitet…', + 'conversations.subagent.noOutputYet': 'Noch keine Ausgabe', + 'conversations.subagent.close': 'Schließen', + 'conversations.subagent.statusRunning': 'läuft', + 'conversations.subagent.statusCompleted': 'abgeschlossen', + 'conversations.subagent.statusFailed': 'fehlgeschlagen', 'daemon.serviceBlockingGate.body': 'Körper', 'daemon.serviceBlockingGate.downloadHint': 'Hinweis herunterladen', 'daemon.serviceBlockingGate.downloadLatest': 'Lade die neueste Version herunter', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index f7b21fd46..d6dd86c03 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -2561,6 +2561,17 @@ const en: TranslationMap = { 'conversations.taskKanban.updateFailed': 'Could not update task; changes were not saved.', 'conversations.toolTimeline.turn': 'turn', 'conversations.toolTimeline.workerThread': 'worker thread', + 'conversations.subagent.viewProcessing': 'View full processing', + 'conversations.subagent.parent': 'Parent', + 'conversations.subagent.thinking': 'Thinking', + 'conversations.subagent.response': 'Response', + 'conversations.subagent.toolCalls': 'Tool calls', + 'conversations.subagent.working': 'Working…', + 'conversations.subagent.noOutputYet': 'No output yet', + 'conversations.subagent.close': 'Close', + 'conversations.subagent.statusRunning': 'running', + 'conversations.subagent.statusCompleted': 'completed', + 'conversations.subagent.statusFailed': 'failed', 'daemon.serviceBlockingGate.body': 'Retrying in the background. This usually resolves in a few seconds.', 'daemon.serviceBlockingGate.downloadHint': diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index 7909b9a61..d6cd23f66 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -2399,6 +2399,17 @@ const messages: TranslationMap = { 'No se pudo actualizar la tarea; los cambios no se guardaron.', 'conversations.toolTimeline.turn': 'turno', 'conversations.toolTimeline.workerThread': 'hilo de worker', + 'conversations.subagent.viewProcessing': 'Ver procesamiento completo', + 'conversations.subagent.parent': 'Principal', + 'conversations.subagent.thinking': 'Pensando', + 'conversations.subagent.response': 'Respuesta', + 'conversations.subagent.toolCalls': 'Llamadas a herramientas', + 'conversations.subagent.working': 'Trabajando…', + 'conversations.subagent.noOutputYet': 'Aún no hay resultados', + 'conversations.subagent.close': 'Cerrar', + 'conversations.subagent.statusRunning': 'en ejecución', + 'conversations.subagent.statusCompleted': 'completado', + 'conversations.subagent.statusFailed': 'fallido', 'daemon.serviceBlockingGate.body': 'Cuerpo', 'daemon.serviceBlockingGate.downloadHint': 'Sugerencia de descarga', 'daemon.serviceBlockingGate.downloadLatest': 'Descargar la última versión', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index ae855fd8d..e8c53c050 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -2409,6 +2409,17 @@ const messages: TranslationMap = { "Impossible de mettre à jour la tâche; les modifications n'ont pas été enregistrées.", 'conversations.toolTimeline.turn': 'tour', 'conversations.toolTimeline.workerThread': 'fil worker', + 'conversations.subagent.viewProcessing': 'Voir le traitement complet', + 'conversations.subagent.parent': 'Parent', + 'conversations.subagent.thinking': 'Réflexion', + 'conversations.subagent.response': 'Réponse', + 'conversations.subagent.toolCalls': 'Appels d’outils', + 'conversations.subagent.working': 'En cours…', + 'conversations.subagent.noOutputYet': 'Aucun résultat pour l’instant', + 'conversations.subagent.close': 'Fermer', + 'conversations.subagent.statusRunning': 'en cours', + 'conversations.subagent.statusCompleted': 'terminé', + 'conversations.subagent.statusFailed': 'échoué', 'daemon.serviceBlockingGate.body': 'Corps', 'daemon.serviceBlockingGate.downloadHint': 'Indice de téléchargement', 'daemon.serviceBlockingGate.downloadLatest': 'Télécharger la dernière version', diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index b5c736476..8118b4c7e 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -2359,6 +2359,17 @@ const messages: TranslationMap = { 'conversations.taskKanban.updateFailed': 'कार्य अद्यतन नहीं कर सका; परिवर्तन बचाया नहीं गया।', 'conversations.toolTimeline.turn': 'टर्न', 'conversations.toolTimeline.workerThread': 'वर्कर थ्रेड', + 'conversations.subagent.viewProcessing': 'पूरी प्रोसेसिंग देखें', + 'conversations.subagent.parent': 'मूल', + 'conversations.subagent.thinking': 'सोच रहा है', + 'conversations.subagent.response': 'प्रतिक्रिया', + 'conversations.subagent.toolCalls': 'टूल कॉल', + 'conversations.subagent.working': 'काम कर रहा है…', + 'conversations.subagent.noOutputYet': 'अभी तक कोई आउटपुट नहीं', + 'conversations.subagent.close': 'बंद करें', + 'conversations.subagent.statusRunning': 'चल रहा है', + 'conversations.subagent.statusCompleted': 'पूर्ण', + 'conversations.subagent.statusFailed': 'विफल', 'daemon.serviceBlockingGate.body': 'विवरण', 'daemon.serviceBlockingGate.downloadHint': 'डाउनलोड संकेत', 'daemon.serviceBlockingGate.downloadLatest': 'नवीनतम संस्करण डाउनलोड करें', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index 58746dc20..f8d7f68e1 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -2362,6 +2362,17 @@ const messages: TranslationMap = { 'conversations.taskKanban.updateFailed': 'Tak bisa memutakhirkan tugas; perubahan tak disimpan.', 'conversations.toolTimeline.turn': 'giliran', 'conversations.toolTimeline.workerThread': 'thread worker', + 'conversations.subagent.viewProcessing': 'Lihat proses lengkap', + 'conversations.subagent.parent': 'Induk', + 'conversations.subagent.thinking': 'Berpikir', + 'conversations.subagent.response': 'Respons', + 'conversations.subagent.toolCalls': 'Panggilan alat', + 'conversations.subagent.working': 'Bekerja…', + 'conversations.subagent.noOutputYet': 'Belum ada keluaran', + 'conversations.subagent.close': 'Tutup', + 'conversations.subagent.statusRunning': 'berjalan', + 'conversations.subagent.statusCompleted': 'selesai', + 'conversations.subagent.statusFailed': 'gagal', 'daemon.serviceBlockingGate.body': 'Isi', 'daemon.serviceBlockingGate.downloadHint': 'Petunjuk unduhan', 'daemon.serviceBlockingGate.downloadLatest': 'Unduh Versi Terbaru', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index 5c4830401..d81b9a8c0 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -2392,6 +2392,17 @@ const messages: TranslationMap = { "Impossibile aggiornare l'attività; le modifiche non sono state salvate.", 'conversations.toolTimeline.turn': 'turno', 'conversations.toolTimeline.workerThread': 'thread worker', + 'conversations.subagent.viewProcessing': 'Visualizza elaborazione completa', + 'conversations.subagent.parent': 'Principale', + 'conversations.subagent.thinking': 'Ragionamento', + 'conversations.subagent.response': 'Risposta', + 'conversations.subagent.toolCalls': 'Chiamate agli strumenti', + 'conversations.subagent.working': 'In corso…', + 'conversations.subagent.noOutputYet': 'Ancora nessun output', + 'conversations.subagent.close': 'Chiudi', + 'conversations.subagent.statusRunning': 'in esecuzione', + 'conversations.subagent.statusCompleted': 'completato', + 'conversations.subagent.statusFailed': 'non riuscito', 'daemon.serviceBlockingGate.body': 'Corpo', 'daemon.serviceBlockingGate.downloadHint': 'Suggerimento di download', 'daemon.serviceBlockingGate.downloadLatest': "Scarica l'ultima versione", diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index 45ffd9cba..6a518c1e4 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -2337,6 +2337,17 @@ const messages: TranslationMap = { '작업을 업데이트할 수 없어 변경 사항이 저장되지 않았습니다.', 'conversations.toolTimeline.turn': '턴', 'conversations.toolTimeline.workerThread': '워커 스레드', + 'conversations.subagent.viewProcessing': '전체 처리 과정 보기', + 'conversations.subagent.parent': '상위', + 'conversations.subagent.thinking': '생각 중', + 'conversations.subagent.response': '응답', + 'conversations.subagent.toolCalls': '도구 호출', + 'conversations.subagent.working': '작업 중…', + 'conversations.subagent.noOutputYet': '아직 출력이 없습니다', + 'conversations.subagent.close': '닫기', + 'conversations.subagent.statusRunning': '실행 중', + 'conversations.subagent.statusCompleted': '완료됨', + 'conversations.subagent.statusFailed': '실패', 'daemon.serviceBlockingGate.body': '본문', 'daemon.serviceBlockingGate.downloadHint': '다운로드 안내', 'daemon.serviceBlockingGate.downloadLatest': '최신 버전 다운로드', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index d091ca74a..49391b129 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -2387,6 +2387,17 @@ const messages: TranslationMap = { 'Nie udało się zaktualizować zadania; zmian nie zapisano.', 'conversations.toolTimeline.turn': 'tura', 'conversations.toolTimeline.workerThread': 'wątek workera', + 'conversations.subagent.viewProcessing': 'Zobacz pełne przetwarzanie', + 'conversations.subagent.parent': 'Nadrzędny', + 'conversations.subagent.thinking': 'Myślenie', + 'conversations.subagent.response': 'Odpowiedź', + 'conversations.subagent.toolCalls': 'Wywołania narzędzi', + 'conversations.subagent.working': 'Pracuje…', + 'conversations.subagent.noOutputYet': 'Brak wyników', + 'conversations.subagent.close': 'Zamknij', + 'conversations.subagent.statusRunning': 'w toku', + 'conversations.subagent.statusCompleted': 'zakończono', + 'conversations.subagent.statusFailed': 'niepowodzenie', 'daemon.serviceBlockingGate.body': 'Rdzeń OpenHuman nie odpowiada. Spróbuj ponownie lub pobierz najnowszą wersję aplikacji.', 'daemon.serviceBlockingGate.downloadHint': diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index 2b3d14310..2aad0a3cd 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -2399,6 +2399,17 @@ const messages: TranslationMap = { 'Não foi possível atualizar a tarefa; as alterações não foram salvas.', 'conversations.toolTimeline.turn': 'turno', 'conversations.toolTimeline.workerThread': 'thread de worker', + 'conversations.subagent.viewProcessing': 'Ver processamento completo', + 'conversations.subagent.parent': 'Principal', + 'conversations.subagent.thinking': 'Pensando', + 'conversations.subagent.response': 'Resposta', + 'conversations.subagent.toolCalls': 'Chamadas de ferramentas', + 'conversations.subagent.working': 'Trabalhando…', + 'conversations.subagent.noOutputYet': 'Ainda sem resultado', + 'conversations.subagent.close': 'Fechar', + 'conversations.subagent.statusRunning': 'em execução', + 'conversations.subagent.statusCompleted': 'concluído', + 'conversations.subagent.statusFailed': 'falhou', 'daemon.serviceBlockingGate.body': 'Corpo', 'daemon.serviceBlockingGate.downloadHint': 'Dica de download', 'daemon.serviceBlockingGate.downloadLatest': 'Baixar Versão Mais Recente', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index 36995e080..f12745307 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -2373,6 +2373,17 @@ const messages: TranslationMap = { 'conversations.taskKanban.updateFailed': 'Не удалось обновить задачу; изменения не сохранились.', 'conversations.toolTimeline.turn': 'ход', 'conversations.toolTimeline.workerThread': 'чат воркера', + 'conversations.subagent.viewProcessing': 'Посмотреть весь процесс', + 'conversations.subagent.parent': 'Родительский', + 'conversations.subagent.thinking': 'Думает', + 'conversations.subagent.response': 'Ответ', + 'conversations.subagent.toolCalls': 'Вызовы инструментов', + 'conversations.subagent.working': 'Выполняется…', + 'conversations.subagent.noOutputYet': 'Пока нет результата', + 'conversations.subagent.close': 'Закрыть', + 'conversations.subagent.statusRunning': 'выполняется', + 'conversations.subagent.statusCompleted': 'завершено', + 'conversations.subagent.statusFailed': 'ошибка', 'daemon.serviceBlockingGate.body': 'Текст', 'daemon.serviceBlockingGate.downloadHint': 'Подсказка по загрузке', 'daemon.serviceBlockingGate.downloadLatest': 'Скачать последнюю версию', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index 7af461ed2..5586cabd6 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -2241,6 +2241,17 @@ const messages: TranslationMap = { 'conversations.taskKanban.updateFailed': '无法更新任务；更改未保存。', 'conversations.toolTimeline.turn': '轮次', 'conversations.toolTimeline.workerThread': '工作线程', + 'conversations.subagent.viewProcessing': '查看完整处理过程', + 'conversations.subagent.parent': '父级', + 'conversations.subagent.thinking': '思考中', + 'conversations.subagent.response': '回复', + 'conversations.subagent.toolCalls': '工具调用', + 'conversations.subagent.working': '处理中…', + 'conversations.subagent.noOutputYet': '暂无输出', + 'conversations.subagent.close': '关闭', + 'conversations.subagent.statusRunning': '运行中', + 'conversations.subagent.statusCompleted': '已完成', + 'conversations.subagent.statusFailed': '失败', 'daemon.serviceBlockingGate.body': '核心服务不可用，请等待或下载最新版本。', 'daemon.serviceBlockingGate.downloadHint': '下载最新版本', 'daemon.serviceBlockingGate.downloadLatest': '下载最新版本', diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx index 9a687559a..bf9e195f4 100644 --- a/app/src/pages/Conversations.tsx +++ b/app/src/pages/Conversations.tsx @@ -78,6 +78,7 @@ import { formatTimelineEntry } from '../utils/toolTimelineFormatting'; import { AgentMessageBubble, BubbleMarkdown } from './conversations/components/AgentMessageBubble'; import { CitationChips, type MessageCitation } from './conversations/components/CitationChips'; import { LimitPill } from './conversations/components/LimitPill'; +import { SubagentDrawer } from './conversations/components/SubagentDrawer'; import { TaskKanbanBoard } from './conversations/components/TaskKanbanBoard'; import { ToolTimelineBlock } from './conversations/components/ToolTimelineBlock'; import { @@ -201,6 +202,9 @@ const Conversations = ({ const [attachments, setAttachments] = useState([]); const fileInputRef = useRef(null); const [copiedMessageId, setCopiedMessageId] = useState(null); + // Sub-agent whose full live transcript is open in the drawer, keyed by the + // owning timeline row's spawn `taskId`. Null when the drawer is closed. + const [openSubagentTaskId, setOpenSubagentTaskId] = useState(null); const [inputMode, setInputMode] = useState('text'); const [replyMode, setReplyMode] = useState('text'); const [isRecording, setIsRecording] = useState(false); @@ -1069,6 +1073,12 @@ const Conversations = ({ const selectedThreadToolTimeline = selectedThreadId ? (toolTimelineByThread[selectedThreadId] ?? []) : []; + // Re-derive the open subagent's live activity (and its row status) from the + // timeline on every render so the drawer streams token-by-token as + // subagent_text_delta / subagent_thinking_delta events land in Redux. + const openSubagentEntry = openSubagentTaskId + ? selectedThreadToolTimeline.find(entry => entry.subagent?.taskId === openSubagentTaskId) + : undefined; const selectedTaskBoard = selectedThreadId ? (taskBoardByThread[selectedThreadId] ?? null) : null; const hasTaskBoard = Boolean(selectedTaskBoard?.cards.length); const visibleMessages = messages.filter(msg => !msg.extraMetadata?.hidden); @@ -1644,7 +1654,10 @@ const Conversations = ({
{shouldRenderTimelineBeforeLatestAgentMessage && latestVisibleAgentMessage?.id === msg.id && ( - + setOpenSubagentTaskId(sub.taskId)} + /> )}
@@ -1929,7 +1942,10 @@ const Conversations = ({ {/* Tool call timeline */} {selectedThreadToolTimeline.length > 0 && !shouldRenderTimelineBeforeLatestAgentMessage && ( - + setOpenSubagentTaskId(sub.taskId)} + /> )} {isSending && rustChat && (
@@ -2311,6 +2327,11 @@ const Conversations = ({ modal={deleteModal} onClose={() => setDeleteModal(prev => ({ ...prev, isOpen: false }))} /> + setOpenSubagentTaskId(null)} + />
); }; diff --git a/app/src/pages/conversations/components/SubagentDrawer.tsx b/app/src/pages/conversations/components/SubagentDrawer.tsx new file mode 100644 index 000000000..59ec3eafc --- /dev/null +++ b/app/src/pages/conversations/components/SubagentDrawer.tsx @@ -0,0 +1,347 @@ +import { type ReactNode, useEffect, useState } from 'react'; + +import { useT } from '../../../lib/i18n/I18nContext'; +import { threadApi } from '../../../services/api/threadApi'; +import type { + SubagentActivity, + SubagentTranscriptItem, + ToolTimelineEntryStatus, +} from '../../../store/chatRuntimeSlice'; +import type { ThreadMessage } from '../../../types/thread'; +import { BubbleMarkdown } from './AgentMessageBubble'; + +/** + * Rebuild a renderable transcript from a worker sub-thread's persisted + * messages so a delegation can be reopened from memory after its live + * stream is gone (navigation / cold boot). The first `user` message is the + * parent's delegation prompt; `agent` messages with a `tool_name` in their + * metadata are tool calls, the rest are the sub-agent's visible text. + * Streamed reasoning isn't persisted, so reopened transcripts omit it. + */ +function transcriptFromMessages(messages: ThreadMessage[]): { + prompt?: string; + items: SubagentTranscriptItem[]; +} { + let prompt: string | undefined; + const items: SubagentTranscriptItem[] = []; + for (const m of messages) { + const meta = m.extraMetadata ?? {}; + const iteration = typeof meta.iteration === 'number' ? meta.iteration : undefined; + if (m.sender === 'user') { + if (prompt === undefined) prompt = m.content; + continue; + } + const toolName = typeof meta.tool_name === 'string' ? meta.tool_name : undefined; + if (toolName) { + items.push({ kind: 'tool', iteration, callId: m.id, toolName, status: 'success' }); + } else if (m.content.trim().length > 0) { + items.push({ kind: 'text', iteration, text: m.content }); + } + } + return { prompt, items }; +} + +/** + * Map a subagent row's terminal/running status to the visual tone used + * across the drawer (header dot, status pill). Mirrors the colour + * language of `ToolTimelineBlock` so the inline card and the drawer read + * as the same surface. + */ +function statusTone(status: ToolTimelineEntryStatus | undefined): { + dot: string; + pill: string; + label: 'statusRunning' | 'statusCompleted' | 'statusFailed'; +} { + if (status === 'success') { + return { + dot: 'bg-sage-500', + pill: 'bg-sage-100 dark:bg-sage-500/20 text-sage-700 dark:text-sage-300', + label: 'statusCompleted', + }; + } + if (status === 'error') { + return { + dot: 'bg-coral-500', + pill: 'bg-coral-100 dark:bg-coral-500/20 text-coral-700 dark:text-coral-300', + label: 'statusFailed', + }; + } + return { + dot: 'bg-amber-500 animate-pulse', + pill: 'bg-amber-100 dark:bg-amber-500/20 text-amber-700 dark:text-amber-300', + label: 'statusRunning', + }; +} + +function formatElapsed(ms: number): string { + return ms >= 1000 ? `${(ms / 1000).toFixed(1)}s` : `${ms}ms`; +} + +/** + * Full live-transcript view for one sub-agent, slid in from the right. + * + * Driven entirely off the live [`SubagentActivity`] the caller passes — + * because the caller re-derives that object from Redux on every render, + * the drawer updates token-by-token as `subagent_text_delta` / + * `subagent_thinking_delta` events stream in. Shows the streamed + * reasoning (collapsible), the streamed visible output (rendered as + * Markdown), and the chronological list of child tool calls with their + * status and timings. + * + * Rendered as `null` when no subagent is selected, so the parent can + * mount it unconditionally and just flip `subagent`. + */ +export function SubagentDrawer({ + subagent, + status, + onClose, +}: { + subagent: SubagentActivity | null; + /** Lifecycle status of the owning timeline row (running/success/error). */ + status?: ToolTimelineEntryStatus; + onClose: () => void; +}) { + const { t } = useT(); + + // Close on Escape for keyboard parity with the backdrop click. + useEffect(() => { + if (!subagent) return; + const onKey = (e: KeyboardEvent) => { + if (e.key === 'Escape') onClose(); + }; + window.addEventListener('keydown', onKey); + return () => window.removeEventListener('keydown', onKey); + }, [subagent, onClose]); + + // Reopen-from-memory: when there's no live transcript (the row was + // restored from a snapshot, or the user navigated back after the turn + // ended) but a worker sub-thread backs it, load that thread's persisted + // messages and render them as the conversation. Failures fall back to the + // empty/working placeholder rather than blocking the drawer. + // Tagged with the worker thread it was fetched for, so a pending request + // for a previous thread can't paint the wrong conversation after the user + // switches subagents. + const [fetched, setFetched] = useState<{ + workerThreadId: string; + prompt?: string; + items: SubagentTranscriptItem[]; + } | null>(null); + const liveTranscript = subagent?.transcript ?? []; + const workerThreadId = subagent?.workerThreadId; + const needsFetch = Boolean(subagent && workerThreadId && liveTranscript.length === 0); + + useEffect(() => { + if (!needsFetch || !workerThreadId) { + setFetched(null); + return; + } + // Clear any prior thread's transcript up front so it can't linger while + // the new request is in flight. + setFetched(null); + let cancelled = false; + void threadApi + .getThreadMessages(workerThreadId) + .then(data => { + if (!cancelled) setFetched({ workerThreadId, ...transcriptFromMessages(data.messages) }); + }) + .catch(() => { + if (!cancelled) setFetched(null); + }); + return () => { + cancelled = true; + }; + }, [needsFetch, workerThreadId]); + + if (!subagent) return null; + + const tone = statusTone(status); + const isRunning = status !== 'success' && status !== 'error'; + // Only trust the fetched transcript when it belongs to the current worker. + const fetchedForCurrent = + fetched && workerThreadId && fetched.workerThreadId === workerThreadId ? fetched : null; + const transcript = liveTranscript.length > 0 ? liveTranscript : (fetchedForCurrent?.items ?? []); + const promptText = subagent.prompt ?? fetchedForCurrent?.prompt; + // The last visible-text item gets the live cursor while the run is in + // flight (the model is mid-sentence on its final/visible output). + let lastTextIdx = -1; + for (let i = transcript.length - 1; i >= 0; i -= 1) { + if (transcript[i].kind === 'text') { + lastTextIdx = i; + break; + } + } + + return ( +
+ {/* Backdrop */} + + + + {/* Body — a parent↔subagent conversation: the parent's delegation + prompt opens it, then the sub-agent replies as one chronological + transcript (thinking, the text it produced, the tool calls that + text triggered, the next turn — exactly as it was emitted). */} +
+ {/* Parent → sub-agent: the delegation prompt (the "input"). */} + {promptText ? ( +
+
+
+ {t('conversations.subagent.parent')} +
+
{promptText}
+
+
+ ) : null} + + {/* Sub-agent side: avatar label + its turns. */} +
+ 🤖 + {subagent.agentId} +
+ + {transcript.length === 0 ? ( +

+ {isRunning + ? t('conversations.subagent.working') + : t('conversations.subagent.noOutputYet')} +

+ ) : ( +
    + {transcript.map((item, idx) => { + // Insert a "Turn N" divider when the iteration advances. + const prevIteration = idx > 0 ? transcript[idx - 1].iteration : undefined; + const showTurn = item.iteration != null && item.iteration !== prevIteration; + const turnDivider = showTurn ? ( +
  1. + + {t('conversations.toolTimeline.turn')} {item.iteration} + +
    2. + ) : null; + + if (item.kind === 'thinking') { + return ( + +
    +
    + + {t('conversations.subagent.thinking')} +
    + 
    +                          {item.text}
+
    +
    +     + ); + } + + if (item.kind === 'text') { + return ( + +
    + + {isRunning && idx === lastTextIdx ? ( + + ) : null} +
    +     + ); + } + + const callTone = + item.status === 'running' + ? 'text-amber-700 dark:text-amber-300' + : item.status === 'success' + ? 'text-sage-700 dark:text-sage-300' + : 'text-coral-700 dark:text-coral-300'; + const statusLabel = + item.status === 'running' + ? t('conversations.subagent.statusRunning') + : item.status === 'success' + ? t('conversations.subagent.statusCompleted') + : t('conversations.subagent.statusFailed'); + return ( + +
    + 🔧 + + {item.toolName} + + {statusLabel} + {item.elapsedMs != null && item.status !== 'running' ? ( + + {formatElapsed(item.elapsedMs)} + + ) : null} +
    +     + ); + })} +
+ )} +
+ +
+ ); +} + +/** Render a transcript row, prefixed by an optional "Turn N" divider. */ +function ItemWrapper({ divider, children }: { divider: ReactNode; children: ReactNode }) { + return ( + <> + {divider} +
  • {children}
    • + + ); +} diff --git a/app/src/pages/conversations/components/ToolTimelineBlock.tsx b/app/src/pages/conversations/components/ToolTimelineBlock.tsx index b2b13f3f9..298aee459 100644 --- a/app/src/pages/conversations/components/ToolTimelineBlock.tsx +++ b/app/src/pages/conversations/components/ToolTimelineBlock.tsx @@ -35,7 +35,18 @@ function workerStatusFromEntry( * present on the entry, which is true for any row produced by the * `subagent_*` socket events from a current core. */ -export function SubagentActivityBlock({ subagent }: { subagent: SubagentActivity }) { +/** Chars of streamed subagent text/thinking shown in the inline card tail. */ +const SUBAGENT_PREVIEW_CHARS = 140; + +export function SubagentActivityBlock({ + subagent, + onView, +}: { + subagent: SubagentActivity; + /** Opens the full-transcript drawer for this subagent. Omitted in + * read-only contexts (e.g. a completed snapshot with no live driver). */ + onView?: () => void; +}) { const { t } = useT(); const headerBits: string[] = []; if (subagent.mode) headerBits.push(subagent.mode); @@ -58,6 +69,22 @@ export function SubagentActivityBlock({ subagent }: { subagent: SubagentActivity : `${subagent.elapsedMs}ms` ); } + + // Live one-line preview of the subagent's streamed processing, derived + // from the ordered transcript: prefer the latest visible-output tail, then + // fall back to the latest reasoning tail while the child is still thinking + // and hasn't emitted visible text yet. Drives the at-a-glance "what is it + // doing right now" affordance on the card. + const transcript = subagent.transcript ?? []; + const lastTextItem = [...transcript].reverse().find(i => i.kind === 'text'); + const lastThinkingItem = [...transcript].reverse().find(i => i.kind === 'thinking'); + const previewItem = lastTextItem ?? lastThinkingItem; + const previewIcon = previewItem?.kind === 'text' ? '📝' : '💭'; + const preview = + previewItem && 'text' in previewItem + ? previewItem.text.replace(/\s+/g, ' ').trim().slice(-SUBAGENT_PREVIEW_CHARS) + : ''; + return (
    ) : null} + {preview ? ( +
    + {previewIcon} + {preview} +
    + ) : null} + {onView ? ( + + ) : null}
    ); } -export function ToolTimelineBlock({ entries }: { entries: ToolTimelineEntry[] }) { +export function ToolTimelineBlock({ + entries, + onViewSubagent, +}: { + entries: ToolTimelineEntry[]; + /** Opens the full-transcript drawer for a subagent row. When omitted, + * subagent cards render without the "view full processing" affordance + * (e.g. interrupted-snapshot rendering with no live driver). */ + onViewSubagent?: (subagent: SubagentActivity) => void; +}) { const latestRunningEntryId = [...entries].reverse().find(entry => entry.status === 'running')?.id; const normalizeToolBody = (value?: string): string | undefined => { @@ -199,7 +252,12 @@ export function ToolTimelineBlock({ entries }: { entries: ToolTimelineEntry[] }) {detailContent} ) : null} - {subagent ? : null} + {subagent ? ( + onViewSubagent(subagent) : undefined} + /> + ) : null} ) : (
    diff --git a/app/src/pages/conversations/components/__tests__/SubagentDrawer.test.tsx b/app/src/pages/conversations/components/__tests__/SubagentDrawer.test.tsx new file mode 100644 index 000000000..11f48108d --- /dev/null +++ b/app/src/pages/conversations/components/__tests__/SubagentDrawer.test.tsx @@ -0,0 +1,176 @@ +import { render, screen, waitFor } from '@testing-library/react'; +import userEvent from '@testing-library/user-event'; +import { describe, expect, it, vi } from 'vitest'; + +import { threadApi } from '../../../../services/api/threadApi'; +import type { SubagentActivity, SubagentTranscriptItem } from '../../../../store/chatRuntimeSlice'; +import { SubagentDrawer } from '../SubagentDrawer'; + +vi.mock('../../../../services/api/threadApi', () => ({ + threadApi: { getThreadMessages: vi.fn() }, +})); + +function activity(overrides: Partial = {}): SubagentActivity { + return { taskId: 'sub-1', agentId: 'researcher', toolCalls: [], transcript: [], ...overrides }; +} + +const INTERLEAVED: SubagentTranscriptItem[] = [ + { kind: 'thinking', iteration: 1, text: 'comparing the two sources' }, + { kind: 'text', iteration: 1, text: 'Let me search for that.' }, + { + kind: 'tool', + iteration: 1, + callId: 'c1', + toolName: 'web_search', + status: 'success', + elapsedMs: 1200, + }, + { kind: 'text', iteration: 2, text: 'The answer is **42**.' }, +]; + +describe('SubagentDrawer', () => { + it('renders nothing when no subagent is selected', () => { + const { container } = render( {}} />); + expect(container.firstChild).toBeNull(); + }); + + it('renders the transcript in chronological order (text where it occurred)', () => { + render( + {}} + /> + ); + const drawer = screen.getByTestId('subagent-drawer'); + expect(drawer.textContent).toContain('researcher'); + + // Walk the rendered transcript items and assert their on-screen order: + // thinking → text → tool → text — i.e. the tool sits between the two + // text blocks, not in a separate section. + const thinking = screen.getByTestId('subagent-transcript-thinking'); + const tool = screen.getByTestId('subagent-drawer-tool-call'); + const texts = screen.getAllByTestId('subagent-transcript-text'); + expect(texts).toHaveLength(2); + + const order = (el: Element) => + Array.prototype.indexOf.call(drawer.querySelectorAll('[data-testid]'), el); + expect(order(thinking)).toBeLessThan(order(texts[0])); + expect(order(texts[0])).toBeLessThan(order(tool)); + expect(order(tool)).toBeLessThan(order(texts[1])); + + expect(thinking.textContent).toContain('comparing the two sources'); + expect(tool.textContent).toContain('web_search'); + expect(tool.textContent).toContain('1.2s'); + expect(texts[1].textContent).toContain('The answer is'); + }); + + it('opens with the parent delegation prompt as a chat bubble', () => { + render( + {}} + /> + ); + const parent = screen.getByTestId('subagent-parent-prompt'); + expect(parent.textContent).toContain('Research Q3 revenue drivers'); + // The parent bubble renders before the sub-agent's reply. + const drawer = screen.getByTestId('subagent-drawer'); + const text = screen.getByTestId('subagent-transcript-text'); + const order = (el: Element) => + Array.prototype.indexOf.call(drawer.querySelectorAll('[data-testid]'), el); + expect(order(parent)).toBeLessThan(order(text)); + }); + + it('inserts a turn divider when the iteration advances', () => { + render( + {}} + /> + ); + // Two distinct iterations (1 and 2) → two turn dividers. + expect(screen.getAllByTestId('subagent-turn-divider')).toHaveLength(2); + }); + + it('shows a working placeholder while running with an empty transcript', () => { + render( {}} />); + expect(screen.getByTestId('subagent-drawer').textContent).toContain('Working'); + }); + + it('reopens from memory: fetches the worker thread when there is no live transcript', async () => { + vi.mocked(threadApi.getThreadMessages).mockResolvedValue({ + count: 3, + messages: [ + { + id: 'm0', + content: 'Research Q3 revenue.', + type: 'text', + sender: 'user', + createdAt: 't0', + extraMetadata: { scope: 'worker_thread' }, + }, + { + id: 'm1', + content: 'Searched the web.', + type: 'text', + sender: 'agent', + createdAt: 't1', + extraMetadata: { tool_name: 'web_search', iteration: 1 }, + }, + { + id: 'm2', + content: 'Revenue grew 18%.', + type: 'text', + sender: 'agent', + createdAt: 't2', + extraMetadata: { iteration: 2, final: true }, + }, + ], + }); + + render( + {}} + /> + ); + + await waitFor(() => expect(threadApi.getThreadMessages).toHaveBeenCalledWith('worker-abc')); + // The persisted conversation renders: parent prompt + a tool call + the text. + await waitFor(() => + expect(screen.getByTestId('subagent-parent-prompt').textContent).toContain('Research Q3') + ); + expect(screen.getByTestId('subagent-drawer-tool-call').textContent).toContain('web_search'); + expect(screen.getByTestId('subagent-transcript-text').textContent).toContain( + 'Revenue grew 18%' + ); + }); + + it('does not fetch when a live transcript is present', () => { + render( + {}} + /> + ); + expect(threadApi.getThreadMessages).not.toHaveBeenCalled(); + }); + + it('invokes onClose from the close button', async () => { + const onClose = vi.fn(); + render(); + await userEvent.click(screen.getByText('✕')); + expect(onClose).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx b/app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx index a81b241e7..ccb4892fc 100644 --- a/app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx +++ b/app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx @@ -1,6 +1,7 @@ import { render, screen } from '@testing-library/react'; +import userEvent from '@testing-library/user-event'; import { Provider } from 'react-redux'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { store } from '../../../../store'; import type { ToolTimelineEntry } from '../../../../store/chatRuntimeSlice'; @@ -75,6 +76,62 @@ describe('SubagentActivityBlock', () => { expect(calls[1].textContent).toContain('·t2'); expect(calls[2].textContent).toContain('error'); }); + + it('shows a live preview of streamed visible text (preferred over thinking)', () => { + renderInStore( + + ); + const preview = screen.getByTestId('subagent-preview'); + expect(preview.textContent).toContain('Here is what I found so far'); + // Visible text takes precedence, so the thinking tail is not shown. + expect(preview.textContent).not.toContain('pondering'); + }); + + it('falls back to the thinking tail while only reasoning has streamed', () => { + renderInStore( + + ); + expect(screen.getByTestId('subagent-preview').textContent).toContain( + 'I should search the web first' + ); + }); + + it('renders the view-processing button only when onView is provided', async () => { + const onView = vi.fn(); + const { rerender } = renderInStore( + + ); + expect(screen.queryByTestId('subagent-view-processing')).toBeNull(); + + rerender( + + + + ); + const btn = screen.getByTestId('subagent-view-processing'); + await userEvent.click(btn); + expect(onView).toHaveBeenCalledTimes(1); + }); }); describe('ToolTimelineBlock — subagent rendering', () => { diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 4c84be021..16ea244e3 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -10,6 +10,8 @@ import { type ChatIterationStartEvent, type ChatSegmentEvent, type ChatSubagentDoneEvent, + type ChatSubagentTextDeltaEvent, + type ChatSubagentThinkingDeltaEvent, type ChatTaskBoardUpdatedEvent, type ChatToolCallEvent, type ChatToolResultEvent, @@ -19,12 +21,15 @@ import { } from '../services/chatService'; import { store } from '../store'; import { + appendSubagentStreamDelta, clearInferenceStatusForThread, clearPendingApprovalForThread, clearStreamingAssistantForThread, endInferenceTurn, markInferenceTurnStreaming, recordChatTurnUsage, + recordSubagentTranscriptTool, + resolveSubagentTranscriptTool, setInferenceStatusForThread, setPendingApprovalForThread, setStreamingAssistantForThread, @@ -297,7 +302,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { const findPendingDelegationContext = ( entries: ToolTimelineEntry[], round: number - ): { sourceToolName?: string; prompt?: string } => { + ): { sourceToolName?: string; prompt?: string; spawnEntryId?: string } => { for (let i = entries.length - 1; i >= 0; i -= 1) { const entry = entries[i]; if (entry.status !== 'running' || entry.round !== round) continue; @@ -305,6 +310,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { return { sourceToolName: entry.name, prompt: entry.detail ?? promptFromArgsBuffer(entry.argsBuffer), + spawnEntryId: entry.id, }; } } @@ -456,11 +462,19 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? []; const pendingContext = findPendingDelegationContext(existing, event.round); + // Collapse the parent's `spawn_subagent`/`delegate_*` tool-call row into + // the subagent row so the timeline shows ONE entry per delegation + // instead of "Research" (the tool call) + "Researching" (the child). + // The tool call's prompt is carried onto the subagent as the parent's + // delegation message, which the drawer renders as the opening turn. + const base = pendingContext.spawnEntryId + ? existing.filter(e => e.id !== pendingContext.spawnEntryId) + : existing; dispatch( setToolTimelineForThread({ threadId: event.thread_id, entries: [ - ...existing, + ...base, decorateEntry({ id: `${event.thread_id}:subagent:${event.skill_id}:${event.tool_name}`, name: `subagent:${event.tool_name}`, @@ -471,9 +485,12 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { subagent: { taskId: event.skill_id, agentId: event.tool_name, + workerThreadId: event.subagent?.worker_thread_id, mode: event.subagent?.mode, dedicatedThread: event.subagent?.dedicated_thread, + prompt: pendingContext.prompt, toolCalls: [], + transcript: [], }, }), ], @@ -562,6 +579,17 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { }, }; dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries: next })); + // Mirror the call into the ordered transcript so the drawer renders + // it right after the text that triggered it (chronological view). + dispatch( + recordSubagentTranscriptTool({ + threadId: event.thread_id, + rowId, + callId: event.tool_call_id, + toolName: event.tool_name, + iteration: event.subagent?.child_iteration, + }) + ); }, onSubagentToolResult: event => { const taskId = event.subagent?.task_id ?? event.skill_id; @@ -585,6 +613,44 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { const next = [...existing]; next[idx] = { ...entry, subagent: { ...entry.subagent, toolCalls: updatedCalls } }; dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries: next })); + dispatch( + resolveSubagentTranscriptTool({ + threadId: event.thread_id, + rowId, + callId: event.tool_call_id, + success: event.success, + elapsedMs: event.subagent?.elapsed_ms, + outputChars: event.subagent?.output_chars, + }) + ); + }, + onSubagentTextDelta: (event: ChatSubagentTextDeltaEvent) => { + const taskId = event.subagent?.task_id; + const agentId = event.subagent?.agent_id; + if (!taskId || !agentId || !event.delta) return; + dispatch( + appendSubagentStreamDelta({ + threadId: event.thread_id, + rowId: `${event.thread_id}:subagent:${taskId}:${agentId}`, + kind: 'text', + delta: event.delta, + iteration: event.subagent?.child_iteration, + }) + ); + }, + onSubagentThinkingDelta: (event: ChatSubagentThinkingDeltaEvent) => { + const taskId = event.subagent?.task_id; + const agentId = event.subagent?.agent_id; + if (!taskId || !agentId || !event.delta) return; + dispatch( + appendSubagentStreamDelta({ + threadId: event.thread_id, + rowId: `${event.thread_id}:subagent:${taskId}:${agentId}`, + kind: 'thinking', + delta: event.delta, + iteration: event.subagent?.child_iteration, + }) + ); }, onSegment: (event: ChatSegmentEvent) => { const eventKey = `segment:${event.thread_id}:${event.request_id}:${event.segment_index}`; diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index 101eef339..165cf9b38 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx @@ -125,6 +125,114 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria expect(timeline[0]?.status).toBe('running'); }); + it('collapses a spawn_subagent tool-call row into the subagent row', () => { + const listeners = renderProvider(); + + act(() => { + // Parent invokes the delegation tool — creates a "spawn_subagent" row. + listeners.onToolCall?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + tool_name: 'spawn_subagent', + skill_id: 'orchestration', + args: {}, + tool_call_id: 'call-spawn', + }); + // The delegation prompt streams in as the tool's args JSON. + listeners.onToolArgsDelta?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + tool_call_id: 'call-spawn', + tool_name: 'spawn_subagent', + delta: '{"prompt":"Research Q3 revenue."}', + }); + // The child spawns — should REPLACE the tool-call row, not add a second. + listeners.onSubagentSpawned?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + tool_name: 'researcher', + skill_id: 'sub-1', + message: 'spawned', + subagent: { mode: 'typed' }, + }); + }); + + const timeline = store.getState().chatRuntime.toolTimelineByThread['t1'] ?? []; + expect(timeline).toHaveLength(1); + expect(timeline[0]?.name).toBe('subagent:researcher'); + // The parent's delegation prompt is carried onto the subagent so the + // drawer can open the conversation with it. + expect(timeline[0]?.subagent?.prompt).toContain('Research Q3 revenue'); + }); + + it('appends streamed subagent text & thinking deltas to the subagent transcript', () => { + const listeners = renderProvider(); + + act(() => { + listeners.onSubagentSpawned?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + tool_name: 'researcher', + skill_id: 'sub-1', + message: 'spawned', + subagent: { mode: 'typed' }, + }); + listeners.onSubagentThinkingDelta?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + delta: 'let me think', + subagent: { task_id: 'sub-1', agent_id: 'researcher', child_iteration: 1 }, + }); + listeners.onSubagentTextDelta?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + delta: 'the answer', + subagent: { task_id: 'sub-1', agent_id: 'researcher', child_iteration: 1 }, + }); + }); + + const row = (store.getState().chatRuntime.toolTimelineByThread['t1'] ?? []).find( + e => e.subagent?.taskId === 'sub-1' + ); + expect(row?.subagent?.transcript).toEqual([ + { kind: 'thinking', iteration: 1, text: 'let me think' }, + { kind: 'text', iteration: 1, text: 'the answer' }, + ]); + }); + + it('ignores subagent deltas missing task/agent/delta', () => { + const listeners = renderProvider(); + act(() => { + listeners.onSubagentSpawned?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + tool_name: 'researcher', + skill_id: 'sub-1', + message: 'spawned', + subagent: { mode: 'typed' }, + }); + // No agent_id → dropped. + listeners.onSubagentTextDelta?.({ + thread_id: 't1', + request_id: 'r1', + round: 0, + delta: 'x', + subagent: { task_id: 'sub-1' }, + }); + }); + const row = (store.getState().chatRuntime.toolTimelineByThread['t1'] ?? []).find( + e => e.subagent?.taskId === 'sub-1' + ); + expect(row?.subagent?.transcript).toEqual([]); + }); + it('drops duplicate chat_done events with the same thread/request', async () => { const listeners = renderProvider(); diff --git a/app/src/services/__tests__/chatService.subagentDelta.test.ts b/app/src/services/__tests__/chatService.subagentDelta.test.ts new file mode 100644 index 000000000..c94aca5c5 --- /dev/null +++ b/app/src/services/__tests__/chatService.subagentDelta.test.ts @@ -0,0 +1,87 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + type ChatSubagentTextDeltaEvent, + type ChatSubagentThinkingDeltaEvent, + subscribeChatEvents, +} from '../chatService'; +import { socketService } from '../socketService'; + +/** + * A minimal fake socket.io client capturing `.on` registrations so a test can + * fire a given event by name and assert the wired callback runs. + */ +function fakeSocket() { + const handlers = new Map void>(); + return { + on: vi.fn((event: string, cb: (payload: unknown) => void) => { + handlers.set(event, cb); + }), + off: vi.fn((event: string) => { + handlers.delete(event); + }), + emit: (event: string, payload: unknown) => handlers.get(event)?.(payload), + has: (event: string) => handlers.has(event), + }; +} + +describe('subscribeChatEvents — subagent delta events', () => { + let socket: ReturnType; + + beforeEach(() => { + socket = fakeSocket(); + vi.spyOn(socketService, 'getSocket').mockReturnValue( + socket as unknown as ReturnType + ); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('registers and dispatches subagent_text_delta / subagent_thinking_delta', () => { + const onSubagentTextDelta = vi.fn(); + const onSubagentThinkingDelta = vi.fn(); + + const unsubscribe = subscribeChatEvents({ onSubagentTextDelta, onSubagentThinkingDelta }); + + expect(socket.has('subagent_text_delta')).toBe(true); + expect(socket.has('subagent_thinking_delta')).toBe(true); + + const textEvent: ChatSubagentTextDeltaEvent = { + thread_id: 't1', + request_id: 'r1', + round: 1, + delta: 'hello', + subagent: { task_id: 'sub-1', agent_id: 'researcher', child_iteration: 1 }, + }; + const thinkingEvent: ChatSubagentThinkingDeltaEvent = { + thread_id: 't1', + request_id: 'r1', + round: 1, + delta: 'pondering', + subagent: { task_id: 'sub-1', agent_id: 'researcher', child_iteration: 1 }, + }; + + socket.emit('subagent_text_delta', textEvent); + socket.emit('subagent_thinking_delta', thinkingEvent); + + expect(onSubagentTextDelta).toHaveBeenCalledWith(textEvent); + expect(onSubagentThinkingDelta).toHaveBeenCalledWith(thinkingEvent); + + unsubscribe(); + }); + + it('does not register subagent delta handlers when listeners are omitted', () => { + subscribeChatEvents({}); + expect(socket.has('subagent_text_delta')).toBe(false); + expect(socket.has('subagent_thinking_delta')).toBe(false); + }); + + it('returns a no-op unsubscribe when there is no socket', () => { + vi.spyOn(socketService, 'getSocket').mockReturnValue( + null as unknown as ReturnType + ); + expect(() => subscribeChatEvents({ onSubagentTextDelta: vi.fn() })()).not.toThrow(); + }); +}); diff --git a/app/src/services/chatService.ts b/app/src/services/chatService.ts index d8f27c458..c77979827 100644 --- a/app/src/services/chatService.ts +++ b/app/src/services/chatService.ts @@ -196,6 +196,8 @@ export interface SubagentProgressDetail { elapsed_ms?: number; iterations?: number; output_chars?: number; + /** Persistent worker sub-thread id backing the delegation (on `subagent_spawned`). */ + worker_thread_id?: string; } /** Extended payload for `subagent_spawned`. */ @@ -250,6 +252,37 @@ export interface ChatSubagentToolResultEvent { subagent?: SubagentProgressDetail; } +/** + * Emitted for each chunk of a sub-agent's streamed assistant text while + * the child iteration is in flight. Distinct from `text_delta` (which is + * the parent's own output) so the UI attributes the token to the running + * subagent row via `subagent.task_id` / `subagent.agent_id` and renders + * it in that row's live transcript. Concatenating `delta`s in order + * yields the child's visible text for the iteration. + */ +export interface ChatSubagentTextDeltaEvent { + thread_id: string; + request_id: string; + /** Parent iteration index (inherited from the parent context). */ + round: number; + /** Text fragment from the sub-agent. */ + delta: string; + subagent?: SubagentProgressDetail; +} + +/** + * Emitted for each chunk of a sub-agent's streamed reasoning / thinking + * output. Counterpart to `thinking_delta` scoped to a child run — only + * sent by models that expose `reasoning_content`. + */ +export interface ChatSubagentThinkingDeltaEvent { + thread_id: string; + request_id: string; + round: number; + delta: string; + subagent?: SubagentProgressDetail; +} + /** * Emitted for each chunk of streamed assistant text that arrives from the * provider during an iteration. Concatenating `delta` values in order yields @@ -309,6 +342,8 @@ export interface ChatEventListeners { onSubagentIterationStart?: (event: ChatSubagentIterationStartEvent) => void; onSubagentToolCall?: (event: ChatSubagentToolCallEvent) => void; onSubagentToolResult?: (event: ChatSubagentToolResultEvent) => void; + onSubagentTextDelta?: (event: ChatSubagentTextDeltaEvent) => void; + onSubagentThinkingDelta?: (event: ChatSubagentThinkingDeltaEvent) => void; onSegment?: (event: ChatSegmentEvent) => void; onTextDelta?: (event: ChatTextDeltaEvent) => void; onThinkingDelta?: (event: ChatThinkingDeltaEvent) => void; @@ -339,6 +374,8 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void { subagentIterationStart: 'subagent_iteration_start', subagentToolCall: 'subagent_tool_call', subagentToolResult: 'subagent_tool_result', + subagentTextDelta: 'subagent_text_delta', + subagentThinkingDelta: 'subagent_thinking_delta', segment: 'chat_segment', textDelta: 'text_delta', thinkingDelta: 'thinking_delta', @@ -513,6 +550,40 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void { handlers.push([EVENTS.subagentToolResult, cb]); } + if (listeners.onSubagentTextDelta) { + const cb = (payload: unknown) => { + const e = payload as ChatSubagentTextDeltaEvent; + chatLog( + '%s thread_id=%s task=%s child_round=%s chars=%d', + EVENTS.subagentTextDelta, + e.thread_id, + e.subagent?.task_id, + e.subagent?.child_iteration, + e.delta?.length ?? 0 + ); + listeners.onSubagentTextDelta?.(e); + }; + socket.on(EVENTS.subagentTextDelta, cb); + handlers.push([EVENTS.subagentTextDelta, cb]); + } + + if (listeners.onSubagentThinkingDelta) { + const cb = (payload: unknown) => { + const e = payload as ChatSubagentThinkingDeltaEvent; + chatLog( + '%s thread_id=%s task=%s child_round=%s chars=%d', + EVENTS.subagentThinkingDelta, + e.thread_id, + e.subagent?.task_id, + e.subagent?.child_iteration, + e.delta?.length ?? 0 + ); + listeners.onSubagentThinkingDelta?.(e); + }; + socket.on(EVENTS.subagentThinkingDelta, cb); + handlers.push([EVENTS.subagentThinkingDelta, cb]); + } + if (listeners.onSegment) { const cb = (payload: unknown) => { const e = payload as ChatSegmentEvent; diff --git a/app/src/store/__tests__/chatRuntimeSlice.subagentStream.test.ts b/app/src/store/__tests__/chatRuntimeSlice.subagentStream.test.ts new file mode 100644 index 000000000..236359011 --- /dev/null +++ b/app/src/store/__tests__/chatRuntimeSlice.subagentStream.test.ts @@ -0,0 +1,192 @@ +import { describe, expect, it } from 'vitest'; + +import reducer, { + appendSubagentStreamDelta, + recordSubagentTranscriptTool, + resolveSubagentTranscriptTool, + setToolTimelineForThread, + type SubagentTranscriptItem, + type ToolTimelineEntry, +} from '../chatRuntimeSlice'; + +const THREAD = 'thread-1'; +const ROW_ID = `${THREAD}:subagent:sub-1:researcher`; + +function withSubagentRow(): ReturnType { + const entry: ToolTimelineEntry = { + id: ROW_ID, + name: 'subagent:researcher', + round: 1, + status: 'running', + subagent: { taskId: 'sub-1', agentId: 'researcher', toolCalls: [], transcript: [] }, + }; + return reducer(undefined, setToolTimelineForThread({ threadId: THREAD, entries: [entry] })); +} + +function transcriptOf(state: ReturnType): SubagentTranscriptItem[] { + return state.toolTimelineByThread[THREAD][0].subagent?.transcript ?? []; +} + +describe('subagent transcript reducers', () => { + it('accumulates consecutive same-kind deltas into one transcript item', () => { + let state = withSubagentRow(); + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'text', + delta: 'Hello ', + iteration: 1, + }) + ); + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'text', + delta: 'world', + iteration: 1, + }) + ); + const t = transcriptOf(state); + expect(t).toHaveLength(1); + expect(t[0]).toMatchObject({ kind: 'text', text: 'Hello world', iteration: 1 }); + }); + + it('does not merge same-kind deltas from different iterations', () => { + let state = withSubagentRow(); + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'text', + delta: 'turn one', + iteration: 1, + }) + ); + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'text', + delta: 'turn two', + iteration: 2, + }) + ); + const t = transcriptOf(state); + expect(t).toHaveLength(2); + expect(t.map(i => (i.kind === 'text' ? i.iteration : null))).toEqual([1, 2]); + }); + + it('interleaves thinking, text, and tool calls in arrival order', () => { + let state = withSubagentRow(); + // turn 1: think → text → tool call + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'thinking', + delta: 'I should search', + iteration: 1, + }) + ); + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'text', + delta: 'Let me look that up.', + iteration: 1, + }) + ); + state = reducer( + state, + recordSubagentTranscriptTool({ + threadId: THREAD, + rowId: ROW_ID, + callId: 'c1', + toolName: 'web_search', + iteration: 1, + }) + ); + // turn 2: text only (final answer) + state = reducer( + state, + appendSubagentStreamDelta({ + threadId: THREAD, + rowId: ROW_ID, + kind: 'text', + delta: 'Found it.', + iteration: 2, + }) + ); + + const t = transcriptOf(state); + expect(t.map(i => i.kind)).toEqual(['thinking', 'text', 'tool', 'text']); + // The tool sits between the turn-1 text and the turn-2 text — i.e. where + // it actually occurred, not bucketed into a separate section. + expect(t[2]).toMatchObject({ kind: 'tool', callId: 'c1', status: 'running' }); + expect(t[3]).toMatchObject({ kind: 'text', text: 'Found it.', iteration: 2 }); + }); + + it('resolves a transcript tool item to its terminal status with timing', () => { + let state = withSubagentRow(); + state = reducer( + state, + recordSubagentTranscriptTool({ + threadId: THREAD, + rowId: ROW_ID, + callId: 'c1', + toolName: 'web_search', + iteration: 1, + }) + ); + state = reducer( + state, + resolveSubagentTranscriptTool({ + threadId: THREAD, + rowId: ROW_ID, + callId: 'c1', + success: true, + elapsedMs: 1200, + outputChars: 480, + }) + ); + const tool = transcriptOf(state).find(i => i.kind === 'tool'); + expect(tool).toMatchObject({ status: 'success', elapsedMs: 1200, outputChars: 480 }); + }); + + it('de-dupes a redelivered tool-call event by callId', () => { + let state = withSubagentRow(); + const record = recordSubagentTranscriptTool({ + threadId: THREAD, + rowId: ROW_ID, + callId: 'c1', + toolName: 'web_search', + }); + state = reducer(state, record); + state = reducer(state, record); + expect(transcriptOf(state).filter(i => i.kind === 'tool')).toHaveLength(1); + }); + + it('is a no-op when the thread or row is unknown', () => { + const state = withSubagentRow(); + expect( + reducer( + state, + appendSubagentStreamDelta({ threadId: 'nope', rowId: ROW_ID, kind: 'text', delta: 'x' }) + ) + ).toEqual(state); + const unknownRow = reducer( + state, + appendSubagentStreamDelta({ threadId: THREAD, rowId: 'missing', kind: 'text', delta: 'x' }) + ); + expect(transcriptOf(unknownRow)).toHaveLength(0); + }); +}); diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index 4baa59439..273d99b85 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -38,10 +38,24 @@ export interface SubagentActivity { taskId: string; /** Sub-agent definition id (e.g. `researcher`). */ agentId: string; + /** + * Persistent worker sub-thread id (`worker-`) backing this + * delegation, when one was created. Lets the drawer reopen the full + * parent↔subagent conversation from memory (via `threadApi.getThreadMessages`) + * after the live transcript is gone — navigation, cold boot, etc. + */ + workerThreadId?: string; /** Resolved spawn mode — `"typed"` or `"fork"`. */ mode?: string; /** `true` when the spawn requested a dedicated worker thread. */ dedicatedThread?: boolean; + /** + * The parent's delegation prompt — what the parent agent asked this + * sub-agent to do. Rendered as the opening (parent) turn in the drawer's + * parent↔subagent chat. Captured from the originating `spawn_subagent` / + * `delegate_*` tool call when the row is created. + */ + prompt?: string; /** Sub-agent's current 1-based iteration index (live). */ childIteration?: number; /** Sub-agent's iteration cap. */ @@ -54,8 +68,44 @@ export interface SubagentActivity { outputChars?: number; /** Child tool calls executed inside the sub-agent, in arrival order. */ toolCalls: SubagentToolCallEntry[]; + /** + * Ordered, interleaved record of everything the sub-agent did, in the + * exact sequence it happened: a run of streamed thinking, then streamed + * visible text, then the tool calls that text triggered, then the next + * iteration's thinking/text, and so on. This is what the full-processing + * drawer renders so reasoning, output, and tool calls appear *where they + * occurred* instead of being split into three flat sections. + * + * Built incrementally from the `subagent_text_delta` / + * `subagent_thinking_delta` / `subagent_tool_call` / `subagent_tool_result` + * socket events in arrival order (the core flushes a child's text/thinking + * deltas before its tool-call events within an iteration, so arrival order + * is chronological order). Text is **not** persisted to the turn-state + * snapshot — on rehydration the transcript is rebuilt from the persisted + * `toolCalls` (tool items only), so an interrupted run still shows its + * tool sequence. Absent on legacy/test rows that predate streaming. + */ + transcript?: SubagentTranscriptItem[]; } +/** + * One entry in a sub-agent's ordered {@link SubagentActivity.transcript}. + * A `thinking`/`text` item accumulates streamed deltas; a `tool` item is a + * child tool call whose `status` flips on its result event. + */ +export type SubagentTranscriptItem = + | { kind: 'thinking'; iteration?: number; text: string } + | { kind: 'text'; iteration?: number; text: string } + | { + kind: 'tool'; + iteration?: number; + callId: string; + toolName: string; + status: ToolTimelineEntryStatus; + elapsedMs?: number; + outputChars?: number; + }; + /** One child tool call performed by a running sub-agent. */ export interface SubagentToolCallEntry { /** Provider-assigned tool call id. */ @@ -174,6 +224,7 @@ function subagentActivityFromPersisted(activity: PersistedSubagentActivity): Sub return { taskId: activity.taskId, agentId: activity.agentId, + workerThreadId: activity.workerThreadId, mode: activity.mode, dedicatedThread: activity.dedicatedThread, childIteration: activity.childIteration, @@ -182,6 +233,19 @@ function subagentActivityFromPersisted(activity: PersistedSubagentActivity): Sub elapsedMs: activity.elapsedMs, outputChars: activity.outputChars, toolCalls: activity.toolCalls.map(subagentToolCallFromPersisted), + // Streamed text/thinking is live-only and never persisted, so a + // rehydrated run can't replay the prose. Rebuild the transcript from + // the persisted tool calls (tool items only) so an interrupted run + // still shows its tool sequence in chronological order. + transcript: activity.toolCalls.map(call => ({ + kind: 'tool' as const, + iteration: call.iteration, + callId: call.callId, + toolName: call.toolName, + status: call.status, + elapsedMs: call.elapsedMs, + outputChars: call.outputChars, + })), }; } @@ -230,6 +294,96 @@ const chatRuntimeSlice = createSlice({ clearToolTimelineForThread: (state, action: PayloadAction<{ threadId: string }>) => { delete state.toolTimelineByThread[action.payload.threadId]; }, + /** + * Append a streamed `subagent_text_delta` / `subagent_thinking_delta` + * chunk to the ordered transcript of the matching subagent row. The row + * is located by its synthetic id (`:subagent::`) + * built from the event's subagent detail — the same id the + * `subagent_spawned` handler created. + * + * Consecutive deltas of the same kind extend the trailing transcript + * item; a kind switch (or an intervening tool call) starts a new item. + * That keeps reasoning, output, and tool calls in the exact order they + * occurred. No-ops if the row isn't present yet (a delta racing ahead of + * its spawn event is dropped rather than resurrecting a context-less row). + */ + appendSubagentStreamDelta: ( + state, + action: PayloadAction<{ + threadId: string; + rowId: string; + kind: 'text' | 'thinking'; + delta: string; + iteration?: number; + }> + ) => { + const { threadId, rowId, kind, delta, iteration } = action.payload; + const entry = state.toolTimelineByThread[threadId]?.find(e => e.id === rowId); + if (!entry?.subagent) return; + const transcript = (entry.subagent.transcript ??= []); + const last = transcript[transcript.length - 1]; + // Extend the trailing item only when it's the same kind AND the same + // iteration — otherwise two same-kind chunks from different turns (with + // no tool call between them) would fuse into one transcript entry. + if ( + last && + (last.kind === 'text' || last.kind === 'thinking') && + last.kind === kind && + last.iteration === iteration + ) { + last.text += delta; + } else { + transcript.push({ kind, iteration, text: delta }); + } + }, + /** + * Record the start of a child tool call as a `tool` item at the current + * tail of the subagent transcript — i.e. right after the text that + * triggered it. De-duped by `callId` so a socket redelivery doesn't + * append twice. Complements the flat `toolCalls` list (kept for the + * compact card + persistence). + */ + recordSubagentTranscriptTool: ( + state, + action: PayloadAction<{ + threadId: string; + rowId: string; + callId: string; + toolName: string; + iteration?: number; + }> + ) => { + const { threadId, rowId, callId, toolName, iteration } = action.payload; + const entry = state.toolTimelineByThread[threadId]?.find(e => e.id === rowId); + if (!entry?.subagent) return; + const transcript = (entry.subagent.transcript ??= []); + if (transcript.some(i => i.kind === 'tool' && i.callId === callId)) return; + transcript.push({ kind: 'tool', iteration, callId, toolName, status: 'running' }); + }, + /** + * Flip a transcript `tool` item to its terminal status when the child + * tool result arrives, recording timing/size. No-op if the matching + * item isn't present. + */ + resolveSubagentTranscriptTool: ( + state, + action: PayloadAction<{ + threadId: string; + rowId: string; + callId: string; + success: boolean; + elapsedMs?: number; + outputChars?: number; + }> + ) => { + const { threadId, rowId, callId, success, elapsedMs, outputChars } = action.payload; + const entry = state.toolTimelineByThread[threadId]?.find(e => e.id === rowId); + const item = entry?.subagent?.transcript?.find(i => i.kind === 'tool' && i.callId === callId); + if (!item || item.kind !== 'tool') return; + item.status = success ? 'success' : 'error'; + if (elapsedMs != null) item.elapsedMs = elapsedMs; + if (outputChars != null) item.outputChars = outputChars; + }, setTaskBoardForThread: ( state, action: PayloadAction<{ threadId: string; board: TaskBoard }> @@ -356,6 +510,9 @@ export const { clearStreamingAssistantForThread, setToolTimelineForThread, clearToolTimelineForThread, + appendSubagentStreamDelta, + recordSubagentTranscriptTool, + resolveSubagentTranscriptTool, setTaskBoardForThread, clearTaskBoardForThread, setPendingApprovalForThread, diff --git a/app/src/types/turnState.ts b/app/src/types/turnState.ts index fbe005dcb..430a91f41 100644 --- a/app/src/types/turnState.ts +++ b/app/src/types/turnState.ts @@ -67,6 +67,8 @@ export interface PersistedSubagentActivity { iterations?: number; elapsedMs?: number; outputChars?: number; + /** Persistent worker sub-thread id backing this delegation (camelCase from core). */ + workerThreadId?: string; toolCalls: PersistedSubagentToolCall[]; } diff --git a/src/core/socketio.rs b/src/core/socketio.rs index a37c115b6..6ee33bc21 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -251,6 +251,11 @@ pub struct SubagentProgressDetail { /// (on `subagent_tool_result`). #[serde(skip_serializing_if = "Option::is_none")] pub output_chars: Option, + /// Persistent worker sub-thread id backing the delegation (on + /// `subagent_spawned`). The frontend stores it on the subagent row and + /// uses it to reopen the full parent↔subagent conversation from memory. + #[serde(skip_serializing_if = "Option::is_none")] + pub worker_thread_id: Option, } #[derive(Debug, Deserialize)] diff --git a/src/openhuman/agent/harness/subagent_runner/ops.rs b/src/openhuman/agent/harness/subagent_runner/ops.rs index 6670d6ab5..166b8dba3 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops.rs @@ -31,7 +31,9 @@ use crate::openhuman::agent::progress::AgentProgress; use crate::openhuman::context::prompt::{ render_subagent_system_prompt, PromptContext, PromptTool, SubagentRenderOptions, }; -use crate::openhuman::inference::provider::{ChatMessage, ChatRequest, Provider, ToolCall}; +use crate::openhuman::inference::provider::{ + ChatMessage, ChatRequest, Provider, ProviderDelta, ToolCall, +}; use crate::openhuman::memory_conversations::ConversationMessage; use crate::openhuman::tools::{Tool, ToolCategory, ToolSpec}; use crate::openhuman::util::truncate_with_ellipsis; @@ -1449,17 +1451,77 @@ async fn run_inner_loop( .await; } - let resp = provider + // Stream the child's tokens to the parent's progress sink so the + // UI can render the sub-agent's thinking/output live, attributed + // to this row via `task_id`. Mirrors the main turn loop + // (`session/turn.rs`): only set up the SSE sink when a listener + // exists, otherwise the channel buffer would back-pressure the + // provider and we'd lose the non-streaming HTTP fast path for + // providers that don't implement streaming. + let child_iteration_for_stream = (iteration + 1) as u32; + let (delta_tx_opt, delta_forwarder) = if let Some(ref sink) = progress_sink { + let (tx, mut rx) = tokio::sync::mpsc::channel::(128); + let sink = sink.clone(); + let agent_id_for_stream = agent_id.to_string(); + let task_id_for_stream = task_id.to_string(); + let forwarder = tokio::spawn(async move { + while let Some(event) = rx.recv().await { + // Only visible text and reasoning deltas attribute to + // the subagent transcript; tool-call arg fragments are + // already surfaced via SubagentToolCall* lifecycle + // events, so they're dropped here to avoid double-render. + let mapped = match event { + ProviderDelta::TextDelta { delta } => AgentProgress::SubagentTextDelta { + agent_id: agent_id_for_stream.clone(), + task_id: task_id_for_stream.clone(), + delta, + iteration: child_iteration_for_stream, + }, + ProviderDelta::ThinkingDelta { delta } => { + AgentProgress::SubagentThinkingDelta { + agent_id: agent_id_for_stream.clone(), + task_id: task_id_for_stream.clone(), + delta, + iteration: child_iteration_for_stream, + } + } + ProviderDelta::ToolCallStart { .. } + | ProviderDelta::ToolCallArgsDelta { .. } => continue, + }; + // Await backpressure so streamed deltas arrive in order + // and aren't silently dropped when the downstream + // progress bridge is slow. + if sink.send(mapped).await.is_err() { + break; + } + } + }); + (Some(tx), Some(forwarder)) + } else { + (None, None) + }; + + let chat_result = provider .chat( ChatRequest { messages: history.as_slice(), tools: request_tools, - stream: None, + stream: delta_tx_opt.as_ref(), }, model, temperature, ) - .await?; + .await; + + // Drop the sender so the forwarder task observes channel close and + // terminates instead of leaking. This must run on BOTH the success + // and error paths — propagating the provider error with `?` before + // joining the forwarder would orphan the task and leak the sender. + drop(delta_tx_opt); + if let Some(forwarder) = delta_forwarder { + let _ = forwarder.await; + } + let resp = chat_result?; if let Some(ref u) = resp.usage { usage.input_tokens += u.input_tokens; diff --git a/src/openhuman/agent/harness/subagent_runner/ops_tests.rs b/src/openhuman/agent/harness/subagent_runner/ops_tests.rs index 41c352680..0a1e3dd17 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops_tests.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops_tests.rs @@ -174,7 +174,7 @@ fn append_subagent_role_contract_is_idempotent() { use crate::openhuman::agent::harness::fork_context::with_parent_context; use crate::openhuman::inference::provider::{ - ChatRequest as PChatRequest, ChatResponse, Provider, ToolCall, + ChatRequest as PChatRequest, ChatResponse, Provider, ProviderDelta, ToolCall, }; use parking_lot::Mutex; use std::sync::Arc; @@ -225,16 +225,45 @@ impl Provider for ScriptedProvider { tool_count: request.tools.map_or(0, |tools| tools.len()), model: model.to_string(), }); - let mut q = self.responses.lock(); - if q.is_empty() { - return Ok(ChatResponse { - text: Some(String::new()), - tool_calls: vec![], - usage: None, - reasoning_content: None, - }); + let response = { + let mut q = self.responses.lock(); + if q.is_empty() { + ChatResponse { + text: Some(String::new()), + tool_calls: vec![], + usage: None, + reasoning_content: None, + } + } else { + q.remove(0) + } + }; + // Mirror a real streaming provider: when the caller attached a + // `stream` sink, forward this response's reasoning then visible + // text as `ProviderDelta`s before returning the aggregate. Lets + // the subagent runner's per-iteration sink exercise the + // `SubagentThinkingDelta` / `SubagentTextDelta` forwarding path. + if let Some(sink) = request.stream { + if let Some(reasoning) = response.reasoning_content.as_deref() { + if !reasoning.is_empty() { + let _ = sink + .send(ProviderDelta::ThinkingDelta { + delta: reasoning.to_string(), + }) + .await; + } + } + if let Some(text) = response.text.as_deref() { + if !text.is_empty() { + let _ = sink + .send(ProviderDelta::TextDelta { + delta: text.to_string(), + }) + .await; + } + } } - Ok(q.remove(0)) + Ok(response) } fn supports_native_tools(&self) -> bool { @@ -251,6 +280,15 @@ fn text_response(text: &str) -> ChatResponse { } } +fn text_response_with_reasoning(text: &str, reasoning: &str) -> ChatResponse { + ChatResponse { + text: Some(text.into()), + tool_calls: vec![], + usage: None, + reasoning_content: Some(reasoning.into()), + } +} + fn tool_response(name: &str, args: &str) -> ChatResponse { ChatResponse { text: Some(String::new()), @@ -859,6 +897,91 @@ async fn typed_mode_emits_child_progress_events_when_sink_attached() { assert_eq!(tool_done[0].2, 1); } +/// A sub-agent's streamed visible text and reasoning are forwarded to the +/// parent's progress sink as `SubagentTextDelta` / `SubagentThinkingDelta` +/// events tagged with the child's `agent_id` / `task_id`, in order, and +/// the concatenated text deltas reconstruct the final assistant text. The +/// web-channel bridge turns these into `subagent_text_delta` / +/// `subagent_thinking_delta` socket events the parent thread renders live. +#[tokio::test] +async fn typed_mode_forwards_child_text_and_thinking_deltas() { + use crate::openhuman::agent::progress::AgentProgress; + + let provider = ScriptedProvider::new(vec![text_response_with_reasoning( + "the final answer", + "let me reason about this", + )]); + let mut parent = make_parent(provider, vec![]); + + let (tx, mut rx) = tokio::sync::mpsc::channel::(64); + parent.on_progress = Some(tx); + + let def = make_def_named_tools(&[]); + let outcome = with_parent_context(parent, async { + run_subagent(&def, "answer me", SubagentRunOptions::default()).await + }) + .await + .expect("runner should succeed"); + assert_eq!(outcome.output, "the final answer"); + + let mut events = Vec::new(); + while let Some(ev) = rx.recv().await { + events.push(ev); + } + + let thinking: Vec<_> = events + .iter() + .filter_map(|e| match e { + AgentProgress::SubagentThinkingDelta { + agent_id, + task_id, + delta, + iteration, + } => Some((agent_id.clone(), task_id.clone(), delta.clone(), *iteration)), + _ => None, + }) + .collect(); + assert_eq!(thinking.len(), 1, "one thinking delta forwarded"); + assert_eq!(thinking[0].2, "let me reason about this"); + assert_eq!(thinking[0].3, 1, "tagged with the child iteration"); + assert!(!thinking[0].1.is_empty(), "carries the child task id"); + + let text: Vec<_> = events + .iter() + .filter_map(|e| match e { + AgentProgress::SubagentTextDelta { + agent_id, + task_id, + delta, + iteration, + } => Some((agent_id.clone(), task_id.clone(), delta.clone(), *iteration)), + _ => None, + }) + .collect(); + assert_eq!(text.len(), 1, "one text delta forwarded"); + assert_eq!(text[0].2, "the final answer"); + assert_eq!(text[0].3, 1); + // Same child identity on both delta kinds so the UI attributes them to + // one subagent row. + assert_eq!(text[0].0, thinking[0].0, "same agent_id"); + assert_eq!(text[0].1, thinking[0].1, "same task_id"); + + // Ordering: the thinking delta precedes the text delta within the + // iteration, matching the provider's emission order. + let thinking_pos = events + .iter() + .position(|e| matches!(e, AgentProgress::SubagentThinkingDelta { .. })) + .unwrap(); + let text_pos = events + .iter() + .position(|e| matches!(e, AgentProgress::SubagentTextDelta { .. })) + .unwrap(); + assert!( + thinking_pos < text_pos, + "thinking streams before visible text" + ); +} + /// Runs without an attached sink must remain backwards compatible — the /// runner is a no-op for child progress and the outcome is unchanged. #[tokio::test] diff --git a/src/openhuman/agent/progress.rs b/src/openhuman/agent/progress.rs index eb76790ab..ac4e2d310 100644 --- a/src/openhuman/agent/progress.rs +++ b/src/openhuman/agent/progress.rs @@ -68,6 +68,11 @@ pub enum AgentProgress { /// whether to render the prompt detail inline or behind a /// "show more" affordance. prompt_chars: usize, + /// Persistent worker sub-thread id backing this delegation, when + /// one was created (`worker-`). The UI uses it to reopen the + /// full parent↔subagent conversation from memory after the live + /// turn ends. `None` for live-only runs (no parent context). + worker_thread_id: Option, }, /// A sub-agent completed successfully. @@ -132,6 +137,34 @@ pub enum AgentProgress { iteration: u32, }, + /// A chunk of a sub-agent's visible assistant text arrived from the + /// provider while the child iteration is still in flight. Distinct + /// from [`Self::TextDelta`] so the parent thread can attribute the + /// streamed token to a specific live subagent row (via `task_id`) + /// and render it inside that row's transcript instead of merging it + /// into the parent's own streaming buffer. Emitted **only from + /// inside [`crate::openhuman::agent::harness::subagent_runner`]** when + /// the parent context carries an `on_progress` sink. + SubagentTextDelta { + agent_id: String, + task_id: String, + delta: String, + /// 1-based child iteration index this delta belongs to. + iteration: u32, + }, + + /// A chunk of a sub-agent's model reasoning / thinking output + /// arrived (for models that emit `reasoning_content`). Counterpart + /// to [`Self::ThinkingDelta`] scoped to a child run — see + /// [`Self::SubagentTextDelta`] for the attribution rationale. + SubagentThinkingDelta { + agent_id: String, + task_id: String, + delta: String, + /// 1-based child iteration index. + iteration: u32, + }, + /// The agent rewrote the per-thread task board. Emitted by the /// `todo` tool (or `openhuman.todos_*` RPC) after the board has been persisted. TaskBoardUpdated { diff --git a/src/openhuman/agent_orchestration/ops.rs b/src/openhuman/agent_orchestration/ops.rs index 4b963a491..b6505567c 100644 --- a/src/openhuman/agent_orchestration/ops.rs +++ b/src/openhuman/agent_orchestration/ops.rs @@ -421,6 +421,7 @@ impl AgentOrchestrationSession { mode: "typed".to_string(), dedicated_thread: false, prompt_chars: prompt.chars().count(), + worker_thread_id: None, }) .await; } diff --git a/src/openhuman/agent_orchestration/tools.rs b/src/openhuman/agent_orchestration/tools.rs index af3e6a865..a099a6138 100644 --- a/src/openhuman/agent_orchestration/tools.rs +++ b/src/openhuman/agent_orchestration/tools.rs @@ -13,6 +13,8 @@ pub mod spawn_worker_thread; #[cfg(test)] #[path = "tools/tools_e2e_tests.rs"] mod tools_e2e_tests; +#[path = "tools/worker_thread.rs"] +mod worker_thread; pub(crate) use dispatch::dispatch_subagent; diff --git a/src/openhuman/agent_orchestration/tools/dispatch.rs b/src/openhuman/agent_orchestration/tools/dispatch.rs index 42518bb14..b7679f9b0 100644 --- a/src/openhuman/agent_orchestration/tools/dispatch.rs +++ b/src/openhuman/agent_orchestration/tools/dispatch.rs @@ -57,6 +57,7 @@ pub(crate) async fn dispatch_subagent( mode: "typed".to_string(), dedicated_thread: false, prompt_chars: prompt.chars().count(), + worker_thread_id: None, }) .await; } diff --git a/src/openhuman/agent_orchestration/tools/spawn_parallel_agents.rs b/src/openhuman/agent_orchestration/tools/spawn_parallel_agents.rs index d7028f94f..10fefdbc7 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_parallel_agents.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_parallel_agents.rs @@ -265,6 +265,7 @@ impl Tool for SpawnParallelAgentsTool { mode: "typed".to_string(), dedicated_thread: false, prompt_chars: prompt.chars().count(), + worker_thread_id: None, }) .await { diff --git a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs index 4b0e45485..032da368b 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs @@ -392,6 +392,25 @@ impl Tool for SpawnSubagentTool { .unwrap_or_else(|| "standalone".into()); let task_id = format!("sub-{}", uuid::Uuid::new_v4()); + // Persist this delegation as a reopenable worker sub-thread, seeded + // with the prompt, so the parent↔subagent conversation survives + // navigation and restarts — the same machinery `spawn_worker_thread` + // uses. Best-effort: with no parent context or thread store the run + // still proceeds live-only (`worker_thread_id: None`). + let worker_thread_id = current_parent().and_then(|p| { + let parent_thread_id = + crate::openhuman::inference::provider::thread_context::current_thread_id()?; + let title: String = prompt.chars().take(60).collect(); + super::worker_thread::create_worker_thread( + p.workspace_dir.clone(), + &parent_thread_id, + &definition.id, + &title, + &prompt, + ) + .ok() + }); + publish_global(DomainEvent::SubagentSpawned { parent_session: parent_session.clone(), agent_id: definition.id.clone(), @@ -413,6 +432,7 @@ impl Tool for SpawnSubagentTool { mode: "typed".to_string(), dedicated_thread, prompt_chars: prompt.chars().count(), + worker_thread_id: worker_thread_id.clone(), }) .await; } @@ -424,7 +444,7 @@ impl Tool for SpawnSubagentTool { context, model_override, task_id: Some(task_id.clone()), - worker_thread_id: None, + worker_thread_id: worker_thread_id.clone(), }; let progress_sink = current_parent().and_then(|p| p.on_progress.clone()); diff --git a/src/openhuman/agent_orchestration/tools/spawn_worker_thread.rs b/src/openhuman/agent_orchestration/tools/spawn_worker_thread.rs index d04ac40ca..4a7e7aeb8 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_worker_thread.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_worker_thread.rs @@ -12,9 +12,7 @@ use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; use crate::openhuman::agent::harness::fork_context::current_parent; use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions}; -use crate::openhuman::memory_conversations::{ - self as conversations, ConversationMessage, CreateConversationThread, -}; +use crate::openhuman::memory_conversations::{self as conversations}; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; use async_trait::async_trait; use serde_json::json; @@ -183,46 +181,14 @@ impl Tool for SpawnWorkerThreadTool { .ok_or_else(|| anyhow::anyhow!("agent_id '{}' not found", agent_id))?; // ── Create Worker Thread ─────────────────────────────────────── - let worker_thread_id = format!("worker-{}", uuid::Uuid::new_v4()); - let now = chrono::Utc::now().to_rfc3339(); - - conversations::ensure_thread( + // Shared with `spawn_subagent` so both delegation paths persist an + // identical, reopenable sub-thread seeded with the prompt. + let worker_thread_id = super::worker_thread::create_worker_thread( parent.workspace_dir.clone(), - CreateConversationThread { - id: worker_thread_id.clone(), - title: task_title.clone(), - created_at: now.clone(), - parent_thread_id: Some(current_thread_id.clone()), - labels: Some(vec!["worker".to_string()]), - personality_id: None, - }, - ) - .map_err(|e| anyhow::anyhow!(e))?; - - tracing::info!( - agent_id = %agent_id, - worker_thread_id = %worker_thread_id, - parent_thread_id = %current_thread_id, - task_title = %task_title, - created_at = %now, - "[spawn_worker_thread] created worker thread" - ); - - // Append initial user message to the worker thread - conversations::append_message( - parent.workspace_dir.clone(), - &worker_thread_id, - ConversationMessage { - id: format!("user:{}", uuid::Uuid::new_v4()), - content: prompt.clone(), - message_type: "text".to_string(), - extra_metadata: json!({ - "scope": "worker_thread", - "agent_id": agent_id, - }), - sender: "user".to_string(), - created_at: now, - }, + ¤t_thread_id, + &agent_id, + &task_title, + &prompt, ) .map_err(|e| anyhow::anyhow!(e))?; @@ -294,6 +260,7 @@ mod tests { }; use crate::openhuman::agent::harness::fork_context::with_parent_context; use crate::openhuman::agent::harness::ParentExecutionContext; + use crate::openhuman::memory_conversations::{ConversationMessage, CreateConversationThread}; use std::path::PathBuf; use std::sync::Arc; use tempfile::TempDir; diff --git a/src/openhuman/agent_orchestration/tools/worker_thread.rs b/src/openhuman/agent_orchestration/tools/worker_thread.rs new file mode 100644 index 000000000..c1cde725d --- /dev/null +++ b/src/openhuman/agent_orchestration/tools/worker_thread.rs @@ -0,0 +1,114 @@ +//! Shared helper for materialising a sub-agent's work as a persistent +//! conversation sub-thread. +//! +//! Both `spawn_worker_thread` (explicit) and `spawn_subagent` (inline) +//! back their run with a `worker-` thread linked to the parent so the +//! delegation is reopenable from memory and rendered as a parent↔subagent +//! chat. The thread is created with `parent_thread_id` set (which hides it +//! from the main sidebar — it surfaces in the "Workers" tab and the +//! subagent drawer instead) and seeded with the delegation prompt as the +//! opening `user` message. The sub-agent runner then appends each turn and +//! tool result to the same thread via its `worker_thread_id` sink. + +use std::path::PathBuf; + +use serde_json::json; + +use crate::openhuman::memory_conversations::{ + self as conversations, ConversationMessage, CreateConversationThread, +}; + +/// Create a worker sub-thread linked to `parent_thread_id` and seed it with +/// the delegation `prompt` as the opening user message. Returns the new +/// thread id, or an `Err` string if the thread store rejected the create. +/// +/// The seed-message append is best-effort: a failure there is logged but +/// does not fail the call (the thread still exists and the runner will +/// append the sub-agent's turns). +pub(crate) fn create_worker_thread( + workspace_dir: PathBuf, + parent_thread_id: &str, + agent_id: &str, + title: &str, + prompt: &str, +) -> Result { + let worker_thread_id = format!("worker-{}", uuid::Uuid::new_v4()); + let now = chrono::Utc::now().to_rfc3339(); + + conversations::ensure_thread( + workspace_dir.clone(), + CreateConversationThread { + id: worker_thread_id.clone(), + title: title.to_string(), + created_at: now.clone(), + parent_thread_id: Some(parent_thread_id.to_string()), + labels: Some(vec!["worker".to_string()]), + personality_id: None, + }, + )?; + + tracing::info!( + agent_id = %agent_id, + worker_thread_id = %worker_thread_id, + parent_thread_id = %parent_thread_id, + "[worker_thread] created sub-thread for delegation" + ); + + if let Err(err) = conversations::append_message( + workspace_dir, + &worker_thread_id, + ConversationMessage { + id: format!("user:{}", uuid::Uuid::new_v4()), + content: prompt.to_string(), + message_type: "text".to_string(), + extra_metadata: json!({ "scope": "worker_thread", "agent_id": agent_id }), + sender: "user".to_string(), + created_at: now, + }, + ) { + tracing::warn!( + worker_thread_id = %worker_thread_id, + error = %err, + "[worker_thread] failed to seed delegation prompt (continuing)" + ); + } + + Ok(worker_thread_id) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn creates_child_thread_linked_to_parent_and_seeds_prompt() { + let dir = std::env::temp_dir().join(format!("wt-test-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + + let id = create_worker_thread( + dir.clone(), + "parent-thread-1", + "researcher", + "Q3", + "Find Q3", + ) + .expect("thread should be created"); + + // The new thread is labelled `worker` and linked to the parent — that + // link is what hides it from the main sidebar (Workers tab only). + let threads = conversations::list_threads(dir.clone()).unwrap(); + let thread = threads + .iter() + .find(|t| t.id == id) + .expect("thread persisted"); + assert_eq!(thread.parent_thread_id.as_deref(), Some("parent-thread-1")); + assert!(thread.labels.contains(&"worker".to_string())); + + // It opens with the delegation prompt as the user message, so the + // drawer can render the parent↔subagent chat from memory on reopen. + let messages = conversations::get_messages(dir, &id).unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].sender, "user"); + assert_eq!(messages[0].content, "Find Q3"); + } +} diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index 17a70d916..f9a532ad0 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -1102,6 +1102,7 @@ fn spawn_progress_bridge( mode, dedicated_thread, prompt_chars, + worker_thread_id, } => { publish_web_channel_event(WebChannelEvent { event: "subagent_spawned".to_string(), @@ -1116,6 +1117,7 @@ fn spawn_progress_bridge( mode: Some(mode), dedicated_thread: Some(dedicated_thread), prompt_chars: Some(prompt_chars as u64), + worker_thread_id, ..Default::default() }), ..Default::default() @@ -1252,6 +1254,54 @@ fn spawn_progress_bridge( ..Default::default() }); } + AgentProgress::SubagentTextDelta { + agent_id, + task_id, + delta, + iteration, + } => { + publish_web_channel_event(WebChannelEvent { + event: "subagent_text_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + round: Some(round), + delta: Some(delta), + delta_kind: Some("text".to_string()), + skill_id: Some(task_id.clone()), + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + agent_id: Some(agent_id), + task_id: Some(task_id), + ..Default::default() + }), + ..Default::default() + }); + } + AgentProgress::SubagentThinkingDelta { + agent_id, + task_id, + delta, + iteration, + } => { + publish_web_channel_event(WebChannelEvent { + event: "subagent_thinking_delta".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + request_id: request_id.clone(), + round: Some(round), + delta: Some(delta), + delta_kind: Some("thinking".to_string()), + skill_id: Some(task_id.clone()), + subagent: Some(SubagentProgressDetail { + child_iteration: Some(iteration), + agent_id: Some(agent_id), + task_id: Some(task_id), + ..Default::default() + }), + ..Default::default() + }); + } AgentProgress::TaskBoardUpdated { board } => { log::debug!( "[web_channel][bridge] task_board_updated client_id={} thread_id={} request_id={} cards={}", diff --git a/src/openhuman/skills/run_log.rs b/src/openhuman/skills/run_log.rs index ea73057bf..3cdf1219e 100644 --- a/src/openhuman/skills/run_log.rs +++ b/src/openhuman/skills/run_log.rs @@ -168,6 +168,8 @@ pub fn format_event(ev: &AgentProgress) -> Option { | AgentProgress::ToolCallArgsDelta { .. } | AgentProgress::TurnCostUpdated { .. } | AgentProgress::TaskBoardUpdated { .. } + | AgentProgress::SubagentTextDelta { .. } + | AgentProgress::SubagentThinkingDelta { .. } | AgentProgress::SubagentIterationStarted { .. } => return None, }; Some(format!( diff --git a/src/openhuman/threads/turn_state/mirror.rs b/src/openhuman/threads/turn_state/mirror.rs index 1356f070b..c9899204d 100644 --- a/src/openhuman/threads/turn_state/mirror.rs +++ b/src/openhuman/threads/turn_state/mirror.rs @@ -144,6 +144,7 @@ impl TurnStateMirror { task_id, mode, dedicated_thread, + worker_thread_id, .. } => { self.state.phase = Some(TurnPhase::Subagent); @@ -167,6 +168,7 @@ impl TurnStateMirror { iterations: None, elapsed_ms: None, output_chars: None, + worker_thread_id: worker_thread_id.clone(), tool_calls: Vec::new(), }), }); @@ -265,6 +267,17 @@ impl TurnStateMirror { } false } + AgentProgress::SubagentTextDelta { .. } + | AgentProgress::SubagentThinkingDelta { .. } => { + // Sub-agent streaming text/thinking is display-only: it is + // rendered live in the parent thread's subagent transcript + // but intentionally **not** persisted to the turn-state + // snapshot. The child's final assistant text lands in the + // thread on completion, so replaying partial deltas after a + // reconnect would add weight without value. Acknowledge + // without mutating the snapshot or flushing. + false + } AgentProgress::TaskBoardUpdated { board } => { self.state.task_board = Some(board.clone()); self.flush(); diff --git a/src/openhuman/threads/turn_state/mirror_tests.rs b/src/openhuman/threads/turn_state/mirror_tests.rs index 5dd16cd15..909a0fdc7 100644 --- a/src/openhuman/threads/turn_state/mirror_tests.rs +++ b/src/openhuman/threads/turn_state/mirror_tests.rs @@ -224,6 +224,7 @@ fn subagent_lifecycle_records_and_clears_active() { mode: "typed".into(), dedicated_thread: false, prompt_chars: 42, + worker_thread_id: None, }); let s = m.snapshot(); assert_eq!(s.active_subagent.as_deref(), Some("researcher")); diff --git a/src/openhuman/threads/turn_state/types.rs b/src/openhuman/threads/turn_state/types.rs index 447000b36..4e71bc86b 100644 --- a/src/openhuman/threads/turn_state/types.rs +++ b/src/openhuman/threads/turn_state/types.rs @@ -87,6 +87,11 @@ pub struct SubagentActivity { pub elapsed_ms: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub output_chars: Option, + /// Persistent worker sub-thread backing this delegation, when one was + /// created. Lets the UI reopen the full parent↔subagent conversation + /// from memory after a cold boot / interrupted turn. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_thread_id: Option, #[serde(default)] pub tool_calls: Vec, }