Metadata-Version: 2.5
Name: ctx-compact
Version: 0.1.0
Summary: Trim a plain OpenAI-shaped message array down to a token budget without ever orphaning a tool result.
Project-URL: Homepage, https://github.com/pjdurden/ctx-compact
Project-URL: Source, https://github.com/pjdurden/ctx-compact
Author: Prajjwal Chittori
License: MIT
Keywords: compaction,context-window,conversation,llm,openai,token-budget,tool-calls
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# ctx-compact

Trim a plain OpenAI-shaped message array down to a token budget without ever orphaning a tool result. This is the Python port of the [ctx-compact](https://www.npmjs.com/package/ctx-compact) npm package.

## The problem

Every agent framework ships its own conversation compaction (LangGraph, Inspect, MS Agent Framework, the Claude SDK all have one), and each is welded to that framework's message type. If you are working with a plain list of `{role, content, ...}` messages, people tend to hand-roll a "drop the oldest N messages" loop. That works until an assistant message with `tool_calls` gets dropped but its matching `tool` result messages do not (or the reverse). Most providers reject that shape outright, so the trim silently turns into an API error on the next call. `ctx-compact` is a small, framework-neutral compactor that keeps tool-call and tool-result messages paired and dropped or kept as a unit.

## Install

```
pip install ctx-compact
```

## Usage

```python
from ctx_compact import compact, compact_with_summary

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "What is the weather in Denver?"},
    {
        "role": "assistant",
        "content": None,
        "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": '{"city":"Denver"}'}}],
    },
    {"role": "tool", "tool_call_id": "call_1", "content": '{"tempF":72}'},
    {"role": "assistant", "content": "It is 72F in Denver."},
    {"role": "user", "content": "What about Austin?"},
    {
        "role": "assistant",
        "content": None,
        "tool_calls": [{"id": "call_2", "type": "function", "function": {"name": "get_weather", "arguments": '{"city":"Austin"}'}}],
    },
    {"role": "tool", "tool_call_id": "call_2", "content": '{"tempF":88}'},
    {"role": "assistant", "content": "It is 88F in Austin."},
    {"role": "user", "content": "And tomorrow in Denver?"},
]

result = compact(messages, max_tokens=150, keep_head=1, keep_tail=2)
# result.tokens_before -> 195
# result.tokens_after  -> 124
# result.fits          -> True (124 <= 150)
# result.dropped       -> the 3 oldest droppable messages: the first "What is
#                          the weather in Denver?" turn and its whole
#                          assistant/tool_calls + tool group

# Synchronous variant: summarize whatever got dropped and splice a note back in.
with_summary = compact_with_summary(
    messages,
    max_tokens=150,
    keep_head=1,
    keep_tail=2,
    summarize=lambda dropped: f"Earlier in this conversation: {len(dropped)} messages were removed.",
)
# with_summary.summary      -> 'Earlier in this conversation: 3 messages were removed.'
# with_summary.tokens_after -> 145 (the 124 kept after dropping, plus the inserted summary message)
# with_summary.fits         -> True (145 <= 150)
```

The example above is exact output from running this code against this package.

## API

### `compact(messages, *, max_tokens, count_tokens=None, keep_head=1, keep_tail=4) -> CompactResult`

Drops messages from the middle of `messages` until the estimated token count fits the budget.

- `messages`: list of dicts shaped like `{role, content, tool_calls?, tool_call_id?, name?}`. `role` is one of `'system' | 'user' | 'assistant' | 'tool'`.
- `max_tokens` (required, keyword-only, `float`) - the budget. Raises `TypeError` if missing or not a positive number.
- `count_tokens` - `(message) -> int`. Default: `math.ceil(len(json.dumps(message, separators=(",", ":"), ensure_ascii=False)) / 4)`, with the length measured in UTF-16 code units (matching JavaScript's `String.length`), not Python codepoints.
- `keep_head` - number of leading messages always kept. Default `1`.
- `keep_tail` - number of trailing messages always kept. Default `4`.

Returns a frozen dataclass:

```python
@dataclass(frozen=True)
class CompactResult:
    messages: list       # the compacted list
    dropped: list        # what was removed, in original order
    tokens_before: int   # summed estimated tokens of the input list
    tokens_after: int    # summed estimated tokens of the output list
    fits: bool           # tokens_after <= max_tokens
```

If the input already fits, it is returned unchanged with `dropped: []`.

### `compact_with_summary(messages, *, max_tokens, count_tokens=None, keep_head=1, keep_tail=4, summarize=None, summary_role="user") -> CompactResultWithSummary`

Same as `compact`, then, if anything was dropped and `summarize` is provided, calls `summarize(dropped)` and inserts the returned string as a message `{"role": summary_role, "content": <summary>}` immediately after the head-kept messages.

- `summarize` - `(dropped: list) -> str`. If omitted, behaves exactly like `compact` and returns `summary: None`.
- `summary_role` - default `'user'`. Some providers reject a second `system` message, which is why the default is `'user'` rather than `'system'`.

The inserted summary message counts toward the budget: after insertion the result is re-checked, and if it no longer fits, additional whole groups are dropped (oldest first) to make room. `fits` is reported `False` if it is still over budget after that.

Returns a frozen dataclass with the same fields as `CompactResult` plus `summary: Optional[str]`.

**Difference from the JavaScript version:** the JS `compactWithSummary` is `async` and does `await options.summarize(dropped)`, since JS summarizers are typically an async LLM call. This Python port is **synchronous**: `summarize` is a plain callable, `dropped -> str`, called directly with no `await`. If your summarizer needs to be async in Python, run it yourself (e.g. via `asyncio.run` or your event loop) before calling `compact_with_summary`, and pass a synchronous wrapper.

### `estimate_tokens(message, count_tokens=None) -> int`

Runs `count_tokens` (or the default heuristic) against a single message. Exported so callers can reuse the same estimator `compact`/`compact_with_summary` use, e.g. to pre-check a message before appending it.

## How it works

1. **Group first.** Before anything is dropped, the whole list is split into groups: an assistant message carrying `tool_calls` plus every immediately-following `tool` message whose `tool_call_id` matches one of that assistant's `tool_calls[].id` forms one group. Every other message (including a `tool` message with no matching assistant) is its own group. Groups are always dropped or kept whole, so a tool result is never left without its assistant call, or vice versa.
2. **Snap keep_head/keep_tail to group boundaries.** `keep_head` and `keep_tail` are counted in messages, but if the boundary would land inside a group, it expands outward to keep that whole group.
3. **Drop oldest-first.** Whatever is left in the middle is droppable. Groups are dropped oldest first until the running token total fits `max_tokens` or nothing droppable is left.
4. Token counts are an estimate (`~length / 4` by default, over the compact JSON serialization of the message, or your own `count_tokens`), not a real tokenizer. There is no LLM call, no tokenizer library, and no streaming. If your `count_tokens` is inaccurate, `fits` will be inaccurate too. Pass a `count_tokens` backed by your provider's real tokenizer if you need exact numbers.
5. `compact_with_summary` never re-summarizes after dropping additional groups to make room for the summary itself; it just drops more of the already-dropped-eligible messages. If you need every dropped message reflected in the summary text, make sure `max_tokens` leaves enough headroom for the summary you expect `summarize` to produce.
6. The default estimator uses `json.dumps(message, separators=(",", ":"), ensure_ascii=False)`, matching the length JavaScript's `JSON.stringify` plus `.length` produces. Two of Python's `json.dumps` defaults disagree with `JSON.stringify` and both are overridden here:
   - `separators` defaults to `", "` and `": "` (with spaces) in Python; `JSON.stringify` never inserts spaces. Passing plain `json.dumps(message)` would inflate every count.
   - `ensure_ascii` defaults to `True` in Python, which `\uXXXX`-escapes every non-ASCII codepoint (accents, CJK, emoji); `JSON.stringify` never escapes non-ASCII text. Leaving `ensure_ascii` at its default would silently inflate the token count, and therefore over-compact, for any non-English or emoji-bearing content.

   A third, more subtle difference is corrected for internally rather than in the `json.dumps` call: JavaScript's `String.length` counts UTF-16 code units, so a character outside the Basic Multilingual Plane (most emoji, e.g. U+1F389) counts as a 2-unit surrogate pair there, while Python's `len()` counts it as a single codepoint. The estimator measures the serialized JSON's length in UTF-16 code units (not `len()` codepoints) so astral-plane characters do not silently disagree with the JS package's token numbers for the same message.

See the JavaScript version at the repo root (`../index.js`, `../README.md`) for the original implementation this port matches.

## License

MIT
