From c81d72fb9c8c8fc525a4e89b52ef56b03808ea24 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?=
 <zhangyanghaha0407@outlook.com>
Date: Thu, 16 Apr 2026 21:57:35 -0400
Subject: [PATCH] fix(api-server): SSE stream pulls incremental events via seq
 cursor

Replace per-poll 5000-row scan + JS filter with indexed WHERE seq > ?
query (uses idx_run_events_run). Batch up to 500 per call, drain loop
on reconnect. Poll interval raised 500ms -> 1000ms.
---
 src/main/api-server/routes-runs.ts | 17 ++++++++++-------
 src/main/database-base.ts          |  1 +
 src/main/database/runs.ts          | 22 ++++++++++++++++++++++
 3 files changed, 33 insertions(+), 7 deletions(-)

diff --git a/src/main/api-server/routes-runs.ts b/src/main/api-server/routes-runs.ts
index 2936d4f5..bd0c6bbf 100644
--- a/src/main/api-server/routes-runs.ts
+++ b/src/main/api-server/routes-runs.ts
@@ -3,7 +3,8 @@ import { requireParam, type AgentApiServer } from "./server-core";
 
 const DEFAULT_EVENT_LIMIT = 500;
 const MAX_EVENT_LIMIT = 5_000;
-const EVENT_POLL_INTERVAL_MS = 500;
+const SSE_BATCH_LIMIT = 500;
+const EVENT_POLL_INTERVAL_MS = 1_000;
 const SSE_HEARTBEAT_MS = 15_000;
 
 function parsePositiveInt(value: string | null, fallback: number) {
@@ -64,12 +65,14 @@ export function registerRunRoutes(server: AgentApiServer) {
     writeSse(res, "connected", { runId, after: cursor });
 
     const flushPendingEvents = () => {
-      const events = server.db
-        .listRunEvents(runId, token.companyId, { offset: 0, limit: MAX_EVENT_LIMIT })
-        .filter((event) => event.seq > cursor);
-      for (const event of events) {
-        writeSse(res, "run-event", event);
-        cursor = event.seq;
+      while (true) {
+        const events = server.db.listRunEventsAfter(runId, token.companyId, cursor, SSE_BATCH_LIMIT);
+        if (events.length === 0) return;
+        for (const event of events) {
+          writeSse(res, "run-event", event);
+          cursor = event.seq;
+        }
+        if (events.length < SSE_BATCH_LIMIT) return;
       }
     };
 
diff --git a/src/main/database-base.ts b/src/main/database-base.ts
index ce081b94..9ea64bb4 100644
--- a/src/main/database-base.ts
+++ b/src/main/database-base.ts
@@ -276,6 +276,7 @@ export class AppDatabaseBase {
   getInterruptedRunsForRecovery() { return this.runs.getInterruptedRunsForRecovery(); }
   appendRunEvent(input: { companyId: string; runId: string; agentId: string; eventType: string; stream?: string | null; level?: string | null; message?: string | null; payload?: Record<string, unknown> | null }) { return this.runs.appendRunEvent(input); }
   listRunEvents(runId: string, companyId: string, opts?: { offset?: number; limit?: number }) { return this.runs.listRunEvents(runId, companyId, opts); }
+  listRunEventsAfter(runId: string, companyId: string, afterSeq: number, limit: number) { return this.runs.listRunEventsAfter(runId, companyId, afterSeq, limit); }
   updateRunActor(runId: string, actorType: string, actorId: string | null) { return this.runs.updateRunActor(runId, actorType, actorId); }
 
   // ─── Approvals ─────────────────────────────────────────────────────────────
diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts
index a77c7dd6..1a76773d 100644
--- a/src/main/database/runs.ts
+++ b/src/main/database/runs.ts
@@ -340,6 +340,28 @@ export function buildRunMethods(
       }));
     },
 
+    listRunEventsAfter(runId: string, companyId: string, afterSeq: number, limit: number): Array<{
+      id: number; runId: string; agentId: string; seq: number; eventType: string;
+      stream: string | null; level: string | null; message: string | null;
+      payload: Record<string, unknown> | null; createdAt: string;
+    }> {
+      return readAll<SqliteRow>(
+        "SELECT id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at FROM run_events WHERE run_id = ? AND company_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?",
+        runId, companyId, afterSeq, limit
+      ).map((row: SqliteRow) => ({
+        id: row.id,
+        runId: row.run_id,
+        agentId: row.agent_id,
+        seq: row.seq,
+        eventType: row.event_type,
+        stream: row.stream,
+        level: row.level,
+        message: row.message,
+        payload: row.payload ? JSON.parse(row.payload) : null,
+        createdAt: row.created_at,
+      }));
+    },
+
     updateRunActor(runId: string, actorType: string, actorId: string | null): void {
       getDb().prepare(
         "UPDATE runs SET actor_type = ?, actor_id = ?, updated_at = ? WHERE id = ?"