feat(kb/ask): live SSE streaming of Shira's progress (v0.2.0)

User wanted to see what Shira is doing while she thinks — like Claude does — instead of staring at an incrementing seconds counter. Now the ask UI shows one line per agent step as it happens: 3s 🧠 קוראת את השאלה 4s 🔍 מחפשת בבסיס הידע: "תקנה 36" 9s ✓ נמצאו 5 קטעים רלוונטיים 10s 🔍 מחפשת בבסיס הידע: "תקנה 36 ביטוח לאומי" 14s ✓ נמצאו 5 קטעים רלוונטיים 15s 🧠 ממשיכה לחקור 22s ✍️ מנסחת תשובה → תשובה Client wiring: - runAsk now opens an EventSource at ?entryPoint=KnowledgeBaseAskStream instead of POSTing JSON. Each SSE event of type thinking/tool_start/ tool_done/tool_error/writing appends a row to a stacked progress log. The final {type:answer,text,sources} event triggers renderAskAnswer with the existing split view. {type:error} surfaces via showError. - Module-level state from v0.1.9 unchanged in spirit but now stores the EventSource + accumulated events. afterRender on view re-mount replays the entire event list and re-binds onmessage/onerror — the SSE connection itself doesn't drop because it lives at module scope. - Elapsed-time pill stays for cadence but is decoration; the real signal is the per-event log. EspoCRM proxy: - New EntryPoint KnowledgeBaseAskStream — required because EventSource is GET-only and ?message= must come via the URL. PHP buffering is disabled (zlib + ob + apache_setenv no-gzip), Apache headers are set for SSE (no-cache, X-Accel-Buffering: no), and we cURL into shira-hermes /kb/ask/stream with CURLOPT_WRITEFUNCTION echoing each byte and flushing immediately. The handler ends with exit; to bypass EspoCRM's Response wrapper which would otherwise emit headers/body on top of our raw stream. - KnowledgeBaseService gained two public accessors (getStreamingUrl, getStreamingApiKey) so the EntryPoint can build the cURL without going through request() (which buffers the whole body). - entryPoints.json registers the new class. Server (shira-hermes 76fd77f): - AgentRunner.run accepts an on_event sync callback fired before each LLM call (thinking) and around each tool call (tool_start, tool_done, tool_error). Hebrew labels in agent_runner._tool_label / _tool_done_label ship over the wire ready-to-display — keeps the client free of translation concerns. - POST /kb/ask/stream wraps the existing agent runner in an asyncio task feeding an asyncio.Queue, returns StreamingResponse with proper SSE headers (X-Accel-Buffering, Cache-Control). 15s comment-line heartbeats keep the connection open through proxy idle-timeouts. - Original POST /kb/ask still works — additive change. Refs Task Master #8
2026-04-25 13:00:58 +00:00
parent 2fe10727eb
commit a00fc3729c
6 changed files with 307 additions and 63 deletions
@@ -0,0 +1,128 @@
+<?php
+
+namespace Espo\Modules\KnowledgeBase\EntryPoints;
+
+use Espo\Core\Api\Request;
+use Espo\Core\Api\Response;
+use Espo\Core\EntryPoint\EntryPoint;
+use Espo\Core\Exceptions\BadRequest;
+use Espo\Core\Exceptions\Error;
+use Espo\Core\Exceptions\Forbidden;
+use Espo\Core\InjectableFactory;
+use Espo\Entities\User;
+use Espo\Modules\KnowledgeBase\Services\KnowledgeBaseService;
+
+/**
+ * Streams Server-Sent Events from shira-hermes /kb/ask/stream to the
+ * browser's EventSource. The browser opens this with a GET (EventSource
+ * has no POST), passes the question via ?message=, and we forward it
+ * to shira-hermes with the firm-wide API key — the credential never
+ * leaves the EspoCRM server.
+ *
+ * Response is text/event-stream with no buffering. We bypass EspoCRM's
+ * Response wrapper entirely (echo + flush + exit) because the framework
+ * is wired for buffered JSON responses, not infinite streams.
+ *
+ * Access: `?entryPoint=KnowledgeBaseAskStream&message=<urlencoded>`
+ */
+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;
+    }
+}
@@ -1,5 +1,8 @@
 {
    "KnowledgeBasePdf": {
        "className": "Espo\\Modules\\KnowledgeBase\\EntryPoints\\KnowledgeBasePdf"
+    },
+    "KnowledgeBaseAskStream": {
+        "className": "Espo\\Modules\\KnowledgeBase\\EntryPoints\\KnowledgeBaseAskStream"
    }
 }
@@ -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);