release: v0.3.5

build/installer.nsh

@@ -176,6 +176,12 @@ Var /GLOBAL KunInstallerStopDiagnosticPath
     Quit
   ${endif}
 
+  ${if} ${isUpdated}
+    # electron-builder keeps existing shortcuts during --updated installs, but
+    # a scope/directory migration may already have removed the old link.
+    !insertmacro addDesktopLink "false"
+  ${endif}
+
   ${if} $KunInstallerInPlaceUpdate == 1
     !insertmacro kunRunMigrationHelper CleanupInPlaceLeftovers
     ${if} $KunInstallerHelperExitCode != 0

kun/src/adapters/file/atomic-write.ts

@@ -3,6 +3,7 @@ import { mkdir, rename, rm, writeFile } from 'node:fs/promises'
 import { dirname } from 'node:path'
 
 export type AtomicWriteFileOptions = {
+  allowDirectWriteFallback?: boolean
   renameRetry?: {
     attempts?: number
     baseDelayMs?: number
@@ -23,9 +24,9 @@ export async function atomicWriteFile(
   try {
     await writeFile(tmp, contents, { encoding: 'utf-8', mode: 0o600 })
     try {
-      await renameWithRetry(tmp, path, options.renameRetry)
+      await renameFileWithRetry(tmp, path, options.renameRetry)
     } catch (error) {
-      if (!shouldFallbackToDirectWrite(error)) {
+      if (options.allowDirectWriteFallback === false || !shouldFallbackToDirectWrite(error)) {
         throw error
       }
       await writeFile(path, contents, { encoding: 'utf-8', mode: 0o600 })
@@ -61,10 +62,10 @@ function describeAtomicWriteError(path: string, error: unknown): unknown {
   return prefixed
 }
 
-async function renameWithRetry(
+export async function renameFileWithRetry(
   from: string,
   to: string,
-  options: NonNullable<AtomicWriteFileOptions['renameRetry']> | undefined
+  options?: NonNullable<AtomicWriteFileOptions['renameRetry']>
 ): Promise<void> {
   const attempts = Math.max(1, Math.floor(options?.attempts ?? DEFAULT_RENAME_RETRY_ATTEMPTS))
   const baseDelayMs = Math.max(0, Math.floor(options?.baseDelayMs ?? DEFAULT_RENAME_RETRY_BASE_DELAY_MS))

kun/src/adapters/file/file-session-jsonl.ts

@@ -1,11 +1,12 @@
 import { randomUUID } from 'node:crypto'
 import { createReadStream, createWriteStream } from 'node:fs'
-import { rename, rm, type FileHandle } from 'node:fs/promises'
+import { rm, type FileHandle } from 'node:fs/promises'
 import type { RuntimeEvent } from '../../contracts/events.js'
 import { isPublicTurnItem, type TurnItem } from '../../contracts/items.js'
 import type { ItemHistoryPage, ItemHistoryPageOptions } from '../../ports/session-store.js'
 import { buildPublicItemHistoryPage, timelineSafeItem } from '../../services/item-history-page.js'
 import { yieldToEventLoop } from '../hybrid/hybrid-thread-support.js'
+import { renameFileWithRetry } from './atomic-write.js'
 
 const MS_PER_DAY = 86_400_000
 const DEFAULT_ITEM_HISTORY_MAX_RECORD_BYTES = 16 * 1024 * 1024
@@ -34,7 +35,12 @@ export function compactUsageEvents(
  */
 export async function compactUsageEventsJsonlFile(
   path: string,
-  options: { nowIso: string; retentionDays: number; maxRecordBytes: number }
+  options: {
+    nowIso: string
+    retentionDays: number
+    maxRecordBytes: number
+    commitReplacement?: (replace: () => Promise<void>) => Promise<boolean>
+  }
 ): Promise<boolean> {
   const cutoffMs = Date.parse(options.nowIso) - options.retentionDays * MS_PER_DAY
   if (!Number.isFinite(cutoffMs)) return false
@@ -57,11 +63,17 @@ export async function compactUsageEventsJsonlFile(
       maxRecordBytes: options.maxRecordBytes,
       keepLine: (event, index) => event.kind !== 'usage' || keepUsageIndexes.has(index)
     })
-    await rename(tmp, path)
+    const replace = async (): Promise<void> => renameFileWithRetry(tmp, path)
+    if (options.commitReplacement) {
+      const committed = await options.commitReplacement(replace)
+      return committed
+    }
+    await replace()
     return true
-  } catch (error) {
+  } finally {
+    // A stale snapshot deliberately declines replacement. Always remove its
+    // prepared rewrite; after a successful rename this is a harmless no-op.
     await rm(tmp, { force: true }).catch(() => undefined)
-    throw error
   }
 }
 
@@ -225,7 +237,12 @@ export async function readLatestItemsFromJsonl(
     maxRecordBytes?: number
     rejectMalformed?: boolean
   } = {}
-): Promise<{ items: TurnItem[]; rawCount: number; malformedCount: number }> {
+): Promise<{
+  items: TurnItem[]
+  rawCount: number
+  malformedCount: number
+  incompleteTrailingRecord: boolean
+}> {
   const maxRecordBytes = Math.max(
     1,
     Math.floor(options.maxRecordBytes ?? DEFAULT_ITEM_HISTORY_MAX_RECORD_BYTES)
@@ -235,9 +252,10 @@ export async function readLatestItemsFromJsonl(
   let remainder = ''
   let rawCount = 0
   let malformedCount = 0
+  let incompleteTrailingRecord = false
   let linesSinceYield = 0
 
-  const acceptLine = async (line: string): Promise<void> => {
+  const acceptLine = async (line: string, trailing = false): Promise<void> => {
     if (!line.trim()) return
     if (Buffer.byteLength(line, 'utf-8') > maxRecordBytes) {
       throw new Error(`item history record exceeds ${maxRecordBytes} bytes`)
@@ -252,7 +270,8 @@ export async function readLatestItemsFromJsonl(
       if (!latestById.has(item.id)) firstSeenIds.push(item.id)
       latestById.set(item.id, item)
     } catch {
-      malformedCount += 1
+      if (trailing) incompleteTrailingRecord = true
+      else malformedCount += 1
     }
     linesSinceYield += 1
     if (linesSinceYield >= YIELD_EVERY_LINES) {
@@ -278,18 +297,20 @@ export async function readLatestItemsFromJsonl(
         throw new Error(`item history record exceeds ${maxRecordBytes} bytes`)
       }
     }
-    await acceptLine(remainder)
+    await acceptLine(remainder, true)
   } catch (error) {
     if ((error as { code?: string }).code !== 'ENOENT') throw error
   }
 
-  if (options.rejectMalformed && malformedCount > 0) {
-    throw new Error(`item history contains ${malformedCount} malformed record(s)`)
+  const rejectedRecords = malformedCount + (incompleteTrailingRecord ? 1 : 0)
+  if (options.rejectMalformed && rejectedRecords > 0) {
+    throw new Error(`item history contains ${rejectedRecords} malformed record(s)`)
   }
   return {
     items: firstSeenIds.map((id) => latestById.get(id)!),
     rawCount,
-    malformedCount
+    malformedCount,
+    incompleteTrailingRecord
   }
 }
 

kun/src/adapters/file/file-session-store.ordering.test.ts

@@ -3,7 +3,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, describe, expect, it } from 'vitest'
 import { makeAssistantTextItem, makeToolCallItem, makeToolResultItem, makeUserItem } from '../../domain/item.js'
-import { FileSessionStore } from './file-session-store.js'
+import { FileSessionStore, readLatestItemsFromJsonl } from './file-session-store.js'
 
 const roots: string[] = []
 
@@ -143,6 +143,32 @@ describe('FileSessionStore item ordering', () => {
     expect(await readFile(path, 'utf-8')).toBe(before)
   })
 
+  it('distinguishes an unterminated trailing write from a malformed completed row', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'kun-session-incomplete-tail-'))
+    roots.push(root)
+    const path = join(root, 'messages.jsonl')
+    const item = makeUserItem({
+      id: 'user_1',
+      threadId: 'thread_incomplete_tail',
+      turnId: 'turn_1',
+      text: 'valid'
+    })
+    await appendFile(path, `${JSON.stringify(item)}\n{"id":`)
+
+    await expect(readLatestItemsFromJsonl(path)).resolves.toMatchObject({
+      items: [expect.objectContaining({ id: 'user_1' })],
+      rawCount: 1,
+      malformedCount: 0,
+      incompleteTrailingRecord: true
+    })
+
+    await appendFile(path, '}\n{broken-json\n')
+    await expect(readLatestItemsFromJsonl(path)).resolves.toMatchObject({
+      malformedCount: 2,
+      incompleteTrailingRecord: false
+    })
+  })
+
   it('does not retain a Session item array that exceeds its byte admission limit', async () => {
     const root = await mkdtemp(join(tmpdir(), 'kun-session-cache-budget-'))
     roots.push(root)

kun/src/adapters/file/file-session-store.search.test.ts

@@ -0,0 +1,134 @@
+import { mkdtemp, rm, stat } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import { makeAssistantTextItem, makeToolResultItem, makeUserItem } from '../../domain/item.js'
+import { FileSessionStore } from './file-session-store.js'
+
+const roots: string[] = []
+
+afterEach(async () => {
+  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
+})
+
+async function newStore(options: { itemHistoryCompactionMinBytes?: number } = {}): Promise<{
+  store: FileSessionStore
+  root: string
+  messagesPath: (threadId: string) => string
+}> {
+  const root = await mkdtemp(join(tmpdir(), 'kun-session-search-'))
+  roots.push(root)
+  return {
+    store: new FileSessionStore({ dataDir: root, ...options }),
+    root,
+    messagesPath: (threadId) => join(root, 'threads', threadId, 'messages.jsonl')
+  }
+}
+
+describe('FileSessionStore.searchItemText', () => {
+  it('finds user and assistant text and returns the matching item text', async () => {
+    const { store } = await newStore()
+    const threadId = 'thread_search'
+    await store.appendItem(threadId, makeUserItem({
+      id: 'i1', turnId: 't1', threadId, text: 'Please rework the checkout flow.'
+    }))
+    await store.appendItem(threadId, makeAssistantTextItem({
+      id: 'i2', turnId: 't1', threadId, text: 'Rewriting the billing module now.'
+    }))
+
+    await expect(store.searchItemText(threadId, 'checkout'))
+      .resolves.toBe('Please rework the checkout flow.')
+    await expect(store.searchItemText(threadId, 'BILLING'))
+      .resolves.toBe('Rewriting the billing module now.')
+    await expect(store.searchItemText(threadId, 'absent')).resolves.toBeNull()
+  })
+
+  it('ignores tool payloads so search never surfaces raw tool output', async () => {
+    const { store } = await newStore()
+    const threadId = 'thread_tools'
+    await store.appendItem(threadId, makeToolResultItem({
+      id: 'i1', turnId: 't1', threadId, callId: 'c1', toolName: 'read', output: { text: 'secret-token' }
+    }))
+    await expect(store.searchItemText(threadId, 'secret-token')).resolves.toBeNull()
+  })
+
+  it('does not match on record metadata that only looks like a hit', async () => {
+    const { store } = await newStore()
+    const threadId = 'thread_meta'
+    await store.appendItem(threadId, makeUserItem({
+      id: 'i1', turnId: 't1', threadId, text: 'unrelated body'
+    }))
+    // 'assistant_text' and the thread id appear in the raw JSON of every
+    // record; only real item text may produce a match.
+    await expect(store.searchItemText(threadId, 'user_message')).resolves.toBeNull()
+    await expect(store.searchItemText(threadId, 'thread_meta')).resolves.toBeNull()
+  })
+
+  it('never schedules a rewrite the way loadItems does', async () => {
+    // A one-byte threshold makes every log "oversized".
+    const { store, messagesPath } = await newStore({ itemHistoryCompactionMinBytes: 1 })
+    const threadId = 'thread_big'
+    await store.appendItem(threadId, makeUserItem({
+      id: 'i1', turnId: 't1', threadId, text: 'first checkout note'
+    }))
+    await store.appendItem(threadId, makeUserItem({
+      id: 'i1', turnId: 't1', threadId, text: 'second checkout note'
+    }))
+    await store.resetMemory()
+    const path = messagesPath(threadId)
+    const before = (await stat(path)).size
+
+    // Searching leaves the log untouched, and queues no deferred rewrite.
+    await expect(store.searchItemText(threadId, 'checkout')).resolves.toContain('checkout')
+    await store.flushScheduledCompaction(threadId)
+    expect((await stat(path)).size).toBe(before)
+
+    // The blocking path schedules the rewrite; this documents the contrast.
+    await store.loadItems(threadId)
+    await store.flushScheduledCompaction(threadId)
+    expect((await stat(path)).size).toBeLessThan(before)
+  })
+
+  it('reads the tail when a log exceeds the scan window', async () => {
+    const { store } = await newStore()
+    const threadId = 'thread_tail'
+    const filler = 'x'.repeat(4_000)
+    await store.appendItem(threadId, makeUserItem({
+      id: 'oldest', turnId: 't0', threadId, text: 'oldest-marker ' + filler
+    }))
+    for (let index = 0; index < 40; index += 1) {
+      await store.appendItem(threadId, makeUserItem({
+        id: 'mid_' + index, turnId: 't1', threadId, text: 'filler ' + filler
+      }))
+    }
+    await store.appendItem(threadId, makeUserItem({
+      id: 'recent', turnId: 't2', threadId, text: 'recent-marker at the end'
+    }))
+    await store.resetMemory()
+
+    await expect(store.searchItemText(threadId, 'recent-marker', { maxBytes: 8_000 }))
+      .resolves.toBe('recent-marker at the end')
+    // Content older than the tail window is outside the bound by design.
+    await expect(store.searchItemText(threadId, 'oldest-marker', { maxBytes: 8_000 }))
+      .resolves.toBeNull()
+    // Widening the window brings it back into range.
+    await expect(store.searchItemText(threadId, 'oldest-marker', { maxBytes: 4_000_000 }))
+      .resolves.toContain('oldest-marker')
+  })
+
+  it('returns null for unsafe thread ids and empty queries', async () => {
+    const { store } = await newStore()
+    await expect(store.searchItemText('../escape', 'anything')).resolves.toBeNull()
+    await expect(store.searchItemText('thread_ok', '')).resolves.toBeNull()
+  })
+
+  it('does not start or return a scan after its deadline', async () => {
+    const { store } = await newStore()
+    const threadId = 'thread_expired'
+    await store.appendItem(threadId, makeUserItem({
+      id: 'i1', turnId: 't1', threadId, text: 'checkout after deadline'
+    }))
+    await expect(store.searchItemText(threadId, 'checkout', { deadlineAtMs: Date.now() - 1 }))
+      .resolves.toBeNull()
+  })
+})