Add SharedContext for multi-agent, rewrite README, fix proxy cleanup

- SharedContext: compressed inter-agent context sharing via put()/get() over existing CCR compression pipeline. Zero new dependencies. - README rewrite: lead with "any agent" positioning, not just coding agents. Add headroom wrap, SharedContext, MCP tools to Quick Start. Reorder integration table: universal first, coding shortcuts last. Update compression pipeline references (LLMLingua → Kompress). - Fix proxy cleanup in headroom wrap: don't kill shared proxy if other clients are still using it (was orphaning terminals 2-N). - New docs: docs/shared-context.md
2026-08-27 14:17:10 -04:00 · 2026-03-17 16:31:04 -07:00 · 2026-03-17 16:31:04 -07:00 · 6ab448800b
commit 6ab448800b
parent 891d4d0cad
6 changed files with 606 additions and 33 deletions
--- a/README.md
+++ b/README.md
@ -5,7 +5,9 @@
  </p>
  <p align="center">
    Every tool call, DB query, file read, and RAG retrieval your agent makes is 70-95% boilerplate.<br>
-    Headroom compresses it away before it hits the model.
+    Headroom compresses it away before it hits the model.<br><br>
+    Works with <b>any agent</b> — coding agents (Claude Code, Codex, Cursor, Aider), custom agents<br>
+    (LangChain, LangGraph, CrewAI, Agno, OpenAI Agents SDK), or your own Python code.
  </p>
 </p>

@ -39,16 +41,18 @@

 ```
 Your Agent / App
+  (coding agents, customer support bots, RAG pipelines,
+   data analysis agents, research agents, any LLM app)
      │
      │  tool calls, logs, DB reads, RAG results, file reads, API responses
      ▼
-   Headroom  ← transparent proxy, no code changes needed
+   Headroom  ← proxy, Python library, or framework integration
      │
      ▼
 LLM Provider  (OpenAI, Anthropic, Google, Bedrock, 100+ via LiteLLM)
 ```

-Headroom sits between your application and the LLM provider. It intercepts requests, compresses the context, and forwards an optimized prompt. Your app doesn't change — just point it at Headroom.
+Headroom sits between your application and the LLM provider. It intercepts requests, compresses the context, and forwards an optimized prompt. Use it as a transparent proxy (zero code changes), a Python function (`compress()`), or a framework integration (LangChain, LiteLLM, Agno).

 ### What gets compressed

@ -69,23 +73,7 @@ Headroom optimizes any data your agent injects into a prompt:
 pip install "headroom-ai[all]"
 ```

-### Proxy (zero code changes)
-
-```bash
-headroom proxy --port 8787
-```
-
-```bash
-# Claude Code — just set the base URL
-ANTHROPIC_BASE_URL=http://localhost:8787 claude
-
-# Cursor, Continue, any OpenAI-compatible tool
-OPENAI_BASE_URL=http://localhost:8787/v1 cursor
-```
-
-Works with any language, any tool, any framework. One env var. **[Proxy docs](docs/proxy.md)**
-
-### Python: One function
+### Any agent — one function

 ```python
 from headroom import compress
