abort and search

src/api/ApiClient.js

@@ -116,13 +116,14 @@ export class ApiClient {
        * Returns an async generator that yields SSE event objects:
        *   { event: 'status'|'llm_call'|'tool_call'|'usage'|'progress'|'result'|'error', data: object }
        */
-      parseSse: async function* (file) {
+      parseSse: async function* (file, { signal } = {}) {
         const formData = new FormData();
         formData.append('file', file, file.name);
         const res = await fetch('/api/delivery/parse', {
           method: 'POST',
           credentials: 'include',
           body: formData,
+          signal,
         });
         if (!res.ok) {
           let message = res.statusText;
@@ -138,28 +139,56 @@ export class ApiClient {
         const decoder = new TextDecoder();
         let buf = '';
         let currentEvent = 'message';
-        while (true) {
-          const { done, value } = await reader.read();
-          if (done) break;
-          buf += decoder.decode(value, { stream: true });
-          const lines = buf.split('\n');
-          buf = lines.pop(); // keep incomplete last line
-          for (const line of lines) {
-            if (line.startsWith('event:')) {
-              currentEvent = line.slice(6).trim();
-            } else if (line.startsWith('data:')) {
-              const raw = line.slice(5).trim();
-              let data;
-              try { data = JSON.parse(raw); } catch { data = raw; }
-              yield { event: currentEvent, data };
-              currentEvent = 'message';
-            }
-            // blank line or comment: reset
+
+        const onAbort = () => {
+          try {
+            reader.cancel();
+          } catch { /* ignore */ }
+        };
+        if (signal) {
+          if (signal.aborted) {
+            onAbort();
+            return;
           }
+          signal.addEventListener('abort', onAbort, { once: true });
+        }
+
+        try {
+          while (true) {
+            const { done, value } = await reader.read();
+            if (done) break;
+            buf += decoder.decode(value, { stream: true });
+            const lines = buf.split('\n');
+            buf = lines.pop(); // keep incomplete last line
+            for (const line of lines) {
+              if (line.startsWith('event:')) {
+                currentEvent = line.slice(6).trim();
+              } else if (line.startsWith('data:')) {
+                const raw = line.slice(5).trim();
+                let data;
+                try { data = JSON.parse(raw); } catch { data = raw; }
+                yield { event: currentEvent, data };
+                currentEvent = 'message';
+              }
+              // blank line or comment: reset
+            }
+          }
+        } catch (err) {
+          if (signal?.aborted || err.name === 'AbortError') {
+            return;
+          }
+          throw err;
+        } finally {
+          if (signal) {
+            signal.removeEventListener('abort', onAbort);
+          }
+          try {
+            reader.releaseLock();
+          } catch { /* ignore */ }
         }
       },
-      parsePdfSse(file) {
-        return this.parseSse(file);
+      parsePdfSse(file, opts) {
+        return this.parseSse(file, opts);
       },
     };
 

src/api/index.js

@@ -710,13 +710,14 @@ export const delivery = {
    * Accepts PDF, TXT or CSV document (max 50 MB)
    * Streams SSE events: status, llm_call, tool_call, usage, progress, result, error
    */
-  async *parseSse(file) {
+  async *parseSse(file, { signal } = {}) {
     const formData = new FormData();
     formData.append('file', file, file.name);
     const res = await fetch('/api/delivery/parse', {
       method: 'POST',
       credentials: 'include',
       body: formData,
+      signal,
     });
     if (!res.ok) {
       let message = res.statusText;
@@ -732,23 +733,51 @@ export const delivery = {
     const decoder = new TextDecoder();
     let buf = '';
     let currentEvent = 'message';
-    while (true) {
-      const { done, value } = await reader.read();
-      if (done) break;
-      buf += decoder.decode(value, { stream: true });
-      const lines = buf.split('\n');
-      buf = lines.pop();
-      for (const line of lines) {
-        if (line.startsWith('event:')) {
-          currentEvent = line.slice(6).trim();
-        } else if (line.startsWith('data:')) {
-          const raw = line.slice(5).trim();
-          let data;
-          try { data = JSON.parse(raw); } catch { data = raw; }
-          yield { event: currentEvent, data };
-          currentEvent = 'message';
+
+    const onAbort = () => {
+      try {
+        reader.cancel();
+      } catch { /* ignore */ }
+    };
+    if (signal) {
+      if (signal.aborted) {
+        onAbort();
+        return;
+      }
+      signal.addEventListener('abort', onAbort, { once: true });
+    }
+
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        buf += decoder.decode(value, { stream: true });
+        const lines = buf.split('\n');
+        buf = lines.pop();
+        for (const line of lines) {
+          if (line.startsWith('event:')) {
+            currentEvent = line.slice(6).trim();
+          } else if (line.startsWith('data:')) {
+            const raw = line.slice(5).trim();
+            let data;
+            try { data = JSON.parse(raw); } catch { data = raw; }
+            yield { event: currentEvent, data };
+            currentEvent = 'message';
+          }
         }
       }
+    } catch (err) {
+      if (signal?.aborted || err.name === 'AbortError') {
+        return;
+      }
+      throw err;
+    } finally {
+      if (signal) {
+        signal.removeEventListener('abort', onAbort);
+      }
+      try {
+        reader.releaseLock();
+      } catch { /* ignore */ }
     }
   },
 };