fix(ccr): make expired retrieve misses terminal (#1781)

## Description

Expired CCR hashes currently come back through `headroom_retrieve` as
the same generic missing-content error used for typos and never-stored
hashes. That leaves agents with no terminal signal, so they can retry a
dead hash instead of rerunning the source command or rereading the
source file. This change uses the cache store's existing TTL status
metadata before the MCP retrieval path loses that distinction, then
returns expired-hash guidance only when the local store proves the entry
existed and expired.

Closes #1776

## 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

- Uses CCR store status metadata to distinguish expired local hashes
from never-stored hashes in the MCP retrieval path.
- Keeps proxy fallback and successful local retrieval behavior
unchanged.
- Adds focused regression coverage for expired stored hashes, the
status-to-retrieve TTL boundary, proxy fallback preservation, and
missing-hash negative space.

## Testing

- [x] Unit tests pass (`uv run pytest tests/test_ccr_mcp_server.py -q`)
- [x] Linting passes (`uv run ruff check headroom/ccr/mcp_server.py
tests/test_ccr_mcp_server.py`)
- [ ] Type checking passes (`uv run mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
Base pytest:
FAILED tests\test_ccr_mcp_server.py::test_mcp_retrieve_expired_hash_returns_terminal_guidance
E   KeyError: 'status'

Head pytest:
tests\test_ccr_mcp_server.py ...s..........                              [100%]
13 passed, 1 skipped in 0.35s

Ruff:
All checks passed!
```

## Real Behavior Proof

- Environment: Windows, focused local pytest through the headless
runner.
- Exact command / steps: Store a CCR entry with a short TTL, advance
beyond expiry, call `HeadroomMCPServer._retrieve_content(hash)`, force a
second entry to cross TTL between status inspection and `retrieve()`,
stub a proxy-backed retrieval for local misses, then call the same
method with a never-stored hash and no proxy hit.
- Observed result: The already-expired hash and the hash that expires
during retrieval both return terminal expired guidance with `status:
expired`; missing and expired local hashes still return proxy data when
the proxy fallback succeeds; a never-stored hash with no proxy hit still
returns the generic missing-hash error and no expired status.
- Not tested: Full suite, live agent retry behavior, and live external
proxy-backed retrieval.

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

## Additional Notes

Documentation and changelog are left unchecked because this is a narrow
MCP error-shape fix and Headroom's changelog is generated from
conventional commits.

headroom/ccr/mcp_server.py

@@ -34,6 +34,7 @@ from typing import Any
 
 from headroom import paths as _paths
 from headroom import savings_ledger
+from headroom.cache.compression_store import format_retrieval_miss_detail
 
 # fcntl is Unix-only; on Windows we skip file locking (stats are best-effort).
 # Keep the module typed as Any so Windows mypy runs don't try to resolve Unix-only attrs.
@@ -440,7 +441,9 @@ class HeadroomMCPServer:
         """
         # Check local store first
         store = self._get_local_store()
+        entry_status = store.get_entry_status(hash_key, clean_expired=False)
         entry = store.retrieve(hash_key)
+        expired_entry_status = None
         if entry:
             self._stats.record_retrieval(hash_key)
             return {
@@ -451,6 +454,19 @@ class HeadroomMCPServer:
                 "compressed_item_count": entry.compressed_item_count,
                 "retrieval_count": entry.retrieval_count,
             }
+        if entry_status.get("status") == "expired":
+            expired_entry_status = entry_status
+        elif entry_status.get("status") == "available":
+            created_at = entry_status.get("created_at")
+            ttl_seconds = entry_status.get("ttl_seconds")
+            if isinstance(created_at, (int, float)) and isinstance(ttl_seconds, (int, float)):
+                age_seconds = time.time() - created_at
+                if age_seconds > ttl_seconds:
+                    expired_entry_status = {
+                        **entry_status,
+                        "status": "expired",
+                        "age_seconds": age_seconds,
+                    }
 
         # Fall back to proxy if available
         if self.check_proxy and HTTPX_AVAILABLE:
@@ -463,6 +479,26 @@ class HeadroomMCPServer:
             except Exception:
                 pass  # Proxy unavailable, that's fine
 
+        if expired_entry_status:
+            ttl_seconds = expired_entry_status.get(
+                "ttl_seconds",
+                expired_entry_status["default_ttl_seconds"],
+            )
+            return {
+                "error": (
+                    f"{format_retrieval_miss_detail(expired_entry_status)}. "
+                    "Do not retry the same hash. Re-run the source command or re-read the source file."
+                ),
+                "hash": hash_key,
+                "status": "expired",
+                "ttl_seconds": ttl_seconds,
+                "age_seconds": expired_entry_status.get("age_seconds"),
+                "hint": (
+                    "Use the source of truth to regenerate fresh content. "
+                    "Re-run the command or re-read the file."
+                ),
+            }
+
         return {
             "error": "Content not found. It may have expired or the hash may be incorrect.",
             "hash": hash_key,

tests/test_ccr_mcp_server.py

@@ -5,6 +5,7 @@ import json
 
 import pytest
 
+from headroom.cache import compression_store as compression_store_module
 from headroom.cache.compression_store import (
     get_compression_store,
     reset_compression_store,
@@ -105,11 +106,132 @@ def test_mcp_retrieve_returns_full_content(fresh_store) -> None:
     assert result["original_content"] == original
 
 
+def test_mcp_retrieve_expired_hash_returns_terminal_guidance(
+    monkeypatch,
+    fresh_store,
+) -> None:
+    """An expired local hash should say it expired and tell the agent to stop retrying."""
+    current_time = [1000.0]
+
+    def fake_time() -> float:
+        return current_time[0]
+
+    monkeypatch.setattr(mcp_server.time, "time", fake_time)
+    monkeypatch.setattr(compression_store_module.time, "time", fake_time)
+
+    store = get_compression_store()
+    hash_key = store.store("expired content", "<<small>>", ttl=1)
+    current_time[0] = 1002.0
+
+    server = mcp_server.HeadroomMCPServer(check_proxy=False)
+    result = asyncio.run(server._retrieve_content(hash_key))
+
+    assert result["status"] == "expired"
+    assert result["ttl_seconds"] == 1
+    assert result["age_seconds"] == pytest.approx(2.0)
+    assert "Entry expired" in result["error"]
+    assert "do not retry the same hash" in result["error"].lower()
+    assert "re-run the command" in result["hint"].lower()
+
+
+def test_mcp_retrieve_hash_expiring_during_lookup_returns_terminal_guidance(
+    monkeypatch,
+    fresh_store,
+) -> None:
+    phase = "store"
+    status_seen = False
+
+    def fake_time() -> float:
+        if phase == "store":
+            return 1000.0
+        return 1001.1 if status_seen else 1000.5
+
+    monkeypatch.setattr(mcp_server.time, "time", fake_time)
+    monkeypatch.setattr(compression_store_module.time, "time", fake_time)
+
+    store = get_compression_store()
+    hash_key = store.store("expired during retrieve", "<<small>>", ttl=1)
+    phase = "retrieve"
+
+    original_get_entry_status = store.get_entry_status
+    original_retrieve = store.retrieve
+
+    def get_entry_status_then_expire(*args, **kwargs):
+        nonlocal status_seen
+        result = original_get_entry_status(*args, **kwargs)
+        status_seen = True
+        return result
+
+    monkeypatch.setattr(store, "get_entry_status", get_entry_status_then_expire)
+    monkeypatch.setattr(store, "retrieve", original_retrieve)
+
+    server = mcp_server.HeadroomMCPServer(check_proxy=False)
+    result = asyncio.run(server._retrieve_content(hash_key))
+
+    assert result["status"] == "expired"
+    assert result["ttl_seconds"] == 1
+    assert result["age_seconds"] == pytest.approx(1.1)
+    assert "Entry expired" in result["error"]
+    assert "do not retry the same hash" in result["error"].lower()
+
+
+def test_mcp_retrieve_missing_local_hash_can_still_hit_proxy(
+    monkeypatch,
+    fresh_store,
+) -> None:
+    monkeypatch.setattr(mcp_server, "HTTPX_AVAILABLE", True)
+    server = mcp_server.HeadroomMCPServer(check_proxy=True)
+
+    async def retrieve_via_proxy(hash_key: str) -> dict[str, object]:
+        return {"hash": hash_key, "original_content": "from proxy"}
+
+    server._retrieve_via_proxy = retrieve_via_proxy
+
+    result = asyncio.run(server._retrieve_content("proxy_hash"))
+
+    assert result["source"] == "proxy"
+    assert result["hash"] == "proxy_hash"
+    assert result["original_content"] == "from proxy"
+
+
+def test_mcp_retrieve_expired_local_hash_can_still_hit_proxy(
+    monkeypatch,
+    fresh_store,
+) -> None:
+    current_time = [1000.0]
+
+    def fake_time() -> float:
+        return current_time[0]
+
+    monkeypatch.setattr(mcp_server, "HTTPX_AVAILABLE", True)
+    monkeypatch.setattr(mcp_server.time, "time", fake_time)
+    monkeypatch.setattr(compression_store_module.time, "time", fake_time)
+
+    store = get_compression_store()
+    hash_key = store.store("expired local content", "<<small>>", ttl=1)
+    current_time[0] = 1002.0
+
+    server = mcp_server.HeadroomMCPServer(check_proxy=True)
+
+    async def retrieve_via_proxy(proxy_hash_key: str) -> dict[str, object]:
+        return {"hash": proxy_hash_key, "original_content": "from proxy"}
+
+    server._retrieve_via_proxy = retrieve_via_proxy
+
+    result = asyncio.run(server._retrieve_content(hash_key))
+
+    assert result["source"] == "proxy"
+    assert result["hash"] == hash_key
+    assert result["original_content"] == "from proxy"
+
+
 def test_mcp_retrieve_missing_hash_still_errors(fresh_store) -> None:
-    """A genuinely missing hash must still report "Content not found"."""
+    """A never-stored hash must stay on the generic missing path, not expired guidance."""
     server = mcp_server.HeadroomMCPServer(check_proxy=False)
     result = asyncio.run(server._retrieve_content("nonexistent_hash"))
-    assert "Content not found" in result.get("error", "")
+    assert result.get("status") is None
+    assert result["error"] == "Content not found. It may have expired or the hash may be incorrect."
+    assert "do not retry the same hash" not in result.get("hint", "").lower()
 
 
 def test_handle_stats_session_output_is_window_scoped() -> None: