Keep pre-2.7 keep-warm records warm across the upgrade (#677)

CHANGELOG.md

@@ -4,6 +4,7 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi.
 
 ## Unreleased
 
+- Addresses #677: on the first OpenCode 2 or Pi start after upgrading to 2.7, keep-warm sessions saved by earlier versions are kept warm instead of dropped; the limit of 24 warms without a real request counts from the restart.
 - A server-fallback boundary is now restored even after a context manager clears its thinking text, instead of being sent to Anthropic as an invalid thinking block.
 ## 2.7.1
 

packages/core/src/cachekeep.ts

@@ -1283,7 +1283,7 @@ export class CacheKeepManager {
     >
   }): { tracked: true } | { tracked: false; reason: string } {
     const now = this.options.now?.() ?? Date.now()
-    const acceptedAt = input.acceptedAt ?? now
+    const acceptedAt = input.acceptedAt ?? input.retained?.lastTouchAt ?? now
     if (input.retained && input.retained.cacheExpiresAt <= now)
       return { tracked: false, reason: 'cache expired' }
     // Measured only once the policy checks pass: track() runs on every live
@@ -1337,8 +1337,7 @@ export class CacheKeepManager {
     if (
       input.retained &&
       (planned || isCacheKeepAlways(input.storage)) &&
-      (!input.retained.policyState ||
-        input.retained.policyState.needsRealRequest)
+      input.retained.policyState?.needsRealRequest
     )
       return retire('unsupported')
     if (
@@ -1446,6 +1445,9 @@ export class CacheKeepManager {
       oauthAccountIdentity: input.oauthAccountIdentity,
       accountStoragePath: input.accountStoragePath,
       isSubagent: input.isSubagent ?? false,
+      // Every 2.7 writer saves policyState for tracked targets. Its absence means
+      // an older record: start its count here using the same state as a new target.
+      // onTargetChanged and reserveWarm persist that state before the first send.
       policyState: input.retained?.policyState
         ? {
             ...input.retained.policyState,
@@ -1460,8 +1462,9 @@ export class CacheKeepManager {
             lastRealRequestAt: acceptedAt,
             warmsSinceLastRealRequest: 0,
             warmCostSoFar: 0,
-            hasWarmed: previous?.policyState?.hasWarmed ?? false,
-            ...(input.retained && { needsRealRequest: true }),
+            hasWarmed: input.retained
+              ? false
+              : (previous?.policyState?.hasWarmed ?? false),
           },
       ...(input.sourceWarm === true && { sourceWarm: true }),
     })

packages/core/src/tests/cachekeep-policy.test.ts

@@ -385,14 +385,26 @@ test.each([
   },
 )
 
-test('old restart metadata cannot reset a planned or local always count', () => {
+test('restart metadata requiring a real request cannot reset a planned or local always count', () => {
   const f = plannedFixture()
   const retained = {
     cacheExpiresAt: f.now + 300_000,
     nextPrewarmAt: f.now + 270_000,
     ttlMs: 300_000,
+    policyState: {
+      ownerSessionId: f.input.sessionId,
+      planned: false,
+      lastRealRequestAt: f.now - 60_000,
+      warmsSinceLastRealRequest: 24,
+      warmCostSoFar: 0,
+      hasWarmed: true,
+      needsRealRequest: true,
+    },
   }
-  expect(f.track({ retained }).tracked).toBe(false)
+  expect(f.track({ retained })).toEqual({
+    tracked: false,
+    reason: 'unsupported',
+  })
   f.setPolicy(null)
   expect(
     f.track({
@@ -403,8 +415,78 @@ test('old restart metadata cannot reset a planned or local always count', () =>
         claudeCache: { enabled: true, mode: 'hybrid' },
         cacheKeep: { enabled: true, always: true },
       },
-    }).tracked,
-  ).toBe(false)
+    }),
+  ).toEqual({ tracked: false, reason: 'unsupported' })
+})
+
+test.each([false, true])(
+  'legacy restore starts a fresh 24-warm count using its last touch (planned=%s)',
+  async (planned) => {
+    const f = plannedFixture({}, { ttl: 'long', keep_warm: 'always' })
+    if (!planned) {
+      f.setPolicy(null)
+      Object.assign(f.input.storage, {
+        claudeCache: { enabled: true, mode: 'hybrid' },
+        cacheKeep: { enabled: true, always: true },
+      })
+    }
+    const lastTouchAt = f.now - 60_000
+    const interval = planned ? 3_570_000 : 3_300_000
+    expect(
+      f.track({
+        cacheMode: 'hybrid',
+        retained: {
+          cacheExpiresAt: lastTouchAt + 3_600_000,
+          nextPrewarmAt: lastTouchAt + interval,
+          ttlMs: 3_600_000,
+          lastTouchAt,
+          lastTouchKind: 'warm',
+        },
+      }),
+    ).toEqual({ tracked: true })
+    expect(f.target.policyState).toEqual({
+      ownerSessionId: f.input.sessionId,
+      planned,
+      lastRealRequestAt: lastTouchAt,
+      warmsSinceLastRealRequest: 0,
+      warmCostSoFar: 0,
+      hasWarmed: false,
+    })
+    for (let count = 1; count <= 24; count++) {
+      f.advance(count === 1 ? interval - 60_000 : interval)
+      await f.manager.tick()
+      expect(f.sends).toBe(count)
+    }
+    f.advance(interval)
+    await f.manager.tick()
+    expect(f.sends).toBe(24)
+    expect(f.manager.trackedCount()).toBe(0)
+  },
+)
+
+test('legacy planned auto starts with zero admitted cost', async () => {
+  const f = plannedFixture({}, { ttl: 'short', keep_warm: 'auto' })
+  expect(
+    f.track({
+      retained: {
+        cacheExpiresAt: f.now + 240_000,
+        nextPrewarmAt: f.now + 210_000,
+        ttlMs: 300_000,
+        lastTouchAt: f.now - 60_000,
+        lastTouchKind: 'response',
+      },
+    }),
+  ).toEqual({ tracked: true })
+  expect(f.target.policyState?.warmCostSoFar).toBe(0)
+  f.advance(210_000)
+  await f.manager.tick()
+  expect(f.sends).toBe(1)
+  expect(f.target.policyState).toMatchObject({
+    planned: true,
+    warmCostSoFar: 0,
+    warmsSinceLastRealRequest: 1,
+    hasWarmed: true,
+  })
 })
 
 test('source synchronization does not reset a per-target reserved count or lifetime', async () => {
@@ -478,7 +560,7 @@ test('planned timeout is clipped to expiry minus safety and authorization cannot
   }
 })
 
-test('a restored legacy hours target cannot acquire a planned baseline without a real request', async () => {
+test('a restored hours target requiring a real request cannot acquire a planned baseline', async () => {
   const f = plannedFixture()
   f.setPolicy(null)
   const hour = new Date(f.now).getHours()
@@ -492,6 +574,15 @@ test('a restored legacy hours target cannot acquire a planned baseline without a
     cacheExpiresAt: f.now + 300_000,
     nextPrewarmAt: f.now + 270_000,
     ttlMs: 300_000,
+    policyState: {
+      ownerSessionId: f.input.sessionId,
+      planned: false,
+      lastRealRequestAt: f.now - 60_000,
+      warmsSinceLastRealRequest: 24,
+      warmCostSoFar: 0,
+      hasWarmed: true,
+      needsRealRequest: true,
+    },
   }
   expect(f.track({ retained, storage, cacheMode: 'hybrid' }).tracked).toBe(true)
   f.setPolicy({ ttl: 'short', keep_warm: 'always' })

packages/opencode/src/tests/opencode2-cachekeep.test.ts

@@ -1417,6 +1417,80 @@ test('OpenCode 2 restores and prewarms a live format-1 record from another plugi
   expect(warm.max_tokens).toBe(0)
 })
 
+test('OpenCode 2 restores a 2.6 record without policyState and durably caps its new count', async () => {
+  const lines = captureStartupLogs()
+  const f = await fixture(false, false, 'debug')
+  const live = (await (await f.live()).json()) as Record<string, any>
+  await f.dispose()
+  const [name] = await readdir(f.disk)
+  const path = join(f.disk, name!)
+  const saved = JSON.parse(gunzipSync(await readFile(path)).toString())
+  expect(saved.policyState).toBeDefined()
+  saved.pluginVersion = '2.6.0'
+  delete saved.policyState
+  await writeFile(path, gzipSync(JSON.stringify(saved)))
+  forgetCacheKeepProcess(f.paths.storageId)
+  f.advance(60_000)
+  lines.length = 0
+  const next = await f.start()
+  expect(
+    lines.find((line) => line.includes('restart records loaded')),
+  ).toContain('"restored":1,"dropped":0')
+  const key = Symbol.for('cortexkit.anthropic-auth.opencode2.cachekeep.v2')
+  const owner = (globalThis as any)[key].get(f.paths.storageId)
+  expect(owner.manager.trackedPolicyState('ses_oc2_keep')).toMatchObject({
+    planned: false,
+    lastRealRequestAt: saved.lastTouchAt,
+    warmsSinceLastRealRequest: 0,
+    warmCostSoFar: 0,
+    hasWarmed: false,
+  })
+  await owner.persistence.drain()
+  expect(
+    JSON.parse(gunzipSync(await readFile(path)).toString()).policyState
+      .warmsSinceLastRealRequest,
+  ).toBe(0)
+  const reserved: number[] = []
+  next.setSend(async () => {
+    const persisted = JSON.parse(gunzipSync(await readFile(path)).toString())
+    reserved.push(persisted.policyState.warmsSinceLastRealRequest)
+    return Response.json({})
+  })
+  for (let count = 1; count <= 25; count++) {
+    f.due()
+    await f.timers.fire()
+    expect(next.sent).toHaveLength(Math.min(count, 24))
+  }
+  expect(reserved).toEqual(Array.from({ length: 24 }, (_, i) => i + 1))
+  const warm = (await next.sent[0]!.json()) as Record<string, any>
+  expect(warm.messages).toEqual(live.messages)
+  expect(warm.system.slice(1)).toEqual(live.system.slice(1))
+  expect(owner.manager.trackedCount()).toBe(0)
+})
+
+test('OpenCode 2 still drops a restart record requiring a real request', async () => {
+  const lines = captureStartupLogs()
+  const f = await fixture(false, false, 'debug')
+  await f.live()
+  await f.dispose()
+  const [name] = await readdir(f.disk)
+  const path = join(f.disk, name!)
+  const saved = JSON.parse(gunzipSync(await readFile(path)).toString())
+  saved.policyState.needsRealRequest = true
+  saved.policyState.warmsSinceLastRealRequest = 24
+  await writeFile(path, gzipSync(JSON.stringify(saved)))
+  forgetCacheKeepProcess(f.paths.storageId)
+  lines.length = 0
+  const next = await f.start()
+  expect(lines.some((line) => line.includes('"reason":"unsupported"'))).toBe(
+    true,
+  )
+  expect(await readdir(f.disk)).toEqual([])
+  f.due()
+  await f.timers.fire()
+  expect(next.sent).toEqual([])
+})
+
 test.each([false, true])(
   'an OpenCode 2 adopted entry writes through the newer store (legacy callbacks: %s)',
   async (legacy) => {

packages/pi/src/tests/cachekeep.test.ts

@@ -1,8 +1,9 @@
 import { afterEach, expect, test } from 'bun:test'
 import { createHash } from 'node:crypto'
-import { mkdtemp, readdir, readFile, rm } from 'node:fs/promises'
+import { mkdtemp, readdir, readFile, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
+import { gunzipSync, gzipSync } from 'node:zlib'
 import {
   type CacheKeepManager,
   cacheKeepStoreDirectory,
@@ -159,6 +160,73 @@ test('a fresh Pi process restores its idle session and warms without a request',
   expect(f.delays[0]).toBeLessThan(3000)
 })
 
+test('Pi restores a 2.6 record without policyState and durably caps its new count', async () => {
+  const f = await fixture()
+  const first = f.create()
+  await first.process.trackAccepted(f.input('legacy-pi'), f.snapshot)
+  await first.process.stop()
+  const directory = cacheKeepStoreDirectory('shared-storage', f.root, 'pi')
+  const [name] = (await readdir(directory)).filter((name) =>
+    name.endsWith('.gz'),
+  )
+  const path = join(directory, name!)
+  const saved = JSON.parse(gunzipSync(await readFile(path)).toString())
+  expect(saved.policyState).toBeDefined()
+  saved.pluginVersion = '2.6.0'
+  delete saved.policyState
+  await writeFile(path, gzipSync(JSON.stringify(saved)))
+  f.advance(60_000)
+  const reserved: number[] = []
+  const resumed = f.create(undefined, async () => {
+    const persisted = JSON.parse(gunzipSync(await readFile(path)).toString())
+    reserved.push(persisted.policyState.warmsSinceLastRealRequest)
+    return Response.json({})
+  })
+  await resumed.process.startSession('pi-storage', 'legacy-pi')
+  expect(resumed.process.manager.trackedPolicyState('legacy-pi')).toMatchObject(
+    {
+      planned: false,
+      lastRealRequestAt: saved.lastTouchAt,
+      warmsSinceLastRealRequest: 0,
+      warmCostSoFar: 0,
+      hasWarmed: false,
+    },
+  )
+  for (let count = 1; count <= 25; count++) {
+    f.advance(55 * 60_000)
+    await resumed.tick()
+    expect(f.sends).toHaveLength(Math.min(count, 24))
+  }
+  expect(reserved).toEqual(Array.from({ length: 24 }, (_, i) => i + 1))
+  expect(f.sends[0]?.body.system).toEqual(JSON.parse(saved.bodyText).system)
+  expect(resumed.process.manager.trackedCount()).toBe(0)
+})
+
+test('Pi still drops a restart record requiring a real request', async () => {
+  const f = await fixture()
+  const first = f.create()
+  await first.process.trackAccepted(f.input('stopped-pi'), f.snapshot)
+  await first.process.stop()
+  const directory = cacheKeepStoreDirectory('shared-storage', f.root, 'pi')
+  const [name] = (await readdir(directory)).filter((name) =>
+    name.endsWith('.gz'),
+  )
+  const path = join(directory, name!)
+  const saved = JSON.parse(gunzipSync(await readFile(path)).toString())
+  saved.policyState.needsRealRequest = true
+  saved.policyState.warmsSinceLastRealRequest = 24
+  await writeFile(path, gzipSync(JSON.stringify(saved)))
+  const resumed = f.create()
+  await resumed.process.startSession('pi-storage', 'stopped-pi')
+  expect(resumed.process.manager.trackedCount()).toBe(0)
+  expect(
+    (await readdir(directory)).filter((name) => name.endsWith('.gz')),
+  ).toEqual([])
+  f.advance(55 * 60_000)
+  await resumed.tick()
+  expect(f.sends).toEqual([])
+})
+
 test('two live Pi processes cannot restore or track the same session', async () => {
   const f = await fixture()
   const seed = f.create()