feat(sse): persist bounded resume cursors

src/main/services/sse-resume-cursor-store.test.ts

@@ -0,0 +1,91 @@
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import { tmpdir } from 'node:os'
+import { describe, expect, it } from 'vitest'
+import { SseResumeCursorStore } from './sse-resume-cursor-store'
+
+const cursor = (scopeId: string, lastSequence: number, updatedAt: string, runtimeGeneration = 'g1') => ({
+  scopeId,
+  streamId: 'stream-1',
+  lastSequence,
+  runtimeGeneration,
+  updatedAt
+})
+
+describe('SseResumeCursorStore', () => {
+  it('persists acknowledged cursors atomically and restores them', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-'))
+    const path = join(root, 'nested', 'cursors.json')
+    try {
+      const store = new SseResumeCursorStore(path)
+      expect(await store.save(cursor('thread-1', 4, '2026-07-14T00:00:00.000Z'))).toBe(true)
+      expect(await store.get('thread-1')).toEqual(cursor('thread-1', 4, '2026-07-14T00:00:00.000Z'))
+      const restored = new SseResumeCursorStore(path)
+      expect(await restored.get('thread-1')).toEqual(cursor('thread-1', 4, '2026-07-14T00:00:00.000Z'))
+      expect(JSON.parse(await readFile(path, 'utf8')).version).toBe(1)
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('does not move a cursor backwards within the same runtime generation', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-'))
+    try {
+      const store = new SseResumeCursorStore(join(root, 'cursors.json'))
+      expect(await store.save(cursor('thread-1', 10, '2026-07-14T00:00:01.000Z'))).toBe(true)
+      expect(await store.save(cursor('thread-1', 9, '2026-07-14T00:00:02.000Z'))).toBe(false)
+      expect(await store.get('thread-1')).toEqual(cursor('thread-1', 10, '2026-07-14T00:00:01.000Z'))
+      expect(await store.save(cursor('thread-1', 1, '2026-07-14T00:00:03.000Z', 'g2'))).toBe(true)
+      expect(await store.get('thread-1')).toEqual(cursor('thread-1', 1, '2026-07-14T00:00:03.000Z', 'g2'))
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('serializes concurrent writes and trims oldest scopes', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-'))
+    try {
+      const store = new SseResumeCursorStore(join(root, 'cursors.json'), { maxEntries: 2 })
+      await Promise.all([
+        store.save(cursor('old', 1, '2026-07-14T00:00:00.000Z')),
+        store.save(cursor('newer', 2, '2026-07-14T00:00:02.000Z')),
+        store.save(cursor('newest', 3, '2026-07-14T00:00:03.000Z'))
+      ])
+      expect(await store.get('old')).toBeNull()
+      expect(await store.list()).toHaveLength(2)
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('fails open on malformed data without overwriting the evidence', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-'))
+    const path = join(root, 'cursors.json')
+    try {
+      await writeFile(path, '{not-json', 'utf8')
+      const store = new SseResumeCursorStore(path)
+      expect(await store.get('thread-1')).toBeNull()
+      expect(await readFile(path, 'utf8')).toBe('{not-json')
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('ignores entries whose storage key does not match the cursor scope', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-'))
+    const path = join(root, 'cursors.json')
+    try {
+      await writeFile(path, JSON.stringify({
+        version: 1,
+        cursors: {
+          'wrong-key': cursor('thread-1', 7, '2026-07-14T00:00:00.000Z')
+        }
+      }), 'utf8')
+      const store = new SseResumeCursorStore(path)
+      expect(await store.get('thread-1')).toBeNull()
+      expect(await store.list()).toEqual([])
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+})

src/main/services/sse-resume-cursor-store.ts

@@ -0,0 +1,131 @@
+import { mkdir, readFile } from 'node:fs/promises'
+import { dirname } from 'node:path'
+import { atomicWriteFile } from '../../../kun/src/adapters/file/atomic-write.js'
+import {
+  SSE_RESUME_CURSOR_SCHEMA_VERSION,
+  SseResumeCursorFileSchema,
+  SseResumeCursorSchema,
+  type SseResumeCursor
+} from '../../shared/sse-cursor'
+
+const DEFAULT_MAX_ENTRIES = 256
+
+function sameGeneration(left: SseResumeCursor, right: SseResumeCursor): boolean {
+  return !left.runtimeGeneration || !right.runtimeGeneration || left.runtimeGeneration === right.runtimeGeneration
+}
+
+/**
+ * Small, bounded, single-writer store for renderer resume cursors.
+ *
+ * It deliberately does not delete a malformed file: a corrupt cursor file is
+ * evidence for diagnostics, while a fresh in-memory state lets the app recover
+ * without losing the authoritative thread history.
+ */
+export class SseResumeCursorStore {
+  private readonly maxEntries: number
+  private loaded = false
+  private entries = new Map<string, SseResumeCursor>()
+  private operationTail: Promise<void> = Promise.resolve()
+
+  constructor(
+    private readonly filePath: string,
+    options: { maxEntries?: number } = {}
+  ) {
+    const maxEntries = options.maxEntries ?? DEFAULT_MAX_ENTRIES
+    if (!Number.isSafeInteger(maxEntries) || maxEntries < 1 || maxEntries > 10_000) {
+      throw new Error('maxEntries must be a safe integer between 1 and 10000')
+    }
+    this.maxEntries = maxEntries
+  }
+
+  async get(scopeId: string): Promise<SseResumeCursor | null> {
+    return this.enqueue(async () => {
+      await this.ensureLoaded()
+      const cursor = this.entries.get(scopeId)
+      return cursor ? { ...cursor } : null
+    })
+  }
+
+  async list(): Promise<SseResumeCursor[]> {
+    return this.enqueue(async () => {
+      await this.ensureLoaded()
+      return [...this.entries.values()]
+        .sort((left, right) => right.updatedAt.localeCompare(left.updatedAt))
+        .map((cursor) => ({ ...cursor }))
+    })
+  }
+
+  async save(cursor: SseResumeCursor): Promise<boolean> {
+    return this.enqueue(async () => {
+      await this.ensureLoaded()
+      const normalized = SseResumeCursorSchema.parse(cursor)
+      const previous = this.entries.get(normalized.scopeId)
+      if (
+        previous &&
+        sameGeneration(previous, normalized) &&
+        normalized.lastSequence < previous.lastSequence
+      ) {
+        return false
+      }
+      this.entries.set(normalized.scopeId, normalized)
+      this.trim()
+      await this.persist()
+      return true
+    })
+  }
+
+  async clear(scopeId: string): Promise<boolean> {
+    return this.enqueue(async () => {
+      await this.ensureLoaded()
+      const removed = this.entries.delete(scopeId)
+      if (removed) await this.persist()
+      return removed
+    })
+  }
+
+  private enqueue<T>(task: () => Promise<T>): Promise<T> {
+    const next = this.operationTail.then(task, task)
+    this.operationTail = next.then(() => undefined, () => undefined)
+    return next
+  }
+
+  private async ensureLoaded(): Promise<void> {
+    if (this.loaded) return
+    this.loaded = true
+    let raw: string
+    try {
+      raw = await readFile(this.filePath, 'utf8')
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code === 'ENOENT') return
+      throw error
+    }
+    try {
+      const parsed = SseResumeCursorFileSchema.parse(JSON.parse(raw))
+      this.entries = new Map(
+        Object.entries(parsed.cursors)
+          .filter(([scopeId, cursor]) => scopeId === cursor.scopeId)
+      )
+      this.trim()
+    } catch {
+      // Keep the malformed file untouched and start from a recoverable cursor set.
+      this.entries = new Map()
+    }
+  }
+
+  private trim(): void {
+    if (this.entries.size <= this.maxEntries) return
+    const sorted = [...this.entries.values()]
+      .sort((left, right) => right.updatedAt.localeCompare(left.updatedAt))
+      .slice(0, this.maxEntries)
+    this.entries = new Map(sorted.map((cursor) => [cursor.scopeId, cursor]))
+  }
+
+  private async persist(): Promise<void> {
+    await mkdir(dirname(this.filePath), { recursive: true })
+    const cursors = Object.fromEntries(this.entries)
+    await atomicWriteFile(this.filePath, JSON.stringify({
+      version: SSE_RESUME_CURSOR_SCHEMA_VERSION,
+      cursors
+    }, null, 2))
+  }
+}

src/shared/sse-cursor.ts

@@ -0,0 +1,21 @@
+import { z } from 'zod'
+
+/** A renderer-owned acknowledgement cursor persisted across process restarts. */
+export const SseResumeCursorSchema = z.object({
+  scopeId: z.string().trim().min(1).max(128),
+  streamId: z.string().trim().min(1).max(128),
+  lastSequence: z.number().int().nonnegative().safe(),
+  runtimeGeneration: z.string().trim().min(1).max(128).optional(),
+  updatedAt: z.string().datetime({ offset: true })
+}).strict()
+
+export type SseResumeCursor = z.infer<typeof SseResumeCursorSchema>
+
+export const SSE_RESUME_CURSOR_SCHEMA_VERSION = 1 as const
+
+export const SseResumeCursorFileSchema = z.object({
+  version: z.literal(SSE_RESUME_CURSOR_SCHEMA_VERSION),
+  cursors: z.record(z.string().min(1).max(128), SseResumeCursorSchema)
+}).strict()
+
+export type SseResumeCursorFile = z.infer<typeof SseResumeCursorFileSchema>