diff --git a/.taskmaster/tasks/tasks.json b/.taskmaster/tasks/tasks.json index 823dd9d..4b47a87 100644 --- a/.taskmaster/tasks/tasks.json +++ b/.taskmaster/tasks/tasks.json @@ -37,6 +37,18 @@ "dependencies": [], "createdAt": "2026-04-25T10:00:00Z" }, + { + "id": 8, + "title": "feat(kb/ask): SSE streaming with live agent progress", + "description": "User wanted Shira's progress to be visible while she thinks — like Claude does — instead of just an incrementing seconds counter. Implementation: AgentRunner.run accepts an optional sync on_event callback that emits {type, label, ...} dicts at each agent step. New POST /kb/ask/stream returns text/event-stream, with the runner running in an asyncio task whose events feed an asyncio.Queue that the response generator drains as SSE. Client opens EventSource against a new EspoCRM EntryPoint KnowledgeBaseAskStream that proxies the SSE bytes through cURL with output buffering disabled (echo + flush + exit, bypassing the framework Response). The progress UI now stacks one line per event (thinking, tool_start with the query Shira chose, tool_done with the chunk count, writing) instead of just a seconds counter. v0.1.9 _activeAsk persistence pattern preserved — events accumulate at module scope so a view switch + return replays the live log.", + "status": "done", + "priority": "high", + "details": "shira-hermes 76fd77f. EspoCRM EntryPoint registered in metadata/app/entryPoints.json. Hebrew labels live in agent_runner._tool_label / _tool_done_label so the wire format already carries display-ready strings (no client-side mapping). 15s heartbeats keep the connection open through proxy idle-timeouts.", + "testStrategy": "curl -N POST /kb/ask/stream {message:'מהי תקנה 36'} should produce data: lines with thinking → tool_start (with query) → tool_done (with chunk count) → ... → writing → answer. Browser EventSource should drive the same flow into the UI.", + "subtasks": [], + "dependencies": [], + "createdAt": "2026-04-25T13:00:00Z" + }, { "id": 7, "title": "fix: search preview duplicates left card for text sources", diff --git a/files/client/custom/modules/knowledge-base/src/views/kb/index.js b/files/client/custom/modules/knowledge-base/src/views/kb/index.js index 387c862..6117ea4 100644 --- a/files/client/custom/modules/knowledge-base/src/views/kb/index.js +++ b/files/client/custom/modules/knowledge-base/src/views/kb/index.js @@ -115,7 +115,9 @@ define('modules/knowledge-base/views/kb/index', ['view'], function (Dep) { const $input = this.$el.find('input[data-name="query"]'); if ($input.length) $input.val(_activeAsk.question); this.setLoading(true); - this.startAskProgress(_activeAsk.startedAt); + // Pass the whole ask so startAskProgress can replay the + // accumulated event log into the freshly-rendered DOM. + this.startAskProgress(_activeAsk); this._attachAskHandlers(_activeAsk); } else if (_lastAsk) { const $input = this.$el.find('input[data-name="query"]'); @@ -205,93 +207,173 @@ define('modules/knowledge-base/views/kb/index', ['view'], function (Dep) { _clear(sessionStorage, SS_ASK); this.setLoading(true); - // Ask goes through the full agent loop (search + rerank + LLM), - // which routinely takes 30-90s. The default jQuery Ajax timeout - // is too short; override it here so the browser doesn't give up - // before shira-hermes responds. - const promise = Espo.Ajax.postRequest('KnowledgeBase/action/ask', { - message: message, - }, {timeout: 240000}); - _activeAsk = {question: message, promise: promise, startedAt: Date.now()}; - this.startAskProgress(_activeAsk.startedAt); + // Streaming via Server-Sent Events. The browser opens an + // EventSource against KnowledgeBaseAskStream which proxies + // /kb/ask/stream on shira-hermes; one SSE event per agent step + // (thinking, tool_start, tool_done) plus a final answer event. + const url = '?entryPoint=KnowledgeBaseAskStream&message=' + + encodeURIComponent(message); + const es = new EventSource(url); + _activeAsk = { + question: message, + es: es, + events: [], + startedAt: Date.now(), + }; + this.startAskProgress(_activeAsk); this._attachAskHandlers(_activeAsk); }, - // Wire .then/.catch on a (possibly module-level) ask promise to - // whichever view is currently mounted. Called from runAsk for the - // initial bind and from afterRender when the user navigates back to - // the KB view while an earlier request is still in flight. + // Wire SSE handlers on a (possibly module-level) ask. Called from + // runAsk for the initial bind and from afterRender when the user + // navigates back to the KB view while an earlier stream is still + // open. The EventSource itself lives at module scope (in _activeAsk) + // so handlers can be re-bound without dropping the connection. _attachAskHandlers: function (ask) { const self = this; - ask.promise.then(res => { - const entry = { - question: ask.question, - text: res.text || '', - sources: res.sources || [], - completedAt: Date.now(), - }; - _lastAsk = entry; - _saveJson(sessionStorage, SS_ASK, entry); - if (_activeAsk === ask) _activeAsk = null; - if (self.$el && self.$el.length) { - self.stopAskProgress(); - self.setLoading(false); - if (self.mode === 'ask') { - self.renderAskAnswer(entry.text, entry.sources); + ask.es.onmessage = function (msgEvent) { + let data; + try { data = JSON.parse(msgEvent.data); } + catch (e) { + console.warn('KB: malformed SSE payload', e); + return; + } + if (data.type === 'answer') { + self._finishAsk(ask, { + question: ask.question, + text: data.text || '', + sources: data.sources || [], + completedAt: Date.now(), + }); + } else if (data.type === 'error') { + self._failAsk(ask, data.message || 'שגיאה בשרת'); + } else { + // Status event — record it and append to the live log. + ask.events.push(Object.assign({_clientTs: Date.now()}, data)); + if (self.$el && self.$el.length && self.mode === 'ask') { + self._appendAskEvent(ask, ask.events[ask.events.length - 1]); } } - }).catch(err => { - if (_activeAsk === ask) _activeAsk = null; - if (self.$el && self.$el.length) { - self.stopAskProgress(); - self.setLoading(false); - self.showError(err); - } - }); + }; + ask.es.onerror = function () { + // Don't tear down on transient blips while the answer is + // already arriving; only treat it as a hard failure if no + // answer has been delivered. + if (ask._completed) return; + self._failAsk(ask, 'הזרם נסגר לפני שהגיעה תשובה'); + }; }, - startAskProgress: function (startedAt) { - // Show a live elapsed-time indicator so users know the request - // is still running rather than stuck. Typical answers arrive in - // 30-90s; we warn after 2 minutes. `startedAt` is passed in so - // a request that began before a view re-mount keeps showing the - // accumulated elapsed time, not 0. + _finishAsk: function (ask, entry) { + ask._completed = true; + try { ask.es.close(); } catch (e) { /* ignore */ } + _lastAsk = entry; + _saveJson(sessionStorage, SS_ASK, entry); + if (_activeAsk === ask) _activeAsk = null; + if (this.$el && this.$el.length) { + this.stopAskProgress(); + this.setLoading(false); + if (this.mode === 'ask') { + this.renderAskAnswer(entry.text, entry.sources); + } + } + }, + + _failAsk: function (ask, message) { + try { ask.es.close(); } catch (e) { /* ignore */ } + if (_activeAsk === ask) _activeAsk = null; + if (this.$el && this.$el.length) { + this.stopAskProgress(); + this.setLoading(false); + this.showError(new Error(message)); + } + }, + + startAskProgress: function (askOrStartedAt) { + // Render a live progress panel: a "thinking" header with elapsed + // time, plus a stacked log of the events Shira's emitted so far + // (one line per LLM step / tool call). `askOrStartedAt` is + // either the active-ask object (from runAsk and afterRender) or + // a bare timestamp (legacy call sites). Re-mounting on view + // switch replays the accumulated event list from ask.events. if (this.mode !== 'ask') return; this.stopAskProgress(); - startedAt = startedAt || Date.now(); + let ask = null; + let startedAt; + if (askOrStartedAt && typeof askOrStartedAt === 'object') { + ask = askOrStartedAt; + startedAt = ask.startedAt || Date.now(); + } else { + startedAt = askOrStartedAt || Date.now(); + } + const $results = this.$el.find('.kb-results'); $results.html( '
' + '
' + - ' ' + - ' שירה חושבת…' + - '
' + - ' שירה מחפשת בבסיס הידע, מסכמת ומציגה ציטוטים מדויקים.' + - ' זה יכול לקחת בין 30 ל-90 שניות.' + - '
' + - '
' + - ' זמן שעבר: ' + - ' 0' + - ' שניות' + + '
' + + ' ' + + ' שירה חושבת' + + ' 0 שניות' + '
' + + '
' + '
' + '
' ); + + // Replay any events already collected (matters when the user + // navigates away mid-think and comes back to a still-streaming + // ask — the stream kept going at module scope, but the DOM was + // freshly re-rendered). + if (ask && Array.isArray(ask.events)) { + ask.events.forEach(ev => this._appendAskEvent(ask, ev)); + } + const tick = () => { const sec = Math.round((Date.now() - startedAt) / 1000); - const $elapsed = this.$el.find('.kb-elapsed'); - if ($elapsed.length) $elapsed.text(sec); - if (sec === 120) { - this.$el.find('.kb-ask-progress .panel-body').append( - '
' + - 'לוקח יותר זמן מהרגיל. עדיין ממתינים…
' - ); - } + this.$el.find('.kb-elapsed-pill').text(sec + ' שניות'); }; tick(); this._askTimer = setInterval(tick, 1000); }, + _appendAskEvent: function (ask, ev) { + const $events = this.$el.find('.kb-events'); + if (!$events.length) return; + const startedAt = (ask && ask.startedAt) || Date.now(); + const ts = ev._clientTs || Date.now(); + const sec = Math.max(0, Math.round((ts - startedAt) / 1000)); + const icon = { + thinking: '🧠', + tool_start: '🔍', + tool_done: '✓', + tool_error: '⚠', + writing: '✍️', + }[ev.type] || '·'; + const colour = { + thinking: '#475569', + tool_start: '#0f5298', + tool_done: '#1e6f40', + tool_error: '#a02622', + writing: '#5a3a91', + }[ev.type] || '#475569'; + const label = this.escape(ev.label || ev.type); + const row = $( + '
' + + ' ' + + sec + 'ש' + + ' ' + icon + '' + + ' ' + label + '' + + '
' + ); + $events.append(row); + // Always keep the latest line visible. + $events.scrollTop($events[0].scrollHeight); + }, + stopAskProgress: function () { if (this._askTimer) { clearInterval(this._askTimer); diff --git a/files/custom/Espo/Modules/KnowledgeBase/EntryPoints/KnowledgeBaseAskStream.php b/files/custom/Espo/Modules/KnowledgeBase/EntryPoints/KnowledgeBaseAskStream.php new file mode 100644 index 0000000..0059c7f --- /dev/null +++ b/files/custom/Espo/Modules/KnowledgeBase/EntryPoints/KnowledgeBaseAskStream.php @@ -0,0 +1,128 @@ +` + */ +class KnowledgeBaseAskStream implements EntryPoint +{ + public function __construct( + private InjectableFactory $injectableFactory, + private User $user, + ) {} + + public function run(Request $request, Response $response): void + { + if ($this->user->isPortal()) { + throw new Forbidden('Portal users have no KB access.'); + } + + $message = $request->getQueryParam('message'); + if (!is_string($message) || trim($message) === '') { + throw new BadRequest('message (string) is required.'); + } + if (strlen($message) > 2000) { + throw new BadRequest('message is too long.'); + } + + $service = $this->injectableFactory->create(KnowledgeBaseService::class); + try { + $upstreamUrl = $service->getStreamingUrl('/kb/ask/stream'); + $apiKey = $service->getStreamingApiKey(); + } catch (\Throwable $e) { + throw new Error('SmartAssistant integration is not configured: ' . $e->getMessage()); + } + + // Disable PHP/Apache output buffering and compression so each SSE + // event flushes to the wire as soon as it arrives from upstream. + // Order matters: clean any open buffers BEFORE setting headers. + while (ob_get_level() > 0) { + @ob_end_clean(); + } + @ini_set('zlib.output_compression', 'Off'); + @ini_set('output_buffering', 'Off'); + @ini_set('implicit_flush', '1'); + if (function_exists('apache_setenv')) { + @apache_setenv('no-gzip', '1'); + } + + ignore_user_abort(false); + @set_time_limit(300); + + header('Content-Type: text/event-stream; charset=utf-8'); + header('Cache-Control: no-cache, no-transform'); + header('X-Accel-Buffering: no'); + header('Connection: keep-alive'); + // Defense in depth: same locked-down CSP we use for the PDF entry. + header("Content-Security-Policy: default-src 'self'"); + + $payload = json_encode(['message' => $message], JSON_UNESCAPED_UNICODE); + + $ch = curl_init(); + curl_setopt_array($ch, [ + CURLOPT_URL => $upstreamUrl, + CURLOPT_POST => true, + CURLOPT_POSTFIELDS => $payload, + CURLOPT_HTTPHEADER => [ + 'Content-Type: application/json', + 'Accept: text/event-stream', + 'X-Api-Key: ' . $apiKey, + ], + CURLOPT_RETURNTRANSFER => false, + CURLOPT_BUFFERSIZE => 1024, + CURLOPT_TIMEOUT => 300, + CURLOPT_CONNECTTIMEOUT => 10, + CURLOPT_WRITEFUNCTION => function ($ch, $data) { + if (connection_aborted()) { + // Browser closed the EventSource — bail so cURL stops. + return -1; + } + echo $data; + @ob_flush(); + @flush(); + return strlen($data); + }, + ]); + $ok = curl_exec($ch); + $errno = curl_errno($ch); + $errmsg = curl_error($ch); + curl_close($ch); + + if (!$ok && $errno !== 23 /* aborted by callback */) { + // Best effort: if upstream errored before any data was + // flushed, surface a final SSE error event so the client + // doesn't silently spin. + echo "data: " . json_encode( + ['type' => 'error', 'message' => 'upstream: ' . $errmsg], + JSON_UNESCAPED_UNICODE, + ) . "\n\n"; + @ob_flush(); + @flush(); + } + + // We've fully written the response. Bypass EspoCRM's Response + // emit() to avoid double-headers / extra buffering. + exit; + } +} diff --git a/files/custom/Espo/Modules/KnowledgeBase/Resources/metadata/app/entryPoints.json b/files/custom/Espo/Modules/KnowledgeBase/Resources/metadata/app/entryPoints.json index 7ef6603..9130c85 100644 --- a/files/custom/Espo/Modules/KnowledgeBase/Resources/metadata/app/entryPoints.json +++ b/files/custom/Espo/Modules/KnowledgeBase/Resources/metadata/app/entryPoints.json @@ -1,5 +1,8 @@ { "KnowledgeBasePdf": { "className": "Espo\\Modules\\KnowledgeBase\\EntryPoints\\KnowledgeBasePdf" + }, + "KnowledgeBaseAskStream": { + "className": "Espo\\Modules\\KnowledgeBase\\EntryPoints\\KnowledgeBaseAskStream" } } diff --git a/files/custom/Espo/Modules/KnowledgeBase/Services/KnowledgeBaseService.php b/files/custom/Espo/Modules/KnowledgeBase/Services/KnowledgeBaseService.php index 6e8d5a4..6bff3f8 100644 --- a/files/custom/Espo/Modules/KnowledgeBase/Services/KnowledgeBaseService.php +++ b/files/custom/Espo/Modules/KnowledgeBase/Services/KnowledgeBaseService.php @@ -140,6 +140,25 @@ class KnowledgeBaseService return $data->apiKey ?? null; } + /** + * Public accessors so the streaming EntryPoint can build a cURL + * connection to shira-hermes without going through the JSON request() + * helper (which buffers the whole body). + */ + public function getStreamingUrl(string $path): string + { + return rtrim($this->getBaseUrl(), '/') . $path; + } + + public function getStreamingApiKey(): string + { + $key = $this->getApiKey(); + if (!$key) { + throw new Error('SmartAssistant API key is not configured.'); + } + return $key; + } + private function get(string $path): array { return $this->request('GET', $path, null); diff --git a/manifest.json b/manifest.json index 2d6f7d3..a3e8202 100644 --- a/manifest.json +++ b/manifest.json @@ -1,7 +1,7 @@ { "name": "KnowledgeBase", "module": "KnowledgeBase", - "version": "0.1.11", + "version": "0.2.0", "acceptableVersions": [ ">=8.0.0" ],