fix(ccr): make retrieval TTL configurable (#715)

## Description

Make the CCR retrieval store TTL configurable so long-running agent jobs
can keep `headroom_retrieve` markers resolvable beyond the previous
hard-coded 300-second window.

Fixes #714

## 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)
- [x] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Add `HEADROOM_CCR_TTL_SECONDS` for the global CCR `CompressionStore`
default TTL, with validation and fallback to 300 seconds.
- Expose the effective CCR TTL as `store.default_ttl_seconds` from
`/v1/retrieve/stats`.
- Distinguish missing vs expired CCR retrieval failures in
`/v1/retrieve`, `/v1/retrieve/{hash}`, tool-call handling, and CCR
response handling.
- Update CCR docs, README wording, and CHANGELOG for the user-facing TTL
behavior.
- Add regression tests for env-configured TTL, invalid env fallback,
explicit TTL precedence, stats exposure, and expired retrieval detail.

## Reproduction

Before this change, the proxy-global CCR store always used the default
300-second TTL when callers used `get_compression_store()` without an
explicit `default_ttl`. A long-running agent job could receive a
`<<ccr:...>>` marker and fail later with a generic not-found/expired
response after the fixed 5-minute retention window.

The new tests cover this by setting `HEADROOM_CCR_TTL_SECONDS`, creating
CCR entries through the same global store used by the proxy, and
asserting that stored entries and `/v1/retrieve/stats` use the
configured retention.

## Real behavior proof

Setup tested:

- macOS 15.7.2
- Python 3.13.9 via `uv run --frozen --extra dev --extra proxy`
- Local Headroom proxy subprocess: `python -m headroom.proxy.server`
- Proxy config: `HEADROOM_TELEMETRY=off`,
`HEADROOM_CACHE_ENABLED=false`, `HEADROOM_RATE_LIMIT_ENABLED=false`
- Provider/model: no live upstream provider needed; exercised local
`/v1/compress` -> `/v1/retrieve` CCR HTTP flow with model `gpt-4o`

Exact steps run after the patch:

1. Start a real Headroom proxy process with
`HEADROOM_CCR_TTL_SECONDS=7200`.
2. POST a 200-item tool-result payload to `/v1/compress`.
3. Extract the emitted `<<ccr:...>>` marker hash from the compressed
messages.
4. GET `/v1/retrieve/stats`.
5. POST `/v1/retrieve` with the marker hash.
6. Repeat with `HEADROOM_CCR_TTL_SECONDS=1`, wait 1.5 seconds, and
retrieve the same way to verify explicit expiration reporting.

Observed result:

```json
{
  "long_ttl": {
    "ccr_hash": "b473e632aa47",
    "retrieve_status": 200,
    "retrieved_content_has_result_199": true,
    "stats_default_ttl_seconds": 7200,
    "stats_entry_count": 1,
    "ttl_seconds": 7200
  },
  "short_ttl_expired": {
    "ccr_hash": "b473e632aa47",
    "retrieve_detail": "Entry expired (CCR TTL: 1 seconds; age: 2 seconds)",
    "retrieve_status": 404,
    "stats_default_ttl_seconds": 1,
    "stats_entry_count": 1,
    "ttl_seconds": 1
  }
}
```

What I did not test:

- A live OpenRouter/OpenAI/Anthropic upstream request.
- A durable/non-memory CCR backend.
- A literal 5+ minute wall-clock wait; the process E2E used `7200` to
prove configured retention and `1` second to prove expiration behavior
quickly.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

## Test Output

```bash
UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy pytest tests/test_compression_store.py tests/test_proxy_ccr.py tests/test_ccr_response_handler.py tests/test_ccr_response_handler_extra.py
# 152 passed, 2 warnings in 12.87s

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy pytest tests/test_adapter_hooks.py tests/test_toin_feedback.py tests/test_toin_fixes.py
# 55 passed, 13 skipped, 1 warning in 0.53s

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy ruff check .
# All checks passed!

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy ruff format --check .
# 776 files already formatted

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy mypy headroom
# Success: no issues found in 346 source files
```

Existing warnings observed in the targeted tests were unrelated to this
change:

- AnthropicProvider tiktoken approximation warning in proxy CCR tests.
- Deprecated `ToolIntelligenceNetwork.get_recommendation()` warning in
existing TOIN assertions.

## 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
- [x] 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 have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

Not applicable.

## Additional Notes

No new dependencies. Default behavior remains 300 seconds unless
`HEADROOM_CCR_TTL_SECONDS` is set.

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 * **proxy:** per-project savings breakdown on the dashboard for all wrapped agents — Claude Code, Codex, aider, Copilot, and Cursor ([#802](https://github.com/chopratejas/headroom/issues/802)). `headroom wrap claude`/`codex` tag requests with an `X-Headroom-Project` header (launch-directory name); `wrap aider`/`copilot`/`cursor` — whose clients cannot send custom headers — use a `/p/<name>` base-URL prefix the proxy strips. Savings are aggregated per project (persisted, schema v3 with transparent v2 migration), exposed as `savings.per_project` in `/stats` and `projects` in `/stats-history`, and shown in a Per-Project Savings dashboard table.
 
+### Bug Fixes
+
+* **ccr:** make retrieval store TTL configurable with `HEADROOM_CCR_TTL_SECONDS`, expose the effective TTL in `/v1/retrieve/stats`, and distinguish expired retrievals from missing hashes.
+
 
 ## [0.24.0](https://github.com/chopratejas/headroom/compare/v0.23.0...v0.24.0) (2026-06-08)
 

README.md

@@ -52,7 +52,7 @@ Headroom compresses everything your AI agent reads — tool outputs, logs, RAG c
 - **MCP server** — `headroom_compress`, `headroom_retrieve`, `headroom_stats` for any MCP client
 - **Cross-agent memory** — shared store across Claude, Codex, Gemini, auto-dedup
 - **`headroom learn`** — mines failed sessions, writes corrections to `CLAUDE.md` / `AGENTS.md`
-- **Reversible (CCR)** — originals never deleted; LLM retrieves on demand
+- **Reversible (CCR)** — originals are cached for retrieval on demand
 
 ## How it works (30 seconds)
 
@@ -159,7 +159,7 @@ Platform support note: macOS auth reuse via Copilot CLI Keychain storage has bee
 **Great fit if you…**
 - run AI coding agents daily and want savings without changing your code
 - work across multiple agents and want shared memory
-- need reversible compression — originals always retrievable via CCR
+- need reversible compression — originals are retrievable via CCR within the configured TTL
 
 **Skip it if you…**
 - only use a single provider's native compaction and don't need cross-agent memory
@@ -287,4 +287,4 @@ Devcontainers in `.devcontainer/` (default + `memory-stack` with Qdrant & Neo4j)
 
 ## License
 
-Apache 2.0 — see [LICENSE](LICENSE).
+Apache 2.0 — see [LICENSE](LICENSE).

docs/content/docs/ccr.mdx

@@ -162,6 +162,18 @@ response = client.chat.completions.create(
 </Tab>
 </Tabs>
 
+## Retention
+
+Proxy CCR originals are kept for 300 seconds by default. For longer autonomous
+agent runs, set `HEADROOM_CCR_TTL_SECONDS` before starting the proxy:
+
+```bash
+HEADROOM_CCR_TTL_SECONDS=7200 headroom proxy
+```
+
+Check the effective setting at `/v1/retrieve/stats` under
+`store.default_ttl_seconds`.
+
 ## Message-level CCR
 
 > **Retired:** The "Message-level CCR via IntelligentContext" feature (where `IntelligentContext` would store dropped messages in CCR with a retrieval marker) was part of the `IntelligentContextConfig` API that was removed in 0.9.x. Context management is now handled automatically by the pipeline without a separate configurable IntelligentContext stage. Tool-output CCR via SmartCrusher and ContentRouter remains fully supported.

headroom/cache/compression_store.py

@@ -53,6 +53,9 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_CCR_TTL_SECONDS = 300
+CCR_TTL_SECONDS_ENV = "HEADROOM_CCR_TTL_SECONDS"
+
 _RETRIEVAL_LOG_PREVIEW_CHARS = 4096
 _SECRET_KEY_VALUE_RE = re.compile(
     r"(?i)\b([A-Z0-9_-]*(?:API[_-]?KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|AUTH)[A-Z0-9_-]*)"
@@ -62,6 +65,48 @@ _AUTH_VALUE_RE = re.compile(r"(?i)\b(Bearer|Basic)\s+[A-Za-z0-9._~+/=-]{12,}")
 _API_KEY_VALUE_RE = re.compile(r"\bsk-[A-Za-z0-9_-]{12,}\b")
 
 
+def _get_env_default_ttl_seconds() -> int:
+    raw_value = os.environ.get(CCR_TTL_SECONDS_ENV)
+    if raw_value is None or not raw_value.strip():
+        return DEFAULT_CCR_TTL_SECONDS
+
+    try:
+        ttl_seconds = int(raw_value)
+    except ValueError:
+        logger.warning(
+            "%s must be a positive integer number of seconds, got %r; using %s",
+            CCR_TTL_SECONDS_ENV,
+            raw_value,
+            DEFAULT_CCR_TTL_SECONDS,
+        )
+        return DEFAULT_CCR_TTL_SECONDS
+
+    if ttl_seconds <= 0:
+        logger.warning(
+            "%s must be greater than 0, got %s; using %s",
+            CCR_TTL_SECONDS_ENV,
+            ttl_seconds,
+            DEFAULT_CCR_TTL_SECONDS,
+        )
+        return DEFAULT_CCR_TTL_SECONDS
+
+    return ttl_seconds
+
+
+def format_retrieval_miss_detail(status: dict[str, Any]) -> str:
+    """Return an operator-facing miss reason for CCR retrieval failures."""
+    default_ttl = status.get("default_ttl_seconds", DEFAULT_CCR_TTL_SECONDS)
+    ttl_seconds = status.get("ttl_seconds", default_ttl)
+
+    if status.get("status") == "expired":
+        age_seconds = status.get("age_seconds")
+        if isinstance(age_seconds, (int, float)):
+            return f"Entry expired (CCR TTL: {ttl_seconds} seconds; age: {age_seconds:.0f} seconds)"
+        return f"Entry expired (CCR TTL: {ttl_seconds} seconds)"
+
+    return f"Entry not found (CCR TTL: {default_ttl} seconds)"
+
+
 def _redact_retrieval_log_payload(payload: str) -> str:
     redacted = _SECRET_KEY_VALUE_RE.sub(r"\1\2\3[REDACTED]", payload)
     redacted = _AUTH_VALUE_RE.sub(r"\1 [REDACTED]", redacted)
@@ -95,7 +140,7 @@ class CompressionEntry:
     tool_call_id: str | None
     query_context: str | None
     created_at: float
-    ttl: int = 300  # 5 minutes default
+    ttl: int = DEFAULT_CCR_TTL_SECONDS
 
     # TOIN integration: Store the tool signature hash for retrieval correlation
     # This MUST match the hash used by SmartCrusher when recording compression
@@ -146,7 +191,7 @@ class CompressionStore:
     Design principles:
     - Zero external dependencies (pure Python)
     - Thread-safe for concurrent access
-    - TTL-based expiration (default 5 minutes)
+    - TTL-based expiration (default 300 seconds, env-configurable)
     - LRU-style eviction when capacity is reached
     - Built-in BM25 search for filtering
     """
@@ -154,7 +199,7 @@ class CompressionStore:
     def __init__(
         self,
         max_entries: int = 1000,
-        default_ttl: int = 300,
+        default_ttl: int = DEFAULT_CCR_TTL_SECONDS,
         enable_feedback: bool = True,
         backend: CompressionStoreBackend | None = None,
     ):
@@ -162,7 +207,7 @@ class CompressionStore:
 
         Args:
             max_entries: Maximum number of entries to store.
-            default_ttl: Default TTL in seconds (5 minutes).
+            default_ttl: Default TTL in seconds.
             enable_feedback: Whether to track retrieval events.
             backend: Storage backend to use. Defaults to InMemoryBackend.
                      Custom backends can be passed for persistence (MongoDB, Redis).
@@ -192,6 +237,11 @@ class CompressionStore:
         # BM25 scorer for search
         self._scorer = BM25Scorer()
 
+    @property
+    def default_ttl_seconds(self) -> int:
+        """Default TTL applied to new entries when callers do not override it."""
+        return self._default_ttl
+
     def store(
         self,
         original: str,
@@ -730,6 +780,42 @@ class CompressionStore:
                 return False
             return True
 
+    def get_entry_status(
+        self,
+        hash_key: str,
+        *,
+        clean_expired: bool = False,
+    ) -> dict[str, Any]:
+        """Return availability and TTL metadata for a stored entry."""
+        now = time.time()
+        with self._lock:
+            entry = self._backend.get(hash_key)
+            if entry is None:
+                return {
+                    "hash": hash_key,
+                    "status": "missing",
+                    "default_ttl_seconds": self._default_ttl,
+                }
+
+            age_seconds = now - entry.created_at
+            expires_at = entry.created_at + entry.ttl
+            expired = age_seconds > entry.ttl
+            status = {
+                "hash": hash_key,
+                "status": "expired" if expired else "available",
+                "ttl_seconds": entry.ttl,
+                "default_ttl_seconds": self._default_ttl,
+                "created_at": entry.created_at,
+                "expires_at": expires_at,
+                "age_seconds": age_seconds,
+            }
+
+            if expired and clean_expired:
+                self._backend.delete(hash_key)
+                self._stale_heap_entries += 1
+
+            return status
+
     def get_stats(self) -> dict[str, Any]:
         """Get store statistics for monitoring."""
         with self._lock:
@@ -748,6 +834,7 @@ class CompressionStore:
             return {
                 "entry_count": self._backend.count(),
                 "max_entries": self._max_entries,
+                "default_ttl_seconds": self._default_ttl,
                 "total_original_tokens": total_original_tokens,
                 "total_compressed_tokens": total_compressed_tokens,
                 "total_retrievals": total_retrievals,
@@ -1134,7 +1221,7 @@ def _create_default_ccr_backend() -> CompressionStoreBackend | None:
 
 def get_compression_store(
     max_entries: int = 1000,
-    default_ttl: int = 300,
+    default_ttl: int | None = None,
     backend: CompressionStoreBackend | None = None,
 ) -> CompressionStore:
     """Get the compression store instance.
@@ -1146,6 +1233,7 @@ def get_compression_store(
     Args:
         max_entries: Maximum entries (only used on first call for global store).
         default_ttl: Default TTL (only used on first call for global store).
+            When omitted, HEADROOM_CCR_TTL_SECONDS overrides the 300-second default.
         backend: Custom storage backend (only used on first call for global store).
                  Defaults to InMemoryBackend if not provided; env backend used if backend is None.
 
@@ -1162,9 +1250,12 @@ def get_compression_store(
             if _compression_store is None:
                 if backend is None:
                     backend = _create_default_ccr_backend()
+                effective_default_ttl = (
+                    default_ttl if default_ttl is not None else _get_env_default_ttl_seconds()
+                )
                 _compression_store = CompressionStore(
                     max_entries=max_entries,
-                    default_ttl=default_ttl,
+                    default_ttl=effective_default_ttl,
                     backend=backend,
                 )
     return _compression_store

headroom/ccr/mcp_server.py

@@ -449,7 +449,7 @@ class HeadroomMCPServer:
             "error": "Content not found. It may have expired or the hash may be incorrect.",
             "hash": hash_key,
             "hint": "Content compressed via headroom_compress is stored for the session. "
-            "Content compressed by the proxy has a shorter TTL (5 minutes).",
+            "Content compressed by the proxy uses the configured CCR TTL.",
         }
 
     async def _retrieve_via_proxy(

headroom/ccr/response_handler.py

@@ -19,7 +19,7 @@ from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import Any
 
-from ..cache.compression_store import get_compression_store
+from ..cache.compression_store import format_retrieval_miss_detail, get_compression_store
 from .tool_injection import CCR_TOOL_NAME, parse_tool_call
 
 logger = logging.getLogger(__name__)
@@ -202,6 +202,28 @@ class CCRResponseHandler:
         store = get_compression_store()
 
         try:
+            get_status = getattr(store, "get_entry_status", None)
+            entry_status = (
+                get_status(ccr_call.hash_key, clean_expired=True) if callable(get_status) else None
+            )
+            if entry_status is not None and entry_status["status"] != "available":
+                content = json.dumps(
+                    {
+                        "error": format_retrieval_miss_detail(entry_status),
+                        "hash": ccr_call.hash_key,
+                        "status": entry_status["status"],
+                        "ttl_seconds": entry_status.get(
+                            "ttl_seconds", entry_status["default_ttl_seconds"]
+                        ),
+                    },
+                    indent=2,
+                )
+                return CCRToolResult(
+                    tool_call_id=ccr_call.tool_call_id,
+                    content=content,
+                    success=False,
+                )
+
             if ccr_call.query:
                 # Search within compressed content
                 results = store.search(ccr_call.hash_key, ccr_call.query)
@@ -241,10 +263,17 @@ class CCRResponseHandler:
                         was_search=False,
                     )
                 else:
+                    miss_status = (
+                        get_status(ccr_call.hash_key, clean_expired=True)
+                        if callable(get_status)
+                        else {"hash": ccr_call.hash_key, "status": "missing"}
+                    )
                     content = json.dumps(
                         {
-                            "error": "Entry not found or expired (TTL: 5 minutes)",
+                            "error": format_retrieval_miss_detail(miss_status),
                             "hash": ccr_call.hash_key,
+                            "status": miss_status["status"],
+                            "ttl_seconds": miss_status.get("ttl_seconds"),
                         },
                         indent=2,
                     )

headroom/config.py

@@ -407,14 +407,14 @@ class CCRConfig:
     - Network effect: retrieval patterns improve compression for all users
 
     GOTCHAS:
-    - Cache has TTL (default 5 min) - retrieval fails after expiration
+    - Cache has TTL (default 300 seconds) - retrieval fails after expiration
     - Memory usage: ~1KB per cached entry
     - Only works with array compression (not string truncation)
     """
 
     enabled: bool = True  # Enable CCR (cache + retrieval markers)
     store_max_entries: int = 1000  # Max entries in compression store
-    store_ttl_seconds: int = 300  # Cache TTL (5 minutes)
+    store_ttl_seconds: int = 300  # Cache TTL in seconds
     inject_retrieval_marker: bool = True  # Add retrieval hint to compressed output
     feedback_enabled: bool = True  # Track retrieval events for learning
     min_items_to_cache: int = 20  # Only cache if original had >= N items

headroom/proxy/server.py

@@ -62,7 +62,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 
 from headroom._version import __version__
 from headroom.cache.compression_feedback import get_compression_feedback
-from headroom.cache.compression_store import get_compression_store
+from headroom.cache.compression_store import format_retrieval_miss_detail, get_compression_store
 from headroom.ccr import (
     CCR_TOOL_NAME,
     # Batch processing
@@ -2506,11 +2506,14 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
 
         store = get_compression_store()
 
+        entry_status = store.get_entry_status(hash_key, clean_expired=True)
+        if entry_status["status"] != "available":
+            raise HTTPException(
+                status_code=404,
+                detail=format_retrieval_miss_detail(entry_status),
+            )
+
         if query:
-            if not store.exists(hash_key, clean_expired=True):
-                raise HTTPException(
-                    status_code=404, detail="Entry not found or expired (TTL: 5 minutes)"
-                )
             # Search within cached content
             results = store.search(hash_key, query)
             return {
@@ -2533,7 +2536,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                     "retrieval_count": entry.retrieval_count,
                 }
             raise HTTPException(
-                status_code=404, detail="Entry not found or expired (TTL: 5 minutes)"
+                status_code=404,
+                detail=format_retrieval_miss_detail(
+                    store.get_entry_status(hash_key, clean_expired=True)
+                ),
             )
 
     @app.get("/v1/retrieve/stats")
@@ -2819,10 +2825,15 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
     async def ccr_retrieve_get(hash_key: str, query: str | None = None):
         """GET version of CCR retrieve for easier testing."""
         store = get_compression_store()
+        entry_status = store.get_entry_status(hash_key, clean_expired=True)
+
+        if entry_status["status"] != "available":
+            raise HTTPException(
+                status_code=404,
+                detail=format_retrieval_miss_detail(entry_status),
+            )
 
         if query:
-            if not store.exists(hash_key, clean_expired=True):
-                raise HTTPException(status_code=404, detail="Entry not found or expired")
             results = store.search(hash_key, query)
             return {
                 "hash": hash_key,
@@ -2842,7 +2853,12 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                     "tool_name": entry.tool_name,
                     "retrieval_count": entry.retrieval_count,
                 }
-            raise HTTPException(status_code=404, detail="Entry not found or expired")
+            raise HTTPException(
+                status_code=404,
+                detail=format_retrieval_miss_detail(
+                    store.get_entry_status(hash_key, clean_expired=True)
+                ),
+            )
 
     # CCR Tool Call Handler - for agent frameworks to call when LLM uses headroom_retrieve
     @app.post("/v1/retrieve/tool_call")
@@ -2896,8 +2912,16 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
 
         # Perform retrieval
         store = get_compression_store()
+        entry_status = store.get_entry_status(hash_key, clean_expired=True)
 
-        if query:
+        if entry_status["status"] != "available":
+            retrieval_data = {
+                "error": format_retrieval_miss_detail(entry_status),
+                "hash": hash_key,
+                "status": entry_status["status"],
+                "ttl_seconds": entry_status.get("ttl_seconds", entry_status["default_ttl_seconds"]),
+            }
+        elif query:
             results = store.search(hash_key, query)
             retrieval_data = {
                 "hash": hash_key,
@@ -2915,9 +2939,14 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                     "compressed_item_count": entry.compressed_item_count,
                 }
             else:
+                miss_status = store.get_entry_status(hash_key, clean_expired=True)
                 retrieval_data = {
-                    "error": "Entry not found or expired (TTL: 5 minutes)",
+                    "error": format_retrieval_miss_detail(miss_status),
                     "hash": hash_key,
+                    "status": miss_status["status"],
+                    "ttl_seconds": miss_status.get(
+                        "ttl_seconds", miss_status["default_ttl_seconds"]
+                    ),
                 }
 
         # Format tool result for provider

tests/test_compression_store.py

@@ -27,6 +27,8 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from headroom.cache.compression_store import (
+    CCR_TTL_SECONDS_ENV,
+    DEFAULT_CCR_TTL_SECONDS,
     CompressionEntry,
     CompressionStore,
     RetrievalEvent,
@@ -128,6 +130,58 @@ def test_search_logs_retrieved_payload_preview():
     assert events[0]["payload_truncated"] is False
 
 
+def test_global_store_uses_env_default_ttl(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv(CCR_TTL_SECONDS_ENV, "7200")
+
+    store = get_compression_store()
+    hash_key = store.store(original="long-running agent payload", compressed="payload")
+    entry = store.retrieve(hash_key)
+
+    assert entry is not None
+    assert entry.ttl == 7200
+    assert store.get_stats()["default_ttl_seconds"] == 7200
+
+
+def test_global_store_invalid_env_ttl_falls_back(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv(CCR_TTL_SECONDS_ENV, "0")
+
+    store = get_compression_store()
+    hash_key = store.store(original="payload", compressed="payload")
+    entry = store.retrieve(hash_key)
+
+    assert entry is not None
+    assert entry.ttl == DEFAULT_CCR_TTL_SECONDS
+    assert store.get_stats()["default_ttl_seconds"] == DEFAULT_CCR_TTL_SECONDS
+
+
+def test_explicit_global_store_ttl_overrides_env(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv(CCR_TTL_SECONDS_ENV, "7200")
+
+    store = get_compression_store(default_ttl=60)
+    hash_key = store.store(original="payload", compressed="payload")
+    entry = store.retrieve(hash_key)
+
+    assert entry is not None
+    assert entry.ttl == 60
+    assert store.get_stats()["default_ttl_seconds"] == 60
+
+
+def test_entry_status_reports_expiration_metadata():
+    store = CompressionStore(default_ttl=1)
+
+    with patch("headroom.cache.compression_store.time.time", return_value=1000.0):
+        hash_key = store.store(original="payload", compressed="payload")
+
+    with patch("headroom.cache.compression_store.time.time", return_value=1002.0):
+        status = store.get_entry_status(hash_key, clean_expired=True)
+
+    assert status["status"] == "expired"
+    assert status["ttl_seconds"] == 1
+    assert status["age_seconds"] == 2
+    assert status["expires_at"] == 1001.0
+    assert store.exists(hash_key) is False
+
+
 # =============================================================================
 # Fixtures
 # =============================================================================

tests/test_proxy_ccr.py

@@ -4,6 +4,7 @@ These tests verify the /v1/retrieve endpoints work correctly.
 """
 
 import json
+from unittest.mock import patch
 
 import pytest
 
@@ -65,7 +66,23 @@ class TestCCRRetrieveEndpoint:
         """Request with nonexistent hash should return 404."""
         response = client.post("/v1/retrieve", json={"hash": "nonexistent123"})
         assert response.status_code == 404
-        assert "not found or expired" in response.json()["detail"]
+        assert "Entry not found" in response.json()["detail"]
+        assert "CCR TTL: 300 seconds" in response.json()["detail"]
+
+    def test_retrieve_expired_hash_reports_expiration_detail(self, client):
+        """Expired entries report expiration separately from missing hashes."""
+        store = get_compression_store(default_ttl=1)
+        with patch("headroom.cache.compression_store.time.time", return_value=1000.0):
+            hash_key = store.store(original="payload", compressed="payload")
+
+        with patch("headroom.cache.compression_store.time.time", return_value=1002.0):
+            response = client.post("/v1/retrieve", json={"hash": hash_key})
+
+        assert response.status_code == 404
+        detail = response.json()["detail"]
+        assert "Entry expired" in detail
+        assert "CCR TTL: 1 seconds" in detail
+        assert "age: 2 seconds" in detail
 
     def test_retrieve_full_content(self, client):
         """Full retrieval returns original content."""
@@ -246,8 +263,19 @@ class TestCCRStatsEndpoint:
         data = response.json()
         assert "store" in data
         assert data["store"]["entry_count"] == 0
+        assert data["store"]["default_ttl_seconds"] == 300
         assert "recent_retrievals" in data
 
+    def test_stats_exposes_env_configured_ttl(self, client, monkeypatch):
+        """Stats expose the effective CCR TTL configured through env."""
+        reset_compression_store()
+        monkeypatch.setenv("HEADROOM_CCR_TTL_SECONDS", "7200")
+
+        response = client.get("/v1/retrieve/stats")
+
+        assert response.status_code == 200
+        assert response.json()["store"]["default_ttl_seconds"] == 7200
+
     def test_stats_with_entries(self, client):
         """Stats reflect store contents."""
         store = get_compression_store()