From 4b4add8afa1442e063ed655a0eb3e6c86b110606 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:10:39 +0100 Subject: [PATCH] mason: preserve legacy keep-warm sessions on restore (#677) --- CHANGELOG.md | 1 + packages/core/src/cachekeep.ts | 13 ++- .../core/src/tests/cachekeep-policy.test.ts | 101 +++++++++++++++++- .../src/tests/opencode2-cachekeep.test.ts | 74 +++++++++++++ packages/pi/src/tests/cachekeep.test.ts | 70 +++++++++++- 5 files changed, 248 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4204ea86..48de3c94 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/packages/core/src/cachekeep.ts b/packages/core/src/cachekeep.ts index 3347381f..e875449e 100644 --- a/packages/core/src/cachekeep.ts +++ b/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 }), }) diff --git a/packages/core/src/tests/cachekeep-policy.test.ts b/packages/core/src/tests/cachekeep-policy.test.ts index 47a6e1e6..8f64666e 100644 --- a/packages/core/src/tests/cachekeep-policy.test.ts +++ b/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' }) diff --git a/packages/opencode/src/tests/opencode2-cachekeep.test.ts b/packages/opencode/src/tests/opencode2-cachekeep.test.ts index 98d318c1..69ca7cb7 100644 --- a/packages/opencode/src/tests/opencode2-cachekeep.test.ts +++ b/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 + 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 + 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) => { diff --git a/packages/pi/src/tests/cachekeep.test.ts b/packages/pi/src/tests/cachekeep.test.ts index 8577a449..11df5237 100644 --- a/packages/pi/src/tests/cachekeep.test.ts +++ b/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()