diff --git a/packages/contrail/src/core/persistent.ts b/packages/contrail/src/core/persistent.ts
index 5eb48a2..4dade65 100644
--- a/packages/contrail/src/core/persistent.ts
+++ b/packages/contrail/src/core/persistent.ts
@@ -120,21 +120,15 @@ async function streamAndFlush(
       });
 
   const buffer: IngestEvent[] = [];
-  let flushDue = false;
+  // Guards against overlap between the periodic timer flush and a main-loop
+  // batchSize-driven flush. The main loop only ever awaits flush() sequentially,
+  // but the setInterval callback is a second entry point on another tick.
   let flushing = false;
-  let flushTimer: ReturnType<typeof setTimeout> | null = null;
-
-  const resetFlushTimer = () => {
-    if (flushTimer) clearTimeout(flushTimer);
-    flushTimer = setTimeout(() => { flushDue = true; }, flushIntervalMs);
-  };
 
   const flush = async () => {
     if (buffer.length === 0 || flushing) return;
     flushing = true;
     const batch = buffer.splice(0);
-    flushDue = false;
-    resetFlushTimer();
 
     try {
       await applyEvents(db, batch, config, { pubsub: opts.pubsub });
@@ -142,7 +136,6 @@ async function streamAndFlush(
       const lastTimeUs = Math.max(...batch.map((e) => e.time_us));
       await saveCursor(db, lastTimeUs);
 
-      // Identity refresh
       const uniqueDids = [...new Set(batch.map((e) => e.did))];
       if (uniqueDids.length > 0) {
         try {
@@ -152,7 +145,6 @@ async function streamAndFlush(
         }
       }
 
-      // Feed pruning
       if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) {
         const maxItems = Math.max(
           ...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS)
@@ -168,38 +160,37 @@ async function streamAndFlush(
     }
   };
 
-  // Handle abort
+  // Periodic flush decoupled from the main loop. Runs even when Jetstream is
+  // idle, which is the whole point — without it, buffered events strand until
+  // the next event or abort. Errors log and retry next interval rather than
+  // propagate, so transient DB hiccups don't force a reconnect.
+  const flushTimer = setInterval(() => {
+    flush().catch((err) => log.error(`Timer flush failed: ${err}`));
+  }, flushIntervalMs);
+
   const onAbort = () => {
-    if (flushTimer) clearTimeout(flushTimer);
+    clearInterval(flushTimer);
   };
   signal?.addEventListener("abort", onAbort, { once: true });
 
-  resetFlushTimer();
-
-  // Use manual iterator so we can race next() against abort signal
   const iterator = subscription[Symbol.asyncIterator]();
 
   try {
-    while (true) {
-      if (signal?.aborted) break;
+    while (!signal?.aborted) {
+      // Per-iteration abort race so the handler can be removed synchronously
+      // after the race settles — otherwise addEventListener calls accumulate on
+      // the signal across the streamAndFlush lifetime.
+      let abortHandler!: () => void;
+      const abortPromise = new Promise<IteratorResult<any>>((resolve) => {
+        abortHandler = () => resolve({ value: undefined, done: true });
+        signal?.addEventListener("abort", abortHandler, { once: true });
+      });
 
-      // Race the next event against abort signal
       let result: IteratorResult<any>;
-      if (signal) {
-        const nextPromise = iterator.next();
-        if (signal.aborted) {
-          result = { value: undefined, done: true };
-        } else {
-          let abortHandler: () => void;
-          const abortPromise = new Promise<IteratorResult<any>>((resolve) => {
-            abortHandler = () => resolve({ value: undefined, done: true });
-            signal.addEventListener("abort", abortHandler, { once: true });
-          });
-          result = await Promise.race([nextPromise, abortPromise]);
-          signal.removeEventListener("abort", abortHandler!);
-        }
-      } else {
-        result = await iterator.next();
+      try {
+        result = await Promise.race([iterator.next(), abortPromise]);
+      } finally {
+        signal?.removeEventListener("abort", abortHandler);
       }
 
       if (result.done) break;
@@ -232,15 +223,14 @@ async function streamAndFlush(
         }
       }
 
-      if (buffer.length >= batchSize || flushDue) {
+      if (buffer.length >= batchSize) {
         await flush();
       }
     }
   } finally {
-    // Clean up iterator
+    clearInterval(flushTimer);
+    signal?.removeEventListener("abort", onAbort);
     await iterator.return?.({ value: undefined, done: true });
-    // Final flush on exit
     await flush();
-    signal?.removeEventListener("abort", onAbort);
   }
 }
diff --git a/packages/contrail/tests/persistent.test.ts b/packages/contrail/tests/persistent.test.ts
index 531304e..5fba923 100644
--- a/packages/contrail/tests/persistent.test.ts
+++ b/packages/contrail/tests/persistent.test.ts
@@ -154,6 +154,53 @@ describe("runPersistent", () => {
     expect(cursor).toBe(3004);
   });
 
+  it("flushes buffered events on timer while subscription is still live", async () => {
+    // Regression test for the idle-stream flush bug. Forces exactly one code
+    // path — timer-driven flush — by:
+    //   - batchSize=100 with only 3 events: batchSize flush can never fire
+    //   - assert BEFORE controller.abort(): the finally-block's final flush
+    //     can't contribute, so records in the DB prove the periodic timer ran
+    // The mock subscription yields 3 events then hangs forever, mimicking a
+    // Jetstream connection that goes quiet. The previous implementation only
+    // checked the flush condition when a new event arrived, so those 3 events
+    // would sit in memory until the next event or shutdown — which in prod
+    // surfaces as "events published but never indexed."
+    const events = Array.from({ length: 3 }, (_, i) => ({
+      kind: "commit" as const,
+      did: `did:plc:idle${i}`,
+      time_us: 5000 + i,
+      commit: {
+        collection: "community.lexicon.calendar.event",
+        operation: "create",
+        rkey: `idle${i}`,
+        cid: `cid${i}`,
+        record: { name: `Idle ${i}`, startsAt: "2026-04-01T10:00:00Z", mode: "online" },
+      },
+    }));
+
+    const controller = new AbortController();
+    const promise = runPersistent(db, TEST_CONFIG, {
+      batchSize: 100,
+      flushIntervalMs: 100,
+      signal: controller.signal,
+      createSubscription: () => mockSubscription(events) as any,
+    });
+
+    // Wait well past the flush interval — no further events will arrive.
+    await new Promise((r) => setTimeout(r, 500));
+
+    // Assert BEFORE abort: the timer must have driven the flush on its own.
+    const mid = await queryRecords(db, TEST_CONFIG, {
+      collection: "community.lexicon.calendar.event",
+      limit: 100,
+    });
+    const idleUris = mid.records.map((r) => r.uri).filter((u) => u.includes("/idle"));
+    expect(idleUris.length, "timer flush did not run while subscription was idle").toBe(3);
+
+    controller.abort();
+    await promise;
+  });
+
   it("skips non-commit events", async () => {
     const events = [
       { kind: "identity" as const, did: "did:plc:someone", time_us: 4000 },