fix(ccr): sliding idle-window TTL with max-lifetime ceiling in the Rust core backends (#2604) (#2631)

## Description

Rust-core counterpart of the CCR mid-session expiry fix. #2604 (and its
duplicate #2616) report that the 30-minute wall-clock TTL kills entries
in the middle of a normal multi-agent burst: the clock starts at
compression time and never refreshes, so an entry the session keeps
touching still dies.

#2607 fixes this on the Python side by turning the TTL into an idle
window that restarts on every successful retrieval, bounded by an
absolute max lifetime (8x the idle TTL) — but it explicitly notes the
caveat that the Rust core still measures TTL from insertion. This PR
closes that gap: all three Rust CCR backends (`InMemoryCcrStore`,
`SqliteCcrStore`, `RedisCcrStore`) now use the same sliding idle-window
+ max-lifetime-ceiling semantics as the Python `CompressionStore`.

Scoped to the Rust core only; it deliberately does not touch
`DEFAULT_TTL`'s value (1800), which #2607 bumps to 3600 — happy to
rebase in lockstep whichever lands first.

Refs #2604, #2616. Complements #2607.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `crates/headroom-core/src/ccr/mod.rs`:
`DEFAULT_MAX_LIFETIME_MULTIPLIER = 8` + `max_lifetime_for()` helper;
documents the idle-window semantics.
- `in_memory.rs`: entries track `last_accessed`; a hit refreshes it
under the shard write lock (`get_mut`), expiry checks idle window OR max
lifetime, and the existing `remove_if` TOCTOU protection now uses the
same predicate. New `with_capacity_and_ttls` constructor for independent
control of window and ceiling.
- `sqlite.rs`: new `last_accessed` column (legacy DBs migrated in place
via `ALTER TABLE`, backfilled from `created_at` so old rows keep their
original expiry baseline); lazy purge and the lookup honour both bounds;
a hit touches the row under the same connection mutex as the read. New
`open_with_ttls` constructor.
- `redis.rs`: a hit re-arms the key's expiry, capped by a companion
`{prefix}:{hash}:born` key whose remaining TTL marks the absolute
ceiling; entries written by pre-sliding builds (no born key) are
backfilled rather than dropped.
- `tests/ccr_backends.rs`: 6 new tests — sliding-window survival and
max-lifetime cap for in-memory and SQLite, legacy-schema migration, and
a gated Redis sliding test.

No public API is broken: existing constructors keep their signatures and
derive the ceiling as 8x the idle TTL.

## Testing

- [x] Unit tests pass (`cargo test -p headroom-core`)
- [x] Linting passes (`cargo clippy -p headroom-core --all-features`,
`cargo fmt --check`)
- [ ] Type checking passes (`mypy headroom`) — N/A, no Python files
touched
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ cargo test -p headroom-core --test ccr_backends
test result: ok. 12 passed; 0 failed; 0 ignored (8.12s)

$ cargo test -p headroom-core ccr          # all ccr-named tests across suites
38 passed, 949 filtered out (13 suites)

$ cargo test -p headroom-core --test ccr_roundtrip --test live_zone_ccr
18 passed (2 suites)

$ cargo check -p headroom-core --features redis   # cfg-gated backend compiles
Finished `dev` profile in 25.09s

$ cargo clippy -p headroom-core --all-features
No issues found
```

## Real Behavior Proof

- Environment: macOS (Darwin 25.5), local checkout at upstream `main`
(57bf720d), `cargo test`.
- Exact command / steps: dropped a proof test file
(`ccr_sliding_ttl_proof.rs`, uses only APIs present on both main and
this branch) into `crates/headroom-core/tests/`, ran it against
unpatched `main` src, then against this branch. The in-memory case
touches an entry every 60ms with a 120ms TTL; the SQLite case touches at
t+2s with a 3s TTL and reads again at t+4s — i.e. the issue's "session
keeps using the entry" timeline scaled down.
- Observed result: on unpatched `main` both proof tests fail (in-memory:
"entry vanished on touch #2 despite constant access"; SQLite: "entry
expired at t+4s even though the session touched it at t+2s"); on this
branch the same tests pass 2/2. Full output:

  Before (main, wall-clock TTL):

  ```text
---- proof_in_memory_entry_survives_while_session_keeps_touching_it
stdout ----
  panicked: entry vanished on touch #2 despite constant access

---- proof_sqlite_entry_survives_while_session_keeps_touching_it stdout
----
panicked: entry expired at t+4s even though the session touched it at
t+2s (wall-clock TTL)

  test result: FAILED. 0 passed; 2 failed
  ```

  After (this branch, sliding idle window):

  ```text
  test result: ok. 2 passed; 0 failed (4.01s)
  ```

- Not tested: the Redis backend against a live Redis (the new
`redis_get_refreshes_idle_ttl` test self-skips without
`HEADROOM_TEST_REDIS_URL`, same as the existing gated tests; it compiles
under `--features redis` and runs in the CI redis matrix).

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md` — it is generated by
release-please from my Conventional Commit PR title (a CI guard enforces
this)

## Additional Notes

- Docs: `docs/content/docs/ccr.mdx` TTL wording is being updated by
#2607; not duplicated here to avoid conflicting hunks.
- The SQLite migration is intentionally in-place and idempotent
(`pragma_table_info` check → `ALTER TABLE ADD COLUMN` → backfill), so a
proxy restarting onto an existing `ccr.sqlite` keeps its rows.
- If #2607 lands first I will rebase; the only expected overlap is the
doc comment around `DEFAULT_TTL`.

crates/headroom-core/src/ccr/backends/in_memory.rs

@@ -15,13 +15,16 @@ use std::time::{Duration, Instant};
 
 use dashmap::DashMap;
 
-use crate::ccr::{CcrStore, DEFAULT_CAPACITY, DEFAULT_TTL};
+use crate::ccr::{max_lifetime_for, CcrStore, DEFAULT_CAPACITY, DEFAULT_TTL};
 
 /// In-memory CCR store backed by [`DashMap`] for sharded concurrent
 /// access.
 ///
-/// - **TTL**: 30 minutes by default. Entries past their TTL are dropped
+/// - **TTL**: 30 minutes by default, treated as an **idle window** —
-///   on the next `get` (lazy expiry — no background reaper thread).
+///   every successful `get` restarts the entry's clock (#2604), bounded
+///   by an absolute max lifetime of 8x the idle TTL measured from
+///   insertion. Entries past their window are dropped on the next `get`
+///   (lazy expiry — no background reaper thread).
 /// - **Capacity**: 1000 entries by default. When `put` would push us
 ///   past capacity, the oldest entry (per insertion order) is evicted.
 /// - **Concurrency**: gets and puts on distinct keys do not contend.
@@ -36,6 +39,7 @@ pub struct InMemoryCcrStore {
     /// they actually evict a real entry.
     order: Mutex<VecDeque<String>>,
     ttl: Duration,
+    max_lifetime: Duration,
     capacity: usize,
 }
 
@@ -43,19 +47,38 @@ pub struct InMemoryCcrStore {
 struct Entry {
     payload: String,
     inserted: Instant,
+    last_accessed: Instant,
+}
+
+impl Entry {
+    /// Expired when idle past `ttl` OR older (since insertion) than
+    /// `max_lifetime` — the absolute ceiling that keeps constant access
+    /// from pinning an entry forever.
+    fn is_expired(&self, ttl: Duration, max_lifetime: Duration) -> bool {
+        self.last_accessed.elapsed() > ttl || self.inserted.elapsed() > max_lifetime
+    }
 }
 
 impl InMemoryCcrStore {
-    /// Default: 1000 entries, 30-minute TTL.
+    /// Default: 1000 entries, 30-minute idle TTL (8x max lifetime).
     pub fn new() -> Self {
         Self::with_capacity_and_ttl(DEFAULT_CAPACITY, DEFAULT_TTL)
     }
 
+    /// `ttl` is the idle window; the absolute max lifetime defaults to
+    /// 8x that (see [`crate::ccr::DEFAULT_MAX_LIFETIME_MULTIPLIER`]).
     pub fn with_capacity_and_ttl(capacity: usize, ttl: Duration) -> Self {
+        Self::with_capacity_and_ttls(capacity, ttl, max_lifetime_for(ttl))
+    }
+
+    /// Full-control constructor: idle window and absolute max lifetime
+    /// specified independently.
+    pub fn with_capacity_and_ttls(capacity: usize, ttl: Duration, max_lifetime: Duration) -> Self {
         Self {
             map: DashMap::with_capacity(capacity),
             order: Mutex::new(VecDeque::with_capacity(capacity)),
             ttl,
+            max_lifetime,
             capacity,
         }
     }
@@ -89,8 +112,10 @@ impl CcrStore for InMemoryCcrStore {
         // in place, leave the order queue alone. Common when the same
         // tool output flows through multiple times in a session.
         if let Some(mut existing) = self.map.get_mut(hash) {
+            let now = Instant::now();
             existing.payload = payload.to_string();
-            existing.inserted = Instant::now();
+            existing.inserted = now;
+            existing.last_accessed = now;
             return;
         }
 
@@ -99,9 +124,11 @@ impl CcrStore for InMemoryCcrStore {
         if self.map.len() >= self.capacity {
             self.evict_until_under_capacity();
         }
+        let now = Instant::now();
         let entry = Entry {
             payload: payload.to_string(),
-            inserted: Instant::now(),
+            inserted: now,
+            last_accessed: now,
         };
         let prev = self.map.insert(hash.to_string(), entry);
         if prev.is_none() {
@@ -117,9 +144,12 @@ impl CcrStore for InMemoryCcrStore {
     }
 
     fn get(&self, hash: &str) -> Option<String> {
-        // Read path: shard read-lock, check TTL, clone payload out.
+        // Hit path: shard write-lock (get_mut), check the idle window +
-        // No global lock involvement at all — distinct hashes hash to
+        // max-lifetime ceiling, refresh `last_accessed`, clone payload
-        // distinct shards and never contend.
+        // out. The TTL is a sliding idle window (#2604): every hit
+        // restarts the clock, so an entry a session keeps touching does
+        // not expire mid-burst. Distinct hashes hash to distinct shards
+        // and never contend.
         //
         // Lazy expiry uses DashMap's `remove_if` so the check-and-remove
         // is atomic on the shard. An earlier 2-step (drop read lock,
@@ -130,8 +160,9 @@ impl CcrStore for InMemoryCcrStore {
         // load this manifested as "I just stored it; why is it gone?"
         // `remove_if` closes the window because the shard write lock
         // is held across both the predicate evaluation and the removal.
-        if let Some(entry) = self.map.get(hash) {
+        if let Some(mut entry) = self.map.get_mut(hash) {
-            if entry.inserted.elapsed() <= self.ttl {
+            if !entry.is_expired(self.ttl, self.max_lifetime) {
+                entry.last_accessed = Instant::now();
                 return Some(entry.payload.clone());
             }
         } else {
@@ -143,7 +174,9 @@ impl CcrStore for InMemoryCcrStore {
         // and re-fetch its payload.
         let was_removed = self
             .map
-            .remove_if(hash, |_, entry| entry.inserted.elapsed() > self.ttl)
+            .remove_if(hash, |_, entry| {
+                entry.is_expired(self.ttl, self.max_lifetime)
+            })
             .is_some();
         if was_removed {
             None

crates/headroom-core/src/ccr/backends/redis.rs

@@ -10,10 +10,12 @@
 //! # Storage model
 //!
 //! Each entry maps to a Redis key `ccr:{hash}` containing the original
-//! payload bytes, with a `SETEX` TTL applied on every write. Read path
+//! payload bytes, with a `SETEX` TTL applied on every write. The TTL is
-//! is a single `GET`. Redis handles purging via key expiry — no
+//! an **idle window** (#2604): every successful `get` re-arms the key's
-//! application-side sweep needed (matching the SQLite backend's
+//! expiry, bounded by an absolute max lifetime tracked in a companion
-//! lazy-purge but at the Redis level).
+//! `ccr:{hash}:born` key whose own expiry marks the ceiling. Redis
+//! handles purging via key expiry — no application-side sweep needed
+//! (matching the SQLite backend's lazy-purge but at the Redis level).
 //!
 //! # Concurrency
 //!
@@ -27,7 +29,7 @@
 
 use redis::Commands;
 
-use crate::ccr::CcrStore;
+use crate::ccr::{max_lifetime_for, CcrStore};
 
 /// Key prefix applied to every CCR entry. Configurable per-deployment
 /// so multiple proxies sharing one Redis don't collide.
@@ -38,6 +40,9 @@ pub struct RedisCcrStore {
     client: redis::Client,
     key_prefix: String,
     default_ttl_seconds: u64,
+    /// Absolute max lifetime (seconds since `put`) that caps the
+    /// sliding idle window. Defaults to 8x the idle TTL.
+    max_lifetime_seconds: u64,
 }
 
 impl RedisCcrStore {
@@ -59,10 +64,13 @@ impl RedisCcrStore {
         // signal.
         let mut conn = client.get_connection()?;
         let _: String = redis::cmd("PING").query(&mut conn)?;
+        let max_lifetime_seconds =
+            max_lifetime_for(std::time::Duration::from_secs(default_ttl_seconds)).as_secs();
         Ok(Self {
             client,
             key_prefix,
             default_ttl_seconds,
+            max_lifetime_seconds,
         })
     }
 
@@ -70,6 +78,12 @@ impl RedisCcrStore {
         format!("{}:{}", self.key_prefix, hash)
     }
 
+    /// Companion key whose expiry marks the entry's absolute max
+    /// lifetime; its remaining TTL caps every idle-window re-arm.
+    fn born_key_for(&self, hash: &str) -> String {
+        format!("{}:{}:born", self.key_prefix, hash)
+    }
+
     /// Default TTL (seconds) applied on every `put`.
     pub fn default_ttl_seconds(&self) -> u64 {
         self.default_ttl_seconds
@@ -102,6 +116,20 @@ impl CcrStore for RedisCcrStore {
                 error = %err,
                 "ccr_redis_put_failed"
             );
+            return;
+        }
+        // Companion max-lifetime marker: its remaining TTL caps every
+        // idle-window re-arm in `get`, so constant access cannot pin an
+        // entry past `max_lifetime_seconds`.
+        let born: redis::RedisResult<()> =
+            conn.set_ex(self.born_key_for(hash), 1_u8, self.max_lifetime_seconds);
+        if let Err(err) = born {
+            tracing::warn!(
+                target = "ccr.redis",
+                hash = %hash,
+                error = %err,
+                "ccr_redis_put_born_failed"
+            );
         }
     }
 
@@ -120,9 +148,9 @@ impl CcrStore for RedisCcrStore {
             }
         };
         let bytes: redis::RedisResult<Option<Vec<u8>>> = conn.get(&key);
-        match bytes {
+        let payload = match bytes {
-            Ok(Some(bytes)) => String::from_utf8(bytes).ok(),
+            Ok(Some(bytes)) => String::from_utf8(bytes).ok()?,
-            Ok(None) => None,
+            Ok(None) => return None,
             Err(err) => {
                 tracing::warn!(
                     target = "ccr.redis",
@@ -130,9 +158,48 @@ impl CcrStore for RedisCcrStore {
                     error = %err,
                     "ccr_redis_get_failed"
                 );
-                None
+                return None;
             }
+        };
+
+        // Sliding idle window (#2604): re-arm the key's expiry on every
+        // hit, capped by the companion born-key's remaining lifetime.
+        let born_key = self.born_key_for(hash);
+        let born_remaining: i64 = conn.ttl(&born_key).unwrap_or(-1);
+        let remaining = if born_remaining >= 0 {
+            born_remaining as u64
+        } else {
+            // Legacy entry written by a pre-sliding build (no born key):
+            // backfill the ceiling from now rather than dropping data.
+            let backfill: redis::RedisResult<()> =
+                conn.set_ex(&born_key, 1_u8, self.max_lifetime_seconds);
+            if let Err(err) = backfill {
+                tracing::warn!(
+                    target = "ccr.redis",
+                    hash = %hash,
+                    error = %err,
+                    "ccr_redis_born_backfill_failed"
+                );
+            }
+            self.max_lifetime_seconds
+        };
+        let new_ttl = self.default_ttl_seconds.min(remaining);
+        if new_ttl == 0 {
+            // Past the max lifetime: purge rather than serve a pinned
+            // entry that should have died.
+            let _: redis::RedisResult<()> = conn.del(&key);
+            return None;
         }
+        let rearm: redis::RedisResult<()> = conn.expire(&key, new_ttl as i64);
+        if let Err(err) = rearm {
+            tracing::warn!(
+                target = "ccr.redis",
+                hash = %hash,
+                error = %err,
+                "ccr_redis_ttl_rearm_failed"
+            );
+        }
+        Some(payload)
     }
 
     fn len(&self) -> usize {

crates/headroom-core/src/ccr/backends/sqlite.rs

@@ -5,16 +5,21 @@
 //!
 //! ```sql
 //! CREATE TABLE IF NOT EXISTS ccr_entries (
-//!     hash         TEXT PRIMARY KEY,
+//!     hash          TEXT PRIMARY KEY,
-//!     original     BLOB NOT NULL,
+//!     original      BLOB NOT NULL,
-//!     created_at   INTEGER NOT NULL,   -- unix-seconds
+//!     created_at    INTEGER NOT NULL,   -- unix-seconds
-//!     ttl_seconds  INTEGER NOT NULL
+//!     ttl_seconds   INTEGER NOT NULL,   -- idle window, restarted on get
+//!     last_accessed INTEGER NOT NULL    -- unix-seconds
 //! );
 //! ```
 //!
-//! On every `get` we lazy-purge stale rows
+//! The TTL is an **idle window** (#2604): every successful `get`
-//! (`WHERE created_at + ttl_seconds <= now`) — no background reaper
+//! restarts the row's clock via `last_accessed`, bounded by an absolute
-//! thread, no cron.
+//! max lifetime measured from `created_at`. On every `get` we
+//! lazy-purge stale rows (`WHERE last_accessed + ttl_seconds <= now OR
+//! created_at + max_lifetime <= now`) — no background reaper thread,
+//! no cron. DBs created by pre-sliding builds are migrated in place
+//! (the `last_accessed` column is added, backfilled from `created_at`).
 //!
 //! All hot statements are prepared once on connection setup and reused
 //! per call (per realignment build constraint #5: performant). Writes
@@ -43,14 +48,17 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use rusqlite::{params, Connection, OptionalExtension};
 
-use crate::ccr::CcrStore;
+use crate::ccr::{max_lifetime_for, CcrStore};
 
 /// SQLite-backed CCR store.
 pub struct SqliteCcrStore {
     conn: Mutex<Connection>,
-    /// Default TTL applied on every `put`. Mirrors Python's
+    /// Default idle TTL applied on every `put`. Mirrors Python's
-    /// `compression_store` 30-minute window.
+    /// `compression_store` idle window.
     default_ttl_seconds: u64,
+    /// Absolute max lifetime (seconds since `created_at`) that caps the
+    /// sliding idle window. Defaults to 8x the idle TTL.
+    max_lifetime_seconds: u64,
     /// Path the connection was opened against — kept for diagnostics
     /// and for the proxy-restart simulation test.
     path: PathBuf,
@@ -58,9 +66,24 @@ pub struct SqliteCcrStore {
 
 impl SqliteCcrStore {
     /// Open or create the DB file at `path` and prepare the schema.
+    /// `default_ttl_seconds` is the idle window; the absolute max
+    /// lifetime defaults to 8x that (see
+    /// [`crate::ccr::DEFAULT_MAX_LIFETIME_MULTIPLIER`]).
     /// Errors surface to the caller (`from_config`); we never silently
     /// fall back to the in-memory backend (`feedback_no_silent_fallbacks.md`).
     pub fn open(path: impl AsRef<Path>, default_ttl_seconds: u64) -> rusqlite::Result<Self> {
+        let max_lifetime =
+            max_lifetime_for(std::time::Duration::from_secs(default_ttl_seconds)).as_secs();
+        Self::open_with_ttls(path, default_ttl_seconds, max_lifetime)
+    }
+
+    /// Full-control constructor: idle window and absolute max lifetime
+    /// specified independently.
+    pub fn open_with_ttls(
+        path: impl AsRef<Path>,
+        default_ttl_seconds: u64,
+        max_lifetime_seconds: u64,
+    ) -> rusqlite::Result<Self> {
         let path_buf = path.as_ref().to_path_buf();
         let conn = Connection::open(&path_buf)?;
 
@@ -73,25 +96,49 @@ impl SqliteCcrStore {
 
         conn.execute(
             "CREATE TABLE IF NOT EXISTS ccr_entries (
-                 hash         TEXT PRIMARY KEY,
+                 hash          TEXT PRIMARY KEY,
-                 original     BLOB NOT NULL,
+                 original      BLOB NOT NULL,
-                 created_at   INTEGER NOT NULL,
+                 created_at    INTEGER NOT NULL,
-                 ttl_seconds  INTEGER NOT NULL
+                 ttl_seconds   INTEGER NOT NULL,
+                 last_accessed INTEGER NOT NULL
              )",
             [],
         )?;
+        Self::migrate_legacy_schema(&conn)?;
         // No secondary index — the schema is one-row-per-PK and the only
         // non-PK lookup (the lazy-purge sweep) is a `WHERE` predicate on
-        // a small table; an index on `created_at + ttl_seconds` would
+        // a small table; an index on the expiry expressions would cost
-        // cost more than it saves.
+        // more than it saves.
 
         Ok(Self {
             conn: Mutex::new(conn),
             default_ttl_seconds,
+            max_lifetime_seconds,
             path: path_buf,
         })
     }
 
+    /// DBs created before the sliding-TTL change lack `last_accessed`.
+    /// Add it in place and backfill from `created_at` so legacy rows
+    /// keep their original expiry baseline rather than being purged or
+    /// artificially refreshed.
+    fn migrate_legacy_schema(conn: &Connection) -> rusqlite::Result<()> {
+        let has_last_accessed = conn
+            .prepare("SELECT 1 FROM pragma_table_info('ccr_entries') WHERE name = 'last_accessed'")?
+            .exists([])?;
+        if !has_last_accessed {
+            conn.execute(
+                "ALTER TABLE ccr_entries ADD COLUMN last_accessed INTEGER NOT NULL DEFAULT 0",
+                [],
+            )?;
+            conn.execute(
+                "UPDATE ccr_entries SET last_accessed = created_at WHERE last_accessed = 0",
+                [],
+            )?;
+        }
+        Ok(())
+    }
+
     /// Path the connection was opened against. Test helper.
     pub fn path(&self) -> &Path {
         &self.path
@@ -102,12 +149,15 @@ impl SqliteCcrStore {
         self.default_ttl_seconds
     }
 
-    /// Drop all expired rows. Lazy — invoked from `get`. Returns the
+    /// Drop all expired rows: idle past their window, or past the
+    /// absolute max lifetime. Lazy — invoked from `get`. Returns the
     /// number of rows purged.
-    fn purge_expired(conn: &Connection, now: u64) -> rusqlite::Result<usize> {
+    fn purge_expired(&self, conn: &Connection, now: u64) -> rusqlite::Result<usize> {
         let purged = conn.execute(
-            "DELETE FROM ccr_entries WHERE created_at + ttl_seconds <= ?1",
+            "DELETE FROM ccr_entries
-            params![now as i64],
+             WHERE last_accessed + ttl_seconds <= ?1
+                OR created_at + ?2 <= ?1",
+            params![now as i64, self.max_lifetime_seconds as i64],
         )?;
         Ok(purged)
     }
@@ -129,12 +179,13 @@ impl CcrStore for SqliteCcrStore {
         // Upsert by PK. ON CONFLICT REPLACE matches the in-memory
         // backend's idempotent re-store semantics.
         let res = conn.execute(
-            "INSERT INTO ccr_entries (hash, original, created_at, ttl_seconds)
+            "INSERT INTO ccr_entries (hash, original, created_at, ttl_seconds, last_accessed)
-             VALUES (?1, ?2, ?3, ?4)
+             VALUES (?1, ?2, ?3, ?4, ?3)
              ON CONFLICT(hash) DO UPDATE SET
-                 original    = excluded.original,
+                 original      = excluded.original,
-                 created_at  = excluded.created_at,
+                 created_at    = excluded.created_at,
-                 ttl_seconds = excluded.ttl_seconds",
+                 ttl_seconds   = excluded.ttl_seconds,
+                 last_accessed = excluded.last_accessed",
             params![
                 hash,
                 payload.as_bytes(),
@@ -165,7 +216,7 @@ impl CcrStore for SqliteCcrStore {
         // Lazy purge sweep, then the real lookup. Both happen under
         // the same mutex so the row we read is guaranteed not to have
         // been just-deleted by another caller.
-        if let Err(err) = Self::purge_expired(&conn, now) {
+        if let Err(err) = self.purge_expired(&conn, now) {
             tracing::warn!(
                 target = "ccr.sqlite",
                 error = %err,
@@ -176,8 +227,10 @@ impl CcrStore for SqliteCcrStore {
         let row: Option<Vec<u8>> = conn
             .query_row(
                 "SELECT original FROM ccr_entries
-                 WHERE hash = ?1 AND created_at + ttl_seconds > ?2",
+                 WHERE hash = ?1
-                params![hash, now as i64],
+                   AND last_accessed + ttl_seconds > ?2
+                   AND created_at + ?3 > ?2",
+                params![hash, now as i64, self.max_lifetime_seconds as i64],
                 |r| r.get::<_, Vec<u8>>(0),
             )
             .optional()
@@ -191,7 +244,22 @@ impl CcrStore for SqliteCcrStore {
                 None
             });
 
-        row.and_then(|bytes| String::from_utf8(bytes).ok())
+        let row = row?;
+        // Sliding idle window (#2604): a successful hit restarts the
+        // row's idle clock. Still under the same mutex as the lookup.
+        if let Err(err) = conn.execute(
+            "UPDATE ccr_entries SET last_accessed = ?2 WHERE hash = ?1",
+            params![hash, now as i64],
+        ) {
+            tracing::warn!(
+                target = "ccr.sqlite",
+                hash = %hash,
+                error = %err,
+                "ccr_sqlite_touch_failed"
+            );
+        }
+
+        String::from_utf8(row).ok()
     }
 
     fn len(&self) -> usize {

crates/headroom-core/src/ccr/mod.rs

@@ -65,6 +65,20 @@ pub const DEFAULT_CAPACITY: usize = 1000;
 /// silently converts "lossless with retrieval" into "lossy".
 pub const DEFAULT_TTL: Duration = Duration::from_secs(1800);
 
+/// The TTL is an **idle window**, not a wall clock: every successful
+/// `get` restarts the entry's clock, so an entry a session keeps
+/// touching survives a long multi-agent burst (#2604). To keep
+/// constant access from pinning an entry forever, an absolute max
+/// lifetime of `DEFAULT_MAX_LIFETIME_MULTIPLIER * ttl` (measured from
+/// insertion) caps the sliding window. Mirrors the Python
+/// `CompressionStore` semantics.
+pub const DEFAULT_MAX_LIFETIME_MULTIPLIER: u32 = 8;
+
+/// Absolute max lifetime for an entry with idle window `idle_ttl`.
+pub fn max_lifetime_for(idle_ttl: Duration) -> Duration {
+    idle_ttl.saturating_mul(DEFAULT_MAX_LIFETIME_MULTIPLIER)
+}
+
 /// Compute the canonical CCR key for `payload`. BLAKE3 → first 24 hex
 /// chars (96 bits — collision-resistant for the bounded LRU population
 /// the proxy will hold). Centralized here so every call site (live-zone

crates/headroom-core/tests/ccr_backends.rs

@@ -152,6 +152,157 @@ fn backend_swap_byte_equal_keys() {
     }
 }
 
+// ─── Sliding (idle-window) TTL semantics — #2604 ───────────────────────
+//
+// The Python `CompressionStore` treats `HEADROOM_CCR_TTL_SECONDS` as an
+// idle window that restarts on every successful retrieval, bounded by an
+// absolute max lifetime (8x the idle TTL). These tests pin the same
+// semantics onto the Rust backends so an entry a session keeps touching
+// does not expire mid-burst.
+
+#[test]
+fn in_memory_get_refreshes_idle_ttl() {
+    let store = InMemoryCcrStore::with_capacity_and_ttl(10, Duration::from_millis(120));
+    let hash = compute_key(b"hot entry");
+    store.put(&hash, "hot entry");
+    // Touch the entry every 60ms for ~4 idle windows' worth of wall
+    // clock. Wall-clock expiry would kill it at 120ms; a sliding idle
+    // window keeps it alive because every hit restarts the clock.
+    for _ in 0..8 {
+        std::thread::sleep(Duration::from_millis(60));
+        assert_eq!(
+            store.get(&hash).as_deref(),
+            Some("hot entry"),
+            "an entry accessed within its idle window must stay alive"
+        );
+    }
+    // Now go idle past the window: the entry must expire.
+    std::thread::sleep(Duration::from_millis(200));
+    assert_eq!(
+        store.get(&hash),
+        None,
+        "an entry idle past its window must expire"
+    );
+}
+
+#[test]
+fn in_memory_max_lifetime_caps_sliding_window() {
+    // Idle TTL 40ms → max lifetime 320ms (8x). Constant access must not
+    // keep the entry alive forever.
+    let store = InMemoryCcrStore::with_capacity_and_ttl(10, Duration::from_millis(40));
+    let hash = compute_key(b"immortal?");
+    store.put(&hash, "immortal?");
+    let deadline = std::time::Instant::now() + Duration::from_millis(600);
+    let mut expired = false;
+    while std::time::Instant::now() < deadline {
+        std::thread::sleep(Duration::from_millis(20));
+        if store.get(&hash).is_none() {
+            expired = true;
+            break;
+        }
+    }
+    assert!(
+        expired,
+        "constant access must not extend an entry past its max lifetime"
+    );
+}
+
+#[test]
+fn sqlite_get_refreshes_idle_ttl() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let path = dir.path().join("ccr.sqlite");
+    // 3-second idle window (unix-second resolution needs whole seconds).
+    let store = SqliteCcrStore::open(&path, 3).expect("open sqlite store");
+    let hash = compute_key(b"sliding sqlite");
+    store.put(&hash, "sliding sqlite");
+    // t+2s: hit inside the window — restarts the idle clock.
+    std::thread::sleep(Duration::from_millis(2_000));
+    assert_eq!(
+        store.get(&hash).as_deref(),
+        Some("sliding sqlite"),
+        "first access within the idle window must hit"
+    );
+    // t+4s: wall-clock expiry would have purged at t+3s; the refresh at
+    // t+2s must keep it alive until t+5s.
+    std::thread::sleep(Duration::from_millis(2_000));
+    assert_eq!(
+        store.get(&hash).as_deref(),
+        Some("sliding sqlite"),
+        "an entry accessed within its idle window must stay alive past the wall-clock TTL"
+    );
+    // Go idle past the window.
+    std::thread::sleep(Duration::from_millis(4_100));
+    assert_eq!(
+        store.get(&hash),
+        None,
+        "an entry idle past its window must be purged"
+    );
+}
+
+#[test]
+fn sqlite_max_lifetime_caps_sliding_window() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let path = dir.path().join("ccr.sqlite");
+    // Idle 2s with a 3s ceiling: constant access must not outlive t+3s.
+    let store =
+        SqliteCcrStore::open_with_ttls(&path, 2, 3).expect("open sqlite store with ceiling");
+    let hash = compute_key(b"capped sqlite");
+    store.put(&hash, "capped sqlite");
+    std::thread::sleep(Duration::from_millis(1_500));
+    assert_eq!(
+        store.get(&hash).as_deref(),
+        Some("capped sqlite"),
+        "entry inside idle window and ceiling must hit"
+    );
+    // Keep touching, but cross the 3s ceiling.
+    std::thread::sleep(Duration::from_millis(2_600));
+    assert_eq!(
+        store.get(&hash),
+        None,
+        "constant access must not extend an entry past its max lifetime"
+    );
+}
+
+#[test]
+fn sqlite_migrates_legacy_schema_without_last_accessed() {
+    // A DB created by a pre-sliding-TTL build has no `last_accessed`
+    // column. Opening it must migrate in place and keep the rows
+    // retrievable (backfilling last_accessed from created_at).
+    let dir = tempfile::tempdir().expect("tempdir");
+    let path = dir.path().join("ccr.sqlite");
+    let payload = "legacy row";
+    let hash = compute_key(payload.as_bytes());
+    {
+        let conn = rusqlite::Connection::open(&path).expect("open raw connection");
+        conn.execute(
+            "CREATE TABLE ccr_entries (
+                 hash         TEXT PRIMARY KEY,
+                 original     BLOB NOT NULL,
+                 created_at   INTEGER NOT NULL,
+                 ttl_seconds  INTEGER NOT NULL
+             )",
+            [],
+        )
+        .expect("create legacy schema");
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_secs() as i64;
+        conn.execute(
+            "INSERT INTO ccr_entries (hash, original, created_at, ttl_seconds)
+             VALUES (?1, ?2, ?3, ?4)",
+            rusqlite::params![hash, payload.as_bytes(), now, 300_i64],
+        )
+        .expect("insert legacy row");
+    }
+    let store = SqliteCcrStore::open(&path, 300).expect("open must migrate legacy schema");
+    assert_eq!(
+        store.get(&hash).as_deref(),
+        Some(payload),
+        "legacy rows must survive the schema migration"
+    );
+}
+
 // ─── Redis-feature-gated tests ─────────────────────────────────────────
 
 #[cfg(feature = "redis")]
@@ -196,4 +347,31 @@ mod redis_tests {
         store.put(&hash, payload);
         assert_eq!(store.get(&hash).as_deref(), Some(payload));
     }
+
+    #[test]
+    fn redis_get_refreshes_idle_ttl() {
+        let Some(url) = redis_url() else {
+            eprintln!("skipping redis_get_refreshes_idle_ttl: HEADROOM_TEST_REDIS_URL not set");
+            return;
+        };
+        // 2-second idle window (Redis EXPIRE has 1s resolution).
+        let store = RedisCcrStore::open_with_prefix(&url, "ccr_test_sliding".to_string(), 2)
+            .expect("open redis store");
+        let payload = "sliding redis";
+        let hash = compute_key(payload.as_bytes());
+        store.put(&hash, payload);
+        // Touch at t+1.5s (inside window) — restarts the idle clock.
+        std::thread::sleep(Duration::from_millis(1_500));
+        assert_eq!(store.get(&hash).as_deref(), Some(payload));
+        // t+3s: wall-clock expiry would have fired at t+2s.
+        std::thread::sleep(Duration::from_millis(1_500));
+        assert_eq!(
+            store.get(&hash).as_deref(),
+            Some(payload),
+            "an entry accessed within its idle window must stay alive past the wall-clock TTL"
+        );
+        // Go idle past the window.
+        std::thread::sleep(Duration::from_millis(3_100));
+        assert_eq!(store.get(&hash), None);
+    }
 }