@ -95,21 +83,68 @@ response = client.messages.create(model="claude-sonnet-4-5-20250929", messages=r
 print(f"Saved {result.tokens_saved} tokens ({result.compression_ratio:.0%})")
 ```

-Works with any Python LLM client — Anthropic, OpenAI, LiteLLM, httpx, anything.
+Works with any Python LLM client — Anthropic, OpenAI, LiteLLM, Bedrock, httpx, anything. Works with any agent framework — LangChain, LangGraph, CrewAI, Agno, OpenAI Agents SDK, or your own code.

-### Already have a proxy or gateway?
+### Any agent — proxy (zero code changes)

-You don't need to replace it. Drop Headroom into your existing stack:
+```bash
+headroom proxy --port 8787
+```
+
+```bash
+# Point any LLM client at the proxy
+ANTHROPIC_BASE_URL=http://localhost:8787 your-app
+OPENAI_BASE_URL=http://localhost:8787/v1 your-app
+```
+
+Works with any language, any tool, any framework. **[Proxy docs](docs/proxy.md)**
+
+### Coding agents — one command
+
+```bash
+headroom wrap claude       # Starts proxy + launches Claude Code
+headroom wrap codex        # Starts proxy + launches OpenAI Codex CLI
+headroom wrap aider        # Starts proxy + launches Aider
+headroom wrap cursor       # Starts proxy + prints Cursor config
+```
+
+Headroom starts a proxy, points your tool at it, and compresses everything automatically.
+
+### Multi-agent — SharedContext
+
+```python
+from headroom import SharedContext
+
+ctx = SharedContext()
+ctx.put("research", big_agent_output)      # Agent A stores (compressed)
+summary = ctx.get("research")               # Agent B reads (~80% smaller)
+full = ctx.get("research", full=True)       # Agent B gets original if needed
+```
+
+Compress what moves between agents — any framework. **[SharedContext Guide](docs/shared-context.md)**
+
+### MCP Tools (Claude Code, Cursor)
+
+```bash
+headroom mcp install && claude
+```
+
+Gives your AI tool three MCP tools: `headroom_compress`, `headroom_retrieve`, `headroom_stats`. **[MCP Guide](docs/mcp.md)**
+
+### Drop into your existing stack

 | Your setup | Add Headroom | One-liner |
 |------------|-------------|-----------|
+| **Any Python app** | `compress()` | `result = compress(messages, model="gpt-4o")` |
+| **Multi-agent** | SharedContext | `ctx = SharedContext(); ctx.put("key", data)` |
 | **LiteLLM** | Callback | `litellm.callbacks = [HeadroomCallback()]` |
 | **Any Python proxy** | ASGI Middleware | `app.add_middleware(CompressionMiddleware)` |
-| **Any Python app** | `compress()` | `result = compress(messages, model="gpt-4o")` |
 | **Agno agents** | Wrap model | `HeadroomAgnoModel(your_model)` |
 | **LangChain** | Wrap model | `HeadroomChatModel(your_llm)` *(experimental)* |
+| **Claude Code** | Wrap | `headroom wrap claude` |
+| **Codex / Aider** | Wrap | `headroom wrap codex` or `headroom wrap aider` |

-**[Full Integration Guide](docs/integration-guide.md)** — detailed setup for LiteLLM, ASGI middleware, compress(), and every framework.
+**[Full Integration Guide](docs/integration-guide.md)** — detailed setup for every framework.

 ---

@ -194,7 +229,7 @@ Headroom never throws data away. It compresses aggressively, stores the original

 ### Smart Content Detection

-Auto-detects what's in your context — JSON arrays, code, logs, plain text — and routes each to the best compressor. JSON goes to SmartCrusher, code goes through AST-aware compression (Python, JS, Go, Rust, Java, C++), prose goes to LLMLingua-2.
+Auto-detects what's in your context — JSON arrays, code, logs, plain text — and routes each to the best compressor. JSON goes to SmartCrusher, code goes through AST-aware compression (Python, JS, Go, Rust, Java, C++), text goes to Kompress (ModernBERT-based, with `[ml]` extra).

 ### Cache Optimization

@ -226,7 +261,7 @@ Reads your conversation history, finds every failed tool call, correlates it wit
 | **Content Router** | Auto-detects content type, routes to optimal compressor |
 | **SmartCrusher** | Universal JSON compression — arrays of dicts, strings, numbers, mixed types, nested objects |
 | **CodeCompressor** | AST-aware compression for Python, JS, Go, Rust, Java, C++ |
-| **LLMLingua-2** | ML-based 20x text compression |
+| **Kompress** | ModernBERT token compression (replaces LLMLingua-2) |
 | **CCR** | Reversible compression — LLM retrieves originals when needed |
 | **Compression Summaries** | Tells the LLM what was omitted ("3 errors, 12 failures") |
 | **CacheAligner** | Stabilizes prefixes for provider KV cache hits |
@ -236,6 +271,9 @@ Reads your conversation history, finds every failed tool call, correlates it wit
 | **Compression Hooks** | Customize compression with pre/post hooks |
 | **Read Lifecycle** | Detects stale/superseded Read outputs, replaces with CCR markers |
 | **`headroom learn`** | Analyzes past failures, writes project-specific learnings to CLAUDE.md/MEMORY.md |
+| **`headroom wrap`** | One-command setup for Claude Code, Codex, Aider, Cursor |
+| **SharedContext** | Compressed inter-agent context sharing for multi-agent workflows |
+| **MCP Tools** | headroom_compress, headroom_retrieve, headroom_stats for Claude Code/Cursor |

 </details>

@ -272,7 +310,7 @@ Context compression is a new space. Here's how the approaches differ:
  2. ContentRouter           Route each content type:
      │                         → SmartCrusher    (JSON)
      │                         → CodeCompressor  (code)
-      │                         → LLMLingua       (text)
+      │                         → Kompress        (text, with [ml])
      ▼
  3. IntelligentContext      Score-based token fitting
      │
@ -291,7 +329,9 @@ Context compression is a new space. Here's how the approaches differ:

 | Integration | Status | Docs |
 |-------------|--------|------|
+| `headroom wrap claude/codex/aider/cursor` | **Stable** | [Proxy Docs](docs/proxy.md) |
 | `compress()` — one function | **Stable** | [Integration Guide](docs/integration-guide.md) |
+| `SharedContext` — multi-agent | **Stable** | [SharedContext Guide](docs/shared-context.md) |
 | LiteLLM callback | **Stable** | [Integration Guide](docs/integration-guide.md#litellm) |
 | ASGI middleware | **Stable** | [Integration Guide](docs/integration-guide.md#asgi-middleware) |
 | Proxy server | **Stable** | [Proxy Docs](docs/proxy.md) |
@ -345,6 +385,7 @@ Python 3.10+
 | [Memory](docs/memory.md) | Persistent memory |
 | [Agno](docs/agno.md) | Agno agent framework |
 | [MCP](docs/mcp.md) | Context engineering toolkit (compress, retrieve, stats) |
+| [SharedContext](docs/shared-context.md) | Compressed inter-agent context sharing |
 | [Learn](docs/learn.md) | Offline failure learning for coding agents |
 | [Configuration](docs/configuration.md) | All options |

--- a/docs/shared-context.md
+++ b/docs/shared-context.md
@ -0,0 +1,156 @@
+# SharedContext — Compressed Inter-Agent Context Sharing
+
+When agents hand off to each other, context gets replayed in full. SharedContext compresses what moves between agents using Headroom's compression pipeline.
+
+## Quick Start
+
+```python
+from headroom import SharedContext
+
+ctx = SharedContext()
+
+# Agent A stores large output
+ctx.put("research", big_research_output, agent="researcher")
+
+# Agent B gets compressed version (~80% smaller)
+summary = ctx.get("research")
+
+# Agent B needs full details
+full = ctx.get("research", full=True)
+```
+
+## API
+
+### `put(key, content, *, agent=None)`
+
+Store content under a key. Compresses automatically using Headroom's full pipeline (SmartCrusher for JSON, CodeCompressor for code, Kompress for text).
+
+```python
+entry = ctx.put("findings", big_json_output, agent="researcher")
+
+entry.original_tokens     # 20,000
+entry.compressed_tokens   # 4,000
+entry.savings_percent     # 80.0
+entry.transforms          # ["router:json:0.20"]
+```
+
+### `get(key, *, full=False)`
+
+Retrieve content. Returns compressed version by default, original with `full=True`.
+
+```python
+compressed = ctx.get("findings")           # 4K tokens
+original = ctx.get("findings", full=True)  # 20K tokens
+missing = ctx.get("nonexistent")           # None
+```
+
+### `get_entry(key)`
+
+Get the full `ContextEntry` with metadata.
+
+```python
+entry = ctx.get_entry("findings")
+entry.key                # "findings"
+entry.agent              # "researcher"
+entry.original_tokens    # 20000
+entry.compressed_tokens  # 4000
+entry.savings_percent    # 80.0
+entry.timestamp          # 1710000000.0
+entry.transforms         # ["router:json:0.20"]
+```
+
+### `keys()`
+
+List all non-expired keys.
+
+### `stats()`
+
+Aggregated stats across all entries.
+
+```python
+stats = ctx.stats()
+stats.entries                  # 3
+stats.total_original_tokens    # 60000
+stats.total_compressed_tokens  # 12000
+stats.total_tokens_saved       # 48000
+stats.savings_percent          # 80.0
+```
+
+### `clear()`
+
+Remove all entries.
+
+## Configuration
+
+```python
+ctx = SharedContext(
+    model="claude-sonnet-4-5-20250929",  # For token counting
+    ttl=3600,                             # 1 hour (default)
+    max_entries=100,                       # Evicts oldest when full
+)
+```
+
+## Framework Examples
+
+### CrewAI
+
+```python
+from headroom import SharedContext
+
+ctx = SharedContext()
+
+# After researcher task
+ctx.put("findings", researcher_task.output.raw)
+
+# Coder task gets compressed context
+coder_context = ctx.get("findings")
+```
+
+### LangGraph
+
+```python
+from headroom import SharedContext
+
+ctx = SharedContext()
+
+def researcher_node(state):
+    result = do_research()
+    ctx.put("research", result)
+    return {"research_summary": ctx.get("research")}
+
+def coder_node(state):
+    # Compressed summary in state, full details on demand
+    full = ctx.get("research", full=True)
+    return {"code": write_code(full)}
+```
+
+### OpenAI Agents SDK
+
+```python
+from headroom import SharedContext
+
+ctx = SharedContext()
+
+def compress_handoff(messages):
+    for msg in messages:
+        if len(msg.content) > 1000:
+            ctx.put(msg.id, msg.content)
+            msg.content = ctx.get(msg.id)
+    return messages
+
+handoff(agent=coder, input_filter=compress_handoff)
+```
+
+### Any Framework
+
+SharedContext is framework-agnostic. It's just `put()` and `get()`. Use it wherever context moves between agents.
+
+## How It Works
+
+Under the hood, `put()` calls `headroom.compress()` (the same pipeline used by the proxy) and stores the original in memory. `get()` returns the compressed version. `get(full=True)` returns the original.
+
+- JSON arrays → SmartCrusher (70-95% compression)
+- Code → CodeCompressor (AST-aware, with `[code]` extra)
+- Text → Kompress (ModernBERT, with `[ml]` extra) or passthrough
+- Entries expire after TTL (default 1 hour)
+- Oldest entries evicted when max_entries reached
--- a/headroom/init.py
+++ b/headroom/init.py
@ -237,8 +237,13 @@ __all__ = [
    "CompressionHooks",
    "CompressContext",
    "CompressEvent",
+    # Shared context
+    "SharedContext",
 ]

 # One-function compression API
 from headroom.compress import CompressResult, compress  # noqa: E402
 from headroom.hooks import CompressContext, CompressEvent, CompressionHooks  # noqa: E402
+
+# Shared context for multi-agent workflows
+from headroom.shared_context import SharedContext  # noqa: E402
--- a/headroom/cli/wrap.py
+++ b/headroom/cli/wrap.py
@ -233,12 +233,36 @@ def _ensure_proxy(port: int, no_proxy: bool) -> subprocess.Popen | None:
        return None


-def _make_cleanup(proxy_proc_holder: list) -> Any:
-    """Create a cleanup function that terminates the proxy on exit."""
+def _make_cleanup(proxy_proc_holder: list, port: int = 8787) -> Any:
+    """Create a cleanup function that terminates the proxy on exit.
+
+    Only kills the proxy if no other headroom-wrapped clients are using it.
+    Checks by looking for other processes with ANTHROPIC_BASE_URL or
+    OPENAI_BASE_URL pointing at our port.
+    """
+
+    def _other_clients_exist() -> bool:
+        """Check if other processes are using this proxy."""
+        try:
+            # Count headroom wrap processes (excluding ourselves)
+            result = subprocess.run(
+                ["pgrep", "-f", f"127.0.0.1:{port}"],
+                capture_output=True,
+                text=True,
+            )
+            pids = [p.strip() for p in result.stdout.strip().split("\n") if p.strip()]
+            my_pid = str(os.getpid())
+            other_pids = [p for p in pids if p != my_pid]
+            return len(other_pids) > 0
+        except Exception:
+            return False  # If we can't check, assume no others

    def cleanup(signum: int | None = None, frame: Any = None) -> None:
        proc = proxy_proc_holder[0] if proxy_proc_holder else None
        if proc and proc.poll() is None:
+            if _other_clients_exist():
+                # Other clients still using the proxy — leave it running
+                return
            proc.terminate()
            try:
                proc.wait(timeout=5)
@ -259,7 +283,7 @@ def _launch_tool(
 ) -> None:
    """Common logic: start proxy, launch tool, clean up."""
    proxy_holder: list[subprocess.Popen | None] = [None]
-    cleanup = _make_cleanup(proxy_holder)
+    cleanup = _make_cleanup(proxy_holder, port)
    signal.signal(signal.SIGINT, cleanup)
    signal.signal(signal.SIGTERM, cleanup)

@ -342,7 +366,7 @@ def claude(port: int, no_rtk: bool, no_proxy: bool, verbose: bool, claude_args:

    # Setup rtk before launching (Claude-specific)
    proxy_holder: list[subprocess.Popen | None] = [None]
-    cleanup = _make_cleanup(proxy_holder)
+    cleanup = _make_cleanup(proxy_holder, port)
    signal.signal(signal.SIGINT, cleanup)
    signal.signal(signal.SIGTERM, cleanup)

@ -530,7 +554,7 @@ def cursor(port: int, no_rtk: bool, no_proxy: bool, verbose: bool) -> None:
        headroom wrap cursor --port 9999    # Custom proxy port
    """
    proxy_holder: list[subprocess.Popen | None] = [None]
-    cleanup = _make_cleanup(proxy_holder)
+    cleanup = _make_cleanup(proxy_holder, port)
    signal.signal(signal.SIGINT, cleanup)
    signal.signal(signal.SIGTERM, cleanup)

--- a/headroom/shared_context.py
+++ b/headroom/shared_context.py
@ -0,0 +1,218 @@
+"""SharedContext — compressed inter-agent context sharing.
+
+When agents hand off to each other, context gets replayed in full.
+SharedContext compresses what moves between agents, using Headroom's
+existing CCR (Compress-Cache-Retrieve) architecture.
+
+Usage:
+
+    from headroom import SharedContext
+
+    ctx = SharedContext()
+
+    # Agent A stores large output
+    ctx.put("research", big_research_output)
+
+    # Agent B gets compressed version (~80% smaller)
+    summary = ctx.get("research")
+
+    # Agent B needs full details on something specific
+    full = ctx.get("research", full=True)
+
+Works with any agent framework. The compression pipeline is the same
+one used by the proxy and MCP server.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ContextEntry:
+    """A stored context entry with original and compressed versions."""
+
+    key: str
+    original: str
+    compressed: str
+    original_tokens: int
+    compressed_tokens: int
+    agent: str | None
+    timestamp: float
+    transforms: list[str] = field(default_factory=list)
+
+    @property
+    def savings_percent(self) -> float:
+        if self.original_tokens == 0:
+            return 0.0
+        return round((1 - self.compressed_tokens / self.original_tokens) * 100, 1)
+
+
+@dataclass
+class SharedContextStats:
+    """Aggregated stats for the shared context."""
+
+    entries: int
+    total_original_tokens: int
+    total_compressed_tokens: int
+    total_tokens_saved: int
+    savings_percent: float
+
+
+class SharedContext:
+    """Compressed shared context for multi-agent workflows.
+
+    Agents put content in, other agents get compressed versions out.
+    Originals are stored for on-demand full retrieval.
+
+    Args:
+        model: Model name for token counting (default: claude-sonnet-4-5).
+        ttl: Time-to-live in seconds (default: 3600 = 1 hour).
+        max_entries: Maximum stored entries (default: 100).
+    """
+
+    def __init__(
+        self,
+        model: str = "claude-sonnet-4-5-20250929",
+        ttl: int = 3600,
+        max_entries: int = 100,
+    ) -> None:
+        self._model = model
+        self._ttl = ttl
+        self._max_entries = max_entries
+        self._entries: dict[str, ContextEntry] = {}
+        self._lock = threading.Lock()
+
+    def put(
+        self,
+        key: str,
+        content: str,
+        *,
+        agent: str | None = None,
+    ) -> ContextEntry:
+        """Store content under a key, compressing automatically.
+
+        Args:
+            key: Name for this context (e.g., "research_findings").
+            content: The content to store and compress.
+            agent: Optional agent identifier for tracking.
+
+        Returns:
+            ContextEntry with compression stats.
+        """
+        from headroom.compress import compress
+
+        messages = [{"role": "tool", "content": content}]
+        result = compress(messages, model=self._model)
+
+        compressed = result.messages[0].get("content", content)
+        if not isinstance(compressed, str):
+            import json
+
+            compressed = json.dumps(compressed)
+
+        entry = ContextEntry(
+            key=key,
+            original=content,
+            compressed=compressed,
+            original_tokens=result.tokens_before,
+            compressed_tokens=result.tokens_after,
+            agent=agent,
+            timestamp=time.time(),
+            transforms=result.transforms_applied,
+        )
+
+        with self._lock:
+            self._evict_if_needed()
+            self._entries[key] = entry
+
+        logger.debug(
+            "SharedContext.put(%s): %d → %d tokens (%.1f%% saved)",
+            key,
+            entry.original_tokens,
+            entry.compressed_tokens,
+            entry.savings_percent,
+        )
+
+        return entry
+
+    def get(
+        self,
+        key: str,
+        *,
+        full: bool = False,
+    ) -> str | None:
+        """Get content by key.
+
+        Args:
+            key: The key to retrieve.
+            full: If True, return the original uncompressed content.
+                  If False (default), return the compressed version.
+
+        Returns:
+            Content string, or None if key not found or expired.
+        """
+        with self._lock:
+            entry = self._entries.get(key)
+            if entry is None:
+                return None
+            if time.time() - entry.timestamp > self._ttl:
+                del self._entries[key]
+                return None
+
+        return entry.original if full else entry.compressed
+
+    def get_entry(self, key: str) -> ContextEntry | None:
+        """Get the full ContextEntry with metadata."""
+        with self._lock:
+            entry = self._entries.get(key)
+            if entry is None:
+                return None
+            if time.time() - entry.timestamp > self._ttl:
+                del self._entries[key]
+                return None
+            return entry
+
+    def keys(self) -> list[str]:
+        """List all non-expired keys."""
+        now = time.time()
+        with self._lock:
+            return [k for k, e in self._entries.items() if now - e.timestamp <= self._ttl]
+
+    def stats(self) -> SharedContextStats:
+        """Get aggregated stats."""
+        now = time.time()
+        with self._lock:
+            active = [e for e in self._entries.values() if now - e.timestamp <= self._ttl]
+        total_orig = sum(e.original_tokens for e in active)
+        total_comp = sum(e.compressed_tokens for e in active)
+        total_saved = total_orig - total_comp
+        pct = round(total_saved / total_orig * 100, 1) if total_orig > 0 else 0.0
+        return SharedContextStats(
+            entries=len(active),
+            total_original_tokens=total_orig,
+            total_compressed_tokens=total_comp,
+            total_tokens_saved=total_saved,
+            savings_percent=pct,
+        )
+
+    def clear(self) -> None:
+        """Remove all entries."""
+        with self._lock:
+            self._entries.clear()
+
+    def _evict_if_needed(self) -> None:
+        """Evict expired and oldest entries if at capacity. Lock must be held."""
+        now = time.time()
+        expired = [k for k, e in self._entries.items() if now - e.timestamp > self._ttl]
+        for k in expired:
+            del self._entries[k]
+
+        while len(self._entries) >= self._max_entries:
+            oldest_key = min(self._entries, key=lambda k: self._entries[k].timestamp)
+            del self._entries[oldest_key]
--- a/tests/test_shared_context.py
+++ b/tests/test_shared_context.py
@ -0,0 +1,129 @@
+"""Tests for SharedContext — compressed inter-agent context sharing."""
+
+from headroom.shared_context import SharedContext
+
+
+class TestPutGet:
+    def test_put_and_get_compressed(self) -> None:
+        ctx = SharedContext()
+        content = " ".join(f"item_{i}: data value {i} with details" for i in range(100))
+        entry = ctx.put("research", content, agent="researcher")
+        assert entry.original_tokens > 0
+        assert entry.key == "research"
+        assert entry.agent == "researcher"
+
+        compressed = ctx.get("research")
+        assert compressed is not None
+        assert len(compressed) <= len(content)
+
+    def test_get_full(self) -> None:
+        ctx = SharedContext()
+        content = "short content that may not compress much"
+        ctx.put("data", content)
+        full = ctx.get("data", full=True)
+        assert full == content
+
+    def test_get_missing_key(self) -> None:
+        ctx = SharedContext()
+        assert ctx.get("nonexistent") is None
+
+    def test_overwrite_key(self) -> None:
+        ctx = SharedContext()
+        ctx.put("k", "first version")
+        ctx.put("k", "second version")
+        assert ctx.get("k", full=True) == "second version"
+
+    def test_get_entry_metadata(self) -> None:
+        ctx = SharedContext()
+        ctx.put("findings", "some data", agent="agent_a")
+        entry = ctx.get_entry("findings")
+        assert entry is not None
+        assert entry.agent == "agent_a"
+        assert entry.original_tokens >= 0
+        assert isinstance(entry.savings_percent, float)
+
+    def test_get_entry_missing(self) -> None:
+        ctx = SharedContext()
+        assert ctx.get_entry("missing") is None
+
+
+class TestExpiry:
+    def test_expired_entry_returns_none(self) -> None:
+        ctx = SharedContext(ttl=0)  # Expire immediately
+        ctx.put("k", "value")
+        import time
+
+        time.sleep(0.01)
+        assert ctx.get("k") is None
+
+    def test_expired_entry_cleaned_from_get_entry(self) -> None:
+        ctx = SharedContext(ttl=0)
+        ctx.put("k", "value")
+        import time
+
+        time.sleep(0.01)
+        assert ctx.get_entry("k") is None
+
+
+class TestKeys:
+    def test_lists_active_keys(self) -> None:
+        ctx = SharedContext()
+        ctx.put("a", "data a")
+        ctx.put("b", "data b")
+        keys = ctx.keys()
+        assert "a" in keys
+        assert "b" in keys
+
+    def test_excludes_expired_keys(self) -> None:
+        ctx = SharedContext(ttl=0)
+        ctx.put("expired", "gone")
+        import time
+
+        time.sleep(0.01)
+        assert "expired" not in ctx.keys()
+
+
+class TestStats:
+    def test_stats_aggregates(self) -> None:
+        ctx = SharedContext()
+        content = " ".join(f"word_{i}" for i in range(50))
+        ctx.put("a", content)
+        ctx.put("b", content)
+        stats = ctx.stats()
+        assert stats.entries == 2
+        assert stats.total_original_tokens > 0
+
+    def test_stats_empty(self) -> None:
+        ctx = SharedContext()
+        stats = ctx.stats()
+        assert stats.entries == 0
+        assert stats.savings_percent == 0.0
+
+
+class TestEviction:
+    def test_evicts_oldest_at_capacity(self) -> None:
+        ctx = SharedContext(max_entries=2)
+        ctx.put("first", "data 1")
+        ctx.put("second", "data 2")
+        ctx.put("third", "data 3")  # Should evict "first"
+        assert ctx.get("first") is None
+        assert ctx.get("second") is not None
+        assert ctx.get("third") is not None
+
+
+class TestClear:
+    def test_clear_removes_all(self) -> None:
+        ctx = SharedContext()
+        ctx.put("a", "x")
+        ctx.put("b", "y")
+        ctx.clear()
+        assert ctx.keys() == []
+
+
+class TestImport:
+    def test_importable_from_headroom(self) -> None:
+        from headroom import SharedContext as SC
+
+        assert SC is not None
+        ctx = SC()
+        assert isinstance(ctx, SharedContext)