feat(server): self-host retention policy with in-process sweep

AGENTS.md

@@ -90,6 +90,7 @@ The desktop app authenticates via the **OAuth 2.0 device authorization grant** (
 One codebase, two deployment modes gated by `KINORA_CLOUD` (env). Self-host (`false`) unlocks every feature; cloud (`true`) enables **Polar** billing.
 
 - **Billing** (`src/billing/`): `polar.ts` (Polar SDK + better-auth plugin), `entitlements.ts` / `usage.ts` (plan limits), `retention.ts` (per-plan run-retention windows, swept by `purge-expired-runs`).
+- **Retention** (`src/billing/retention.ts`): cloud derives windows from the plan tier and is swept by an external cron calling `purge-expired-runs`. Self-host instead reads `retentionPolicy` from env (`KINORA_ARTIFACT_RETENTION_DAYS` drops blobs but keeps runs; `KINORA_RETENTION_DAYS` / `KINORA_KEEP_LAST_RUNS` delete runs), and `src/index.ts` runs a daily in-process sweep gated on that policy being non-null, so cloud never double-sweeps.
 - **Alerts** (`src/alerts/`): per-project notifications on new failures / regressions. Channels are `slack.ts`, `email.ts` (nodemailer/SMTP), `webhook.ts`, dispatched by `notify.ts` with an every-run / on-failure / on-regression policy (`core.ts`).
 - **Feedback** (`src/feedback/`, `feedback` tRPC router): in-app "Send feedback" posts bug/feature reports to the private cloud task tracker. Cloud-only: `resolveFeedbackTracker` in `env.ts` returns null unless `KINORA_CLOUD=true` and all `FEEDBACK_TRACKER_*` vars are set; `config.get.feedbackEnabled` gates the web UI.
 - Email (password reset, verification, invitations, alerts) needs `SMTP_*`; social login needs `GOOGLE_*` / `GITHUB_*`. All optional - empty disables the flow.

packages/server/.env.example

@@ -27,6 +27,15 @@ S3_BUCKET=
 S3_ACCESS_KEY_ID=
 S3_SECRET_ACCESS_KEY=
 
+# Retention sweep (self-host only; cloud retention follows the plan tier). 0 = keep forever.
+# The server runs a daily in-process sweep as soon as one of them is non-zero.
+# KINORA_ARTIFACT_RETENTION_DAYS is the space-efficient one: it deletes stored trace.zip files
+# (screenshots and videos ride inside them) while keeping the runs, so history and trends stay.
+KINORA_ARTIFACT_RETENTION_DAYS=0
+# These two delete whole runs, history included. KINORA_KEEP_LAST_RUNS counts per project.
+KINORA_RETENTION_DAYS=0
+KINORA_KEEP_LAST_RUNS=0
+
 # Google
 GOOGLE_CLIENT_ID=
 GOOGLE_CLIENT_SECRET=

packages/server/scripts/purge-expired-runs.ts

@@ -3,8 +3,8 @@ import { purgeExpiredRuns } from '../src/billing/retention'
 import { logger } from '../src/lib/logger'
 
 purgeExpiredRuns(new Date())
-  .then(({ deleted }) => {
-    logger.info({ deleted }, 'purge-expired-runs complete')
+  .then(({ deleted, artifacts }) => {
+    logger.info({ deleted, artifacts }, 'purge-expired-runs complete')
     process.exit(0)
   })
   .catch((error) => {

packages/server/src/billing/retention.ts

@@ -1,7 +1,7 @@
-import { and, eq, inArray, lt, notInArray } from 'drizzle-orm'
+import { and, desc, eq, inArray, lt, notInArray } from 'drizzle-orm'
 import { db } from '../db'
 import { artifact, project, run, subscription } from '../db/schemas/index'
-import { cloud } from '../lib/env'
+import { cloud, retentionPolicy } from '../lib/env'
 import { logger } from '../lib/logger'
 import { storage } from '../lib/storage'
 import { retentionDaysFor } from './entitlements'
@@ -12,12 +12,24 @@ const DAY_MS = 24 * 60 * 60 * 1000
 interface Scope {
   includeOrgs?: string[]
   excludeOrgs?: string[]
+  projectIds?: string[]
 }
 
 function cutoff(now: Date, days: number): Date {
   return new Date(now.getTime() - days * DAY_MS)
 }
 
+async function deleteBlobs(keys: string[]): Promise<void> {
+  for (const key of keys) {
+    try {
+      await storage.delete(key)
+    }
+    catch (error) {
+      logger.error({ error, key }, 'retention: blob delete failed')
+    }
+  }
+}
+
 export async function purgeScope(before: Date, scope: Scope): Promise<number> {
   // A tier with no users on it: nothing to do.
   if (scope.includeOrgs && scope.includeOrgs.length === 0)
@@ -31,6 +43,8 @@ export async function purgeScope(before: Date, scope: Scope): Promise<number> {
       conds.push(inArray(project.organizationId, scope.includeOrgs))
     if (scope.excludeOrgs && scope.excludeOrgs.length > 0)
       conds.push(notInArray(project.organizationId, scope.excludeOrgs))
+    if (scope.projectIds)
+      conds.push(inArray(run.projectId, scope.projectIds))
 
     const batch = await db
       .select({ id: run.id })
@@ -50,14 +64,7 @@ export async function purgeScope(before: Date, scope: Scope): Promise<number> {
       .select({ key: artifact.storageKey })
       .from(artifact)
       .where(inArray(artifact.runId, ids))
-    for (const blob of blobs) {
-      try {
-        await storage.delete(blob.key)
-      }
-      catch (error) {
-        logger.error({ error, key: blob.key }, 'retention: blob delete failed')
-      }
-    }
+    await deleteBlobs(blobs.map(b => b.key))
 
     await db.delete(run).where(inArray(run.id, ids))
     total += fetched
@@ -66,10 +73,76 @@ export async function purgeScope(before: Date, scope: Scope): Promise<number> {
   return total
 }
 
+// Drop the blobs of older runs but keep the rows: pass rates, trends and flaky history survive,
+// the attachment just loses its URL at read time.
+export async function purgeArtifactsBefore(before: Date): Promise<number> {
+  let total = 0
+  let fetched = BATCH
+  while (fetched === BATCH) {
+    const batch = await db
+      .select({ id: artifact.id, key: artifact.storageKey })
+      .from(artifact)
+      .innerJoin(run, eq(artifact.runId, run.id))
+      .where(lt(run.startedAt, before))
+      .limit(BATCH)
+
+    fetched = batch.length
+    if (fetched === 0)
+      break
+
+    await deleteBlobs(batch.map(b => b.key))
+    await db.delete(artifact).where(inArray(artifact.id, batch.map(b => b.id)))
+    total += fetched
+  }
+
+  return total
+}
+
+// Keep the newest `keep` runs of every project. Runs sharing the boundary timestamp are all kept.
+export async function purgeBeyondLastRuns(keep: number): Promise<number> {
+  if (keep <= 0)
+    return 0
+
+  const projects = await db.select({ id: project.id }).from(project)
+  let deleted = 0
+  for (const p of projects) {
+    const [boundary] = await db
+      .select({ startedAt: run.startedAt })
+      .from(run)
+      .where(eq(run.projectId, p.id))
+      .orderBy(desc(run.startedAt))
+      .offset(keep - 1)
+      .limit(1)
+    if (!boundary)
+      continue
+    deleted += await purgeScope(boundary.startedAt, { projectIds: [p.id] })
+  }
+
+  return deleted
+}
+
+export interface PurgeResult {
+  deleted: number
+  artifacts: number
+}
+
+async function purgeSelfHost(now: Date): Promise<PurgeResult> {
+  if (!retentionPolicy)
+    return { deleted: 0, artifacts: 0 } // unconfigured self-host keeps everything
+
+  const { runDays, keepLastRuns, artifactDays } = retentionPolicy
+  let deleted = 0
+  if (runDays > 0)
+    deleted += await purgeScope(cutoff(now, runDays), {})
+  deleted += await purgeBeyondLastRuns(keepLastRuns)
+  const artifacts = artifactDays > 0 ? await purgeArtifactsBefore(cutoff(now, artifactDays)) : 0
+  return { deleted, artifacts }
+}
+
 // Delete runs (cascading tests + artifacts + their blobs) past each tier's retention window.
-export async function purgeExpiredRuns(now: Date): Promise<{ deleted: number }> {
+export async function purgeExpiredRuns(now: Date): Promise<PurgeResult> {
   if (!cloud)
-    return { deleted: 0 } // self-host keeps everything
+    return purgeSelfHost(now)
 
   const subs = await db
     .select({ organizationId: subscription.organizationId, tier: subscription.tier })
@@ -87,5 +160,5 @@ export async function purgeExpiredRuns(now: Date): Promise<{ deleted: number }>
   deleted += await purgeScope(cutoff(now, retentionDaysFor('team')), { includeOrgs: teamOrgs })
   deleted += await purgeScope(cutoff(now, retentionDaysFor('pro')), { includeOrgs: proOrgs })
   // enterprise retention is unlimited: never purged.
-  return { deleted }
+  return { deleted, artifacts: 0 }
 }

packages/server/src/index.ts

@@ -3,8 +3,9 @@ import './instrument'
 import { serve } from '@hono/node-server'
 import process from 'node:process'
 import { app } from './app'
+import { purgeExpiredRuns } from './billing/retention'
 import { db } from './db'
-import { demo, env } from './lib/env'
+import { demo, env, retentionPolicy } from './lib/env'
 import { logger } from './lib/logger'
 
 // Log stray rejections instead of letting one crash the whole server; uncaught exceptions leave the
@@ -19,6 +20,17 @@ const server = serve({ fetch: app.fetch, port: env.PORT }, (info) => {
   logger.info(`${demo ? '[DEMO] ' : ''}kinora server running on port ${info.port}`)
 })
 
+// Self-host ships no scheduler, so sweep in-process. Cloud leaves retentionPolicy null and
+// keeps sweeping from its own cron (safe with several replicas).
+if (retentionPolicy) {
+  const sweep = (): void => void purgeExpiredRuns(new Date())
+    .then(result => logger.info(result, 'retention sweep complete'))
+    .catch(error => logger.error({ error }, 'retention sweep failed'))
+
+  sweep()
+  setInterval(sweep, 24 * 60 * 60 * 1000).unref()
+}
+
 let shuttingDown = false
 async function shutdown(signal: string): Promise<void> {
   if (shuttingDown)

packages/server/src/lib/env.ts

@@ -31,6 +31,10 @@ const envSchema = z.object({
   POLAR_PRODUCT_PRO_ID: z.string().optional(),
   // Ingest requests per minute per client IP (DoS backstop; sharded CI spreads across IPs). Raise for pathological suites.
   INGEST_RATE_LIMIT: z.coerce.number().int().positive().default(600),
+  // Self-host retention. 0 = keep forever; ignored in cloud, where the plan tier drives it.
+  KINORA_RETENTION_DAYS: z.coerce.number().int().nonnegative().default(0),
+  KINORA_KEEP_LAST_RUNS: z.coerce.number().int().nonnegative().default(0),
+  KINORA_ARTIFACT_RETENTION_DAYS: z.coerce.number().int().nonnegative().default(0),
   STORAGE_DIR: z.string().default('.data/artifacts'),
   S3_ENDPOINT: z.string().optional(),
   S3_REGION: z.string().optional(),
@@ -90,6 +94,27 @@ export const cloud = resolveCloud()
 
 export const demo = env.KINORA_DEMO
 
+export interface RetentionPolicy {
+  runDays: number
+  keepLastRuns: number
+  artifactDays: number
+}
+
+// null = nothing to sweep, which also gates the in-process sweeper (cloud sweeps via its own cron).
+function resolveRetention(): RetentionPolicy | null {
+  if (env.KINORA_CLOUD)
+    return null
+
+  const policy = {
+    runDays: env.KINORA_RETENTION_DAYS,
+    keepLastRuns: env.KINORA_KEEP_LAST_RUNS,
+    artifactDays: env.KINORA_ARTIFACT_RETENTION_DAYS,
+  }
+  return policy.runDays || policy.keepLastRuns || policy.artifactDays ? policy : null
+}
+
+export const retentionPolicy = resolveRetention()
+
 // Social login is enabled per provider only when both its id and secret are set.
 export const googleOauthEnabled = Boolean(env.GOOGLE_CLIENT_ID && env.GOOGLE_CLIENT_SECRET)
 export const githubOauthEnabled = Boolean(env.GITHUB_CLIENT_ID && env.GITHUB_CLIENT_SECRET)

packages/server/test/retention.test.ts

@@ -4,7 +4,7 @@ import { existsSync } from 'node:fs'
 import { resolve } from 'node:path'
 import { eq } from 'drizzle-orm'
 import { beforeEach, describe, expect, it } from 'vitest'
-import { purgeExpiredRuns, purgeScope } from '../src/billing/retention'
+import { purgeArtifactsBefore, purgeBeyondLastRuns, purgeExpiredRuns, purgeScope } from '../src/billing/retention'
 import { db } from '../src/db'
 import { artifact, project, run } from '../src/db/schemas/index'
 import { env } from '../src/lib/env'
@@ -15,9 +15,13 @@ const DAY = 24 * 60 * 60 * 1000
 
 beforeEach(resetDb)
 
-async function seedRun(userId: string, startedAt: Date): Promise<string> {
+async function seedProject(userId: string): Promise<string> {
   const projectId = randomUUID()
   await db.insert(project).values({ id: projectId, organizationId: await ownedOrgId(userId), slug: `s-${projectId}`, name: 'p' })
+  return projectId
+}
+
+async function seedRunIn(projectId: string, startedAt: Date): Promise<string> {
   const runId = randomUUID()
   await db.insert(run).values({
     id: runId,
@@ -29,6 +33,18 @@ async function seedRun(userId: string, startedAt: Date): Promise<string> {
   return runId
 }
 
+async function seedRun(userId: string, startedAt: Date): Promise<string> {
+  return seedRunIn(await seedProject(userId), startedAt)
+}
+
+async function seedArtifact(runId: string, name: string): Promise<string> {
+  const { projectId } = (await db.query.run.findFirst({ where: eq(run.id, runId) }))!
+  const key = `${projectId}/${runId}/${name}`
+  await storage.put(key, Buffer.from('zip'))
+  await db.insert(artifact).values({ id: randomUUID(), projectId, runId, name, contentType: 'application/zip', storageKey: key, size: 3 })
+  return key
+}
+
 function exists(runId: string) {
   return db.query.run.findFirst({ where: eq(run.id, runId), columns: { id: true } })
 }
@@ -74,11 +90,7 @@ describe('purgeScope', () => {
   it('deletes the artifact blobs of purged runs', async () => {
     const u = await createUser()
     const runId = await seedRun(u.id, new Date(Date.now() - 100 * DAY))
-    const projectId = (await db.query.run.findFirst({ where: eq(run.id, runId) }))!.projectId
-    const key = `${projectId}/${runId}/trace.zip`
-    await storage.put(key, Buffer.from('zip'))
-    await db.insert(artifact).values({ id: randomUUID(), projectId, runId, name: 'trace', contentType: 'application/zip', storageKey: key, size: 3 })
-    const dest = resolve(env.STORAGE_DIR, key)
+    const dest = resolve(env.STORAGE_DIR, await seedArtifact(runId, 'trace.zip'))
     expect(existsSync(dest)).toBe(true)
 
     const deleted = await purgeScope(new Date(), {})
@@ -88,10 +100,66 @@ describe('purgeScope', () => {
   })
 })
 
+describe('purgeBeyondLastRuns', () => {
+  it('keeps the newest N runs of every project', async () => {
+    const u = await createUser()
+    const one = await seedProject(u.id)
+    const two = await seedProject(u.id)
+    const now = Date.now()
+    const oldest = await seedRunIn(one, new Date(now - 3 * DAY))
+    const middle = await seedRunIn(one, new Date(now - 2 * DAY))
+    const newest = await seedRunIn(one, new Date(now - 1 * DAY))
+    const other = await seedRunIn(two, new Date(now - 3 * DAY))
+
+    expect(await purgeBeyondLastRuns(2)).toBe(1)
+
+    expect(await exists(oldest)).toBeFalsy()
+    expect(await exists(middle)).toBeTruthy()
+    expect(await exists(newest)).toBeTruthy()
+    expect(await exists(other)).toBeTruthy() // counted per project, not globally
+  })
+
+  it('no-ops when a project has fewer runs than the limit', async () => {
+    const u = await createUser()
+    const runId = await seedRun(u.id, new Date(Date.now() - 100 * DAY))
+
+    expect(await purgeBeyondLastRuns(5)).toBe(0)
+    expect(await exists(runId)).toBeTruthy()
+  })
+
+  it('keeps everything when the limit is 0', async () => {
+    const u = await createUser()
+    const projectId = await seedProject(u.id)
+    await seedRunIn(projectId, new Date(Date.now() - 2 * DAY))
+    const runId = await seedRunIn(projectId, new Date(Date.now() - 1 * DAY))
+
+    expect(await purgeBeyondLastRuns(0)).toBe(0)
+    expect(await exists(runId)).toBeTruthy()
+  })
+})
+
+describe('purgeArtifactsBefore', () => {
+  it('drops old blobs and rows but keeps the runs', async () => {
+    const u = await createUser()
+    const projectId = await seedProject(u.id)
+    const old = await seedRunIn(projectId, new Date(Date.now() - 100 * DAY))
+    const recent = await seedRunIn(projectId, new Date(Date.now() - 1 * DAY))
+    const oldKey = resolve(env.STORAGE_DIR, await seedArtifact(old, 'trace.zip'))
+    const recentKey = resolve(env.STORAGE_DIR, await seedArtifact(recent, 'trace.zip'))
+
+    expect(await purgeArtifactsBefore(new Date(Date.now() - 30 * DAY))).toBe(1)
+
+    expect(existsSync(oldKey)).toBe(false)
+    expect(existsSync(recentKey)).toBe(true)
+    expect(await db.query.artifact.findMany({ where: eq(artifact.runId, old) })).toHaveLength(0)
+    expect(await exists(old)).toBeTruthy() // history survives the blob sweep
+  })
+})
+
 describe('purgeExpiredRuns', () => {
-  it('is a no-op on self-host (cloud off)', async () => {
+  it('is a no-op on self-host with no retention policy set', async () => {
     const a = await createUser()
     await seedRun(a.id, new Date(Date.now() - 1000 * DAY))
-    expect(await purgeExpiredRuns(new Date())).toEqual({ deleted: 0 })
+    expect(await purgeExpiredRuns(new Date())).toEqual({ deleted: 0, artifacts: 0 })
   })
 })

packages/server/test/test-env.ts

@@ -21,6 +21,9 @@ export const TEST_ENV: Record<keyof Env, string> = {
   KINORA_CLOUD: 'false',
   KINORA_DEMO: 'false',
   INGEST_RATE_LIMIT: '600',
+  KINORA_RETENTION_DAYS: '0',
+  KINORA_KEEP_LAST_RUNS: '0',
+  KINORA_ARTIFACT_RETENTION_DAYS: '0',
   POLAR_ACCESS_TOKEN: '',
   POLAR_WEBHOOK_SECRET: '',
   POLAR_PRODUCT_TEAM_ID: 'prod_team_test',

selfhost/.env.example

@@ -33,6 +33,14 @@ SMTP_USER=
 SMTP_PASS=
 SMTP_FROM="kinora <no-reply@example.dev>"
 
+# Retention (optional). 0 = keep everything, the default. Non-zero starts a daily sweep.
+# Deletes stored trace.zip files older than N days (screenshots and videos ride inside them)
+# while keeping the runs, so pass rates, trends and flaky history survive. Start here.
+KINORA_ARTIFACT_RETENTION_DAYS=0
+# These delete whole runs, history included. KINORA_KEEP_LAST_RUNS counts per project.
+KINORA_RETENTION_DAYS=0
+KINORA_KEEP_LAST_RUNS=0
+
 # Artifact storage: leave S3_* empty to store trace.zip on a local volume (default).
 # Set all five to use any S3-compatible store instead.
 S3_ENDPOINT=