diff --git a/docs/agent.md b/docs/agent.md
index fd1f76c..4e43c50 100644
--- a/docs/agent.md
+++ b/docs/agent.md
@@ -162,7 +162,7 @@
 
 Run sequentially after `StreamEnd`. Currently: `CompressionWorker`. Workers receive a `WorkerContext` carrying the active `profile`, so `CompressionWorker` applies per-profile compression overrides (`compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file`). The worker gates on the real `context_tokens` from the last LLM call, then delegates to `compress_and_save_session(reason="postturn")` — the same retry / hard-truncate / safety-net / archiving pipeline as the pre- and mid-turn paths (with the mid-turn `keep_recent_messages`).
 
-Pre-turn compression also runs at the start of `run_stream()`: it estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_tokens` from the last call + heuristic delta; the `chars // 3` heuristic only before the first LLM call of a session) and, when over threshold, is guarded by `would_compress()` before emitting `CompressionStarted`. See [`sessions.md`](sessions.md).
+Pre-turn compression also runs at the start of `run_stream()`: it estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_eval_count` from the last call + calibrated delta; the calibrated `estimate_tokens` only before the first LLM call of a session) and, when over `window × threshold`, is guarded by `would_compress(current_tokens=…)` before emitting `CompressionStarted`. The window is `profile.context_window` or `settings.ollama_num_ctx` (`Agent._window_for`). See [`sessions.md`](sessions.md).
 
 ---
 
diff --git a/docs/api.md b/docs/api.md
index 47967f3..482d507 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -656,7 +656,7 @@
 
 #### `GET /sessions/{session_id}/files/{filename}`
 
-Download or view an uploaded file. Images, PDFs, plain text and HTML are served inline; everything else as an attachment.
+Download or view an uploaded file. Images, audio, video, PDF, HTML and text/source files are served inline; everything else as an attachment. The type comes from `navi/content_types.py` — extension first, then the file's own signature — rather than from the host's `/etc/mime.types`, which calls a TypeScript `.ts` file a video stream.
 
 **Query params**
 - `download` — force attachment download regardless of content type
@@ -675,7 +675,7 @@
 
 **Response `200`**
 ```json
-{ "session_id": "...", "files": [{"name": "...", "size": N, "is_dir": false, "path": "..."}, ...] }
+{ "session_id": "...", "files": [{"name": "...", "size": N, "is_dir": false, "path": "...", "content_type": "audio"}, ...] }
 ```
 
 **Errors** — `404` session not found; `403` no access.
@@ -1008,7 +1008,15 @@
 ```json
 { "type": "heartbeat" }
 ```
-Keepalive ping sent every 20 s during long silent operations. Client can ignore.
+Keepalive sent every 20 s of server silence — mid-run and on an idle socket alike. It is also the client's liveness signal: a socket that stops delivering frames past ~45 s is half-open and must be reopened. A client may ignore the frame itself, but not its absence.
+
+---
+
+#### `pong`
+```json
+{ "type": "pong", "ts": 1748000000000 }
+```
+Answer to a client `{"type": "ping", "ts": ...}` frame; `ts` is echoed untouched so the client can measure the round trip.
 
 ---
 
diff --git a/docs/mechanics.md b/docs/mechanics.md
index 6a592f7..7415487 100644
--- a/docs/mechanics.md
+++ b/docs/mechanics.md
@@ -29,12 +29,14 @@
 | **Cooperative stop** | Checks `current_stop_event` (asyncio.Event) before each LLM call, during streaming, and after tool execution. Uses clean generator close — never `task.cancel()`. | None | `agent.py` | ✅ |
 | **Agent-invoked planning** | The `plan` tool runs the planning pipeline when the model itself decides a task needs it. PlanRunner is bound per tool-loop iteration via the `current_plan_runner` ContextVar; the tool result carries the follow-up instruction (complex → present and wait for confirmation, else proceed). PlanningStatus/PlanReady reach the UI mid-turn via `current_event_sink`. | None | `agent.py`, `plan.py` | ✅ |
 | **Profile reload mid-session** | After each tool execution batch, compares the stored session's profile against the profile the run is **bound** to (`bound_profile_id`), not against `session.profile_id` — after a switch the two agree again, because `set_profile()` either rewrites the row (and the run's own `save()` used to write its stale value back) or mutates the very Session object the run holds. On a difference, reloads profile, tools, schemas, and backend for the next iteration. `save()` no longer writes `profile_id` on conflict: `set_profile()` owns that column after creation. | None | `agent.py`, `pg_session_store.py` | ✅ |
-| **Pre-turn context compression** | Before the assistant reply, estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_tokens` bulk + heuristic delta; chars//3 heuristic only before the first LLM call) and compresses when over threshold. Guarded by `would_compress()` so `CompressionStarted` is only emitted when the partition can actually shrink the stored context. | `CONTEXT_COMPRESSION_ENABLED`, `OLLAMA_NUM_CTX`, `CONTEXT_COMPRESSION_THRESHOLD` | `agent.py` | ✅ |
-| **Mid-turn context compression** | On iterations > 0, estimates tokens via `real_baseline_estimate(session.context, preflight_ctx)` (real prompt_tokens bulk + heuristic delta) and compresses with `keep_recent_messages=max(12, CONTEXT_KEEP_RECENT*2)`. Guarded by `would_compress()`. For long autonomous loops where the entire conversation is one turn. | Same as above + `CONTEXT_KEEP_RECENT` | `agent.py` | ✅ |
+| **Pre-turn context compression** | Before the assistant reply, estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_eval_count` bulk + calibrated delta; `estimate_tokens` only before the first LLM call) and compresses when over `window × threshold`, where `window` is `profile.context_window` or `OLLAMA_NUM_CTX`. Guarded by `would_compress(current_tokens=…)` so `CompressionStarted` is only emitted when the partition can actually shrink the stored context. | `CONTEXT_COMPRESSION_ENABLED`, `OLLAMA_NUM_CTX`, `profile.context_window`, `CONTEXT_COMPRESSION_THRESHOLD` | `agent.py` | ✅ |
+| **Mid-turn context compression** | On iterations > 0, estimates tokens via `real_baseline_estimate(session.context, preflight_ctx)` (real prompt_tokens bulk + calibrated delta) and compresses with `keep_recent_messages=max(12, CONTEXT_KEEP_RECENT*2)`. Guarded by `would_compress(current_tokens=…)`. For long autonomous loops where the entire conversation is one turn. The post-turn worker passes `WorkerContext.max_context_tokens` as the same window. | Same as above + `CONTEXT_KEEP_RECENT` | `agent.py`, `workers/compressor.py` | ✅ |
 | **Forced `/compact`** | `compact_stream()` — bypasses the token threshold and compresses immediately on client demand (`{"type":"compact"}` WS control message). Emits `CompressionStarted` + `ContextCompressed`; raises `NothingToCompactError` when there is nothing to compress. Bound to `Ctrl+X C` in the TUI. | None | `agent.py`, `websocket.py` | ✅ |
-| **Context size check with output reserve** | Raises `ContextTooLargeError` if estimated input tokens exceed `OLLAMA_NUM_CTX - OUTPUT_RESERVE_TOKENS`. When the caller passes `session_context`, the total uses `real_baseline_estimate()` (real bulk + heuristic delta) instead of the chars//3 estimate. Images counted at 500 tokens each. | `OLLAMA_NUM_CTX`, `OUTPUT_RESERVE_TOKENS` | `compressor.py` | ✅ |
-| **Real-token baseline estimator** | Records the real `prompt_tokens` returned by each LLM call (`record_real_baseline(len(session.context), …)`) and reuses them as the bulk of the next estimate, heuristicking only the messages appended since. Used by the mid-turn gate and `check_context_size`. Cleared after compression (context shrank). | None | `compressor.py`, `agent.py` | ✅ |
-| **Local token estimation** | Conservative estimate `chars // 3 + imgs * 500` (`ContextCompressor.estimate_context_tokens`, a staticmethod). Used as the fallback when no real baseline exists and for the per-message delta. | None | `compressor.py` | ✅ |
+| **Context size check with output reserve** | Raises `ContextTooLargeError` if estimated input tokens exceed `window - OUTPUT_RESERVE_TOKENS` (the reserve is capped at a quarter of the window). When the caller passes `session_context`, the total uses `real_baseline_estimate()` (real bulk + calibrated delta) instead of the local estimate. Images counted at 500 tokens each. | `OLLAMA_NUM_CTX`, `profile.context_window`, `OUTPUT_RESERVE_TOKENS` | `compressor.py` | ✅ |
+| **Real-token baseline estimator** | Records the real `prompt_eval_count` returned by each LLM call (`record_real_baseline(len(session.context), …)`, plus the chars of the built context) and reuses it as the bulk of the next estimate, pricing only the messages appended since. Each observation also feeds the per-model token calibration. Used by the pre-turn, mid-turn and post-turn gates and `check_context_size`. Cleared after compression (context shrank). | None | `compressor.py`, `agent.py` | ✅ |
+| **Per-model token calibration** | `TokenCalibration` fits `tokens ≈ overhead + chars × tokens_per_char` (least squares) to the last 8 `(chars, real prompt tokens)` pairs for a model; needs ≥2 observations with a ≥1.25× size spread and rejects slopes outside 1 token/char … 1 per 4 chars. `ContextCompressor.estimate_tokens(context, model)` uses it for every keep-region target, the hard-truncate budget and image budgeting, falling back to `chars / 1.8 + imgs * 500`. `chars // 3` undercounted prod prompts ~1.8×. | None | `token_calibration.py`, `compressor.py` | ✅ |
+| **Per-profile context window** | `AgentProfile.context_window` (default `OLLAMA_NUM_CTX`) is resolved once per turn by `Agent._window_for()` and used for the compression trigger/target, `check_context_size`, the image budget and `WorkerContext.max_context_tokens`; it is also sent to Ollama as `num_ctx`. | `profile.context_window`, `OLLAMA_NUM_CTX` | `agent.py`, `profiles/base.py`, `llm/ollama.py` | ✅ |
+| **Context-only demotion persistence** | `compress_and_save_session` collects demoted rows that exist only in `session.context` (`is_display=False`, seq ≥ 0) and flips them through `SessionStore.set_context_flag()` → a narrow `UPDATE session_messages SET is_context=…` by sequence number. `save()` only writes flags for rows in `session.messages`, so without this an old summary came back on the next load (prod: up to 15 live summaries in one session). | None | `compressor.py`, `pg_session_store.py`, `session.py` | ✅ |
 | **Anti-stall detection** | Tracks two signals: (1) consecutive iterations with no todo status change, (2) identical tool call signatures. When either hits threshold, injects a hard warning system message. | `profile.anti_stall_enabled`, `profile.anti_stall_threshold` | `agent.py` | ✅ |
 | **Goal anchoring** | Injects `[Goal anchor]` system message with original request + todo state every N iterations. | `profile.goal_anchoring_enabled`, `profile.goal_anchoring_interval` | `agent.py` | ✅ |
 | **Bounded autonomy — scope boundary** | Injects a standing `[Scope boundary]` system message keeping the agent within the user's literally requested scope — no expanding to sibling/parent dirs/projects, no executing discovered TODO/roadmap/milestone docs. | `profile.scope_boundary_enabled` | `agent.py`, `context_builder.py` | ✅ |
@@ -103,10 +105,12 @@
 | **Turn-based partitioning** | Groups messages into turns. Keeps last `keep_recent` turns verbatim; older go to summarization. Tool call groups never split. Adaptive: `_turn_importance` scores turns and swaps an important old turn into the kept set in place of a filler-recent one. | `CONTEXT_KEEP_RECENT`, `profile.compression_keep_recent` | `compressor.py` | ✅ |
 | **Intra-turn fallback partitioning** | `partition_current_turn_messages`: for long autonomous loops where the entire conversation is one turn, keeps the current request + newest N messages verbatim and summarizes older messages from the same turn. | `keep_recent_messages` param | `compressor.py` | ✅ |
 | **Honest `CompressionStarted` guard** | `would_compress()` runs the same `_plan_compression` decision as the real compression path (no LLM) + the token-budget check, so the agent emits `CompressionStarted` only when the partition can actually shrink the stored context — never a misleading "compression" status with no work done. The partition/hysteresis/fallback decision lives in one function shared with `compress_context`, so prediction and reality cannot drift apart. | None | `compressor.py` | ✅ |
-| **Token-budget hard-truncate fallback** | When `compress_session` gets a partition no-op but the token gate fired, drops oldest turns (keeping system + newest whole turns) until under `_HARD_TRUNCATE_TOKEN_FRAC=0.5` of the window. Last-resort, no LLM. | `OLLAMA_NUM_CTX` | `compressor.py` | ✅ |
+| **Token-budget hard-truncate fallback** | When `compress_session` gets a partition no-op but the token gate fired, drops oldest turns (keeping system + newest whole turns) until under `_HARD_TRUNCATE_TOKEN_FRAC=0.5` of the window, priced with the calibrated estimator. Last-resort, no LLM. | `OLLAMA_NUM_CTX`, `profile.context_window` | `compressor.py` | ✅ |
 | **Per-message view truncation** | `ContextBuilder._truncate_oversized` head/tail-truncates a single `tool`/`assistant` message whose estimated size exceeds the per-message budget in the *built* context only — stored history is never mutated. 0 = auto (`OLLAMA_NUM_CTX // 6`). | `CONTEXT_MESSAGE_TOKEN_BUDGET` | `context_builder.py` | ✅ |
-| **Profile-aware compression overrides** | `compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file` override the global defaults inside `compress_context` / `compress_session` / the summary system prompt. `Agent.set_profile()` propagates the active profile to the compressor. | `profile.compression_keep_recent`, `profile.compression_max_tokens`, `profile.compression_prompt_file` | `compressor.py`, `profiles/base.py` | ✅ |
-| **Real-token baseline estimator** | `record_real_baseline()` caches the last call's real `prompt_tokens` keyed by `len(session.context)`; `real_baseline_estimate()` returns `real_tokens + heuristic(messages since)` so code-heavy contexts are estimated accurately instead of undercounted by chars//3. Used by the pre-turn, mid-turn and `/compact` gates and `check_context_size`. Cleared after compression. | None | `compressor.py`, `agent.py` | ✅ |
+| **Profile-aware compression overrides** | `compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file` override the global defaults inside `compress_context` / `compress_session` / the summary system prompt; `context_window` overrides the window for every decision and is sent as `num_ctx`. `Agent.set_profile()` propagates the active profile to the compressor. | `profile.context_window`, `profile.compression_keep_recent`, `profile.compression_max_tokens`, `profile.compression_prompt_file` | `compressor.py`, `profiles/base.py` | ✅ |
+| **Real-token baseline estimator** | `record_real_baseline()` caches the last call's real `prompt_eval_count` keyed by `len(session.context)` (and feeds the calibration); `real_baseline_estimate()` returns `real_tokens + calibrated(messages since)` so code-heavy contexts are estimated accurately instead of undercounted by a character heuristic. Used by the pre-turn, mid-turn and `/compact` gates and `check_context_size`. Cleared after compression. | None | `compressor.py`, `agent.py` | ✅ |
+| **Per-model token calibration** | `TokenCalibration.tokens_for(chars)` — least-squares line `overhead + chars × tokens_per_char` over the last 8 real observations for a model; fallback `chars / 1.8`. Every keep-region target and budget is measured in these real tokens. | None | `token_calibration.py` | ✅ |
+| **Context-only demotion persistence** | Demoted rows absent from `session.messages` are flipped in the database through `SessionStore.set_context_flag()` (`UPDATE session_messages SET is_context=…` by sequence number, no `is_display` filter) — `save()` alone would leave them `is_context=True` on the next load. | None | `compressor.py`, `pg_session_store.py` | ✅ |
 | **Summary input formatter** | Renders messages as plain text for summarizer: preserves summaries, notes image counts, renders tool calls compactly, collects base64 images for vision models. Critical tool results (`is_compression_critical` or critical tool names) survive verbatim up to 4000 chars, then head+tail halves (2000+2000) instead of collapsing to the non-critical preview; non-critical results capped at an 800-char preview. | None | `compressor.py` | ✅ |
 | **Summary input truncate** | Hard cap of 32 000 chars on formatted input sent to summarizer LLM, keeping head (75%, oldest messages) + tail (25%, the messages closest to the kept window) — the newest summarized work is never silently dropped. | Hard-coded `_MAX_SUMMARY_INPUT_CHARS=32000` | `compressor.py` | ✅ |
 | **Meta-summary consolidation** | When `to_summarize` contains multiple existing summaries totaling > `_META_SUMMARY_THRESHOLD≈10666` chars, consolidates them into one via `_meta_summarize` first so old summaries don't crowd the summarizer input. | None | `compressor.py` | ✅ |
diff --git a/docs/profiles.md b/docs/profiles.md
index 08a5349..9a4d5b9 100644
--- a/docs/profiles.md
+++ b/docs/profiles.md
@@ -113,10 +113,11 @@
 
 ### Compression
 
-Per-profile overrides for context compression. When set, they take precedence over the global `settings` (`CONTEXT_KEEP_RECENT`, `CONTEXT_SUMMARY_MAX_TOKENS`) inside `compress_context` / `compress_session` and the summary system prompt. The active profile is propagated to the compressor via `Agent.set_profile()` and to the post-turn `CompressionWorker` via `WorkerContext.profile`.
+Per-profile overrides for context compression. When set, they take precedence over the global `settings` (`CONTEXT_KEEP_RECENT`, `CONTEXT_SUMMARY_MAX_TOKENS`, `OLLAMA_NUM_CTX`) inside `compress_context` / `compress_session` and the summary system prompt. The active profile is propagated to the compressor via `Agent.set_profile()` and to the post-turn `CompressionWorker` via `WorkerContext.profile`.
 
 | Key | Type | Default | Description |
 |---|---|---|---|
+| `context_window` | int \| None | `None` | The model's real context window in tokens. Used as the window for every compression decision (trigger, target, hard-truncate, `check_context_size`, image budget) and sent to Ollama as `num_ctx`. `None` = `settings.ollama_num_ctx`. Set it explicitly for cloud-hosted models: `ollama show` does not report `context_length` for them, so nothing can detect it, and compressing a wide-window model against 65k costs a summary pass every ~35 messages. |
 | `compression_keep_recent` | int \| None | `None` | Override `CONTEXT_KEEP_RECENT` — number of recent turns kept verbatim. `navi_code` uses `12`. |
 | `compression_max_tokens` | int \| None | `None` | Override `CONTEXT_SUMMARY_MAX_TOKENS` — max output tokens for the summary LLM call. |
 | `compression_prompt_file` | str \| None | `None` | Extra instructions appended to the summary system prompt (loaded from the profile dir). |
diff --git a/docs/sessions.md b/docs/sessions.md
index 6367bac..d6ecc73 100644
--- a/docs/sessions.md
+++ b/docs/sessions.md
@@ -96,37 +96,49 @@
 
 Keeps the LLM context within the token budget by summarizing old conversation turns.
 
+### Token accounting and the window
+
+Every window decision — the trigger, the hysteresis target, the hard-truncate budget, `check_context_size`, image budgeting — is made in **real prompt tokens**, not in a character heuristic.
+
+`ContextCompressor.estimate_tokens(context, model)` prices a message list through `navi/core/token_calibration.py`:
+
+- The calibration fits a line `tokens ≈ overhead + chars × tokens_per_char` to the last 8 observations of `(chars of the built context, real prompt_eval_count)`, recorded by `record_real_baseline()` whenever a real count comes back from Ollama. It needs at least two observations whose sizes differ by 1.25× or more, and rejects an implausible slope (outside 1 token/char … 1 token per 4 chars) — which in practice only happens when a prompt was truncated server-side.
+- Without a usable line it falls back to `chars / 1.8`. The old `chars // 3` was measured ~1.8× below the real prompt size on prod: "65% of the window" was met on paper while the real prompt stayed above the window, so compression re-fired on the next messages (48 passes in a day on one session).
+- Images add 500 tokens each. Calibration is per model (`calibration_for(model)`), so an observation from one model never prices another, and it lives in a module-level registry because the post-turn worker builds a fresh compressor per pass.
+
+The window is a parameter, not a global. `AgentProfile.context_window` (or `settings.ollama_num_ctx` when unset) is resolved once per turn (`Agent._window_for`) and threaded through `compress_and_save_session(..., window=)` → `compress_context` / `_hard_truncate` / `check_context_size`, and is also sent to Ollama as `num_ctx`. Without this, a model with a larger real window than `ollama_num_ctx` is compressed against the wrong budget.
+
 ### When it triggers
 
 Three trigger points:
 
-1. **Pre-turn** (in `run_stream()` → `_compression_events_preturn`): before the first LLM call of a turn, estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_tokens` from the last call + heuristic delta for the messages appended since it; `chars // 3` heuristic only when no baseline exists yet) and compresses when `tokens >= num_ctx * threshold`. Guarded by `would_compress()` so `CompressionStarted` is only emitted when the partition can actually shrink the stored context.
+1. **Pre-turn** (in `run_stream()` → `_compression_events_preturn`): before the first LLM call of a turn, estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_eval_count` from the last call + calibrated delta for the messages appended since it; `estimate_tokens` only when no baseline exists yet) and compresses when `tokens >= window * threshold`. Guarded by `would_compress(..., current_tokens=estimated_tokens, model=...)` so `CompressionStarted` is only emitted when the partition can actually shrink the stored context; the dry run reuses the caller's real-token count instead of re-deriving it with a different heuristic.
 2. **Mid-turn** (in `run_stream()` → `_compression_events_midturn`, every iteration > 0): estimates tokens via `real_baseline_estimate(session.context, preflight_ctx)` and compresses with `keep_recent_messages=max(12, context_keep_recent*2)`. This is what keeps long autonomous loops (one user message + many tool iterations = one turn) from exhausting the window.
-3. **Post-turn** (via `CompressionWorker`): after `StreamEnd`, the worker re-checks (using the real `context_tokens` from the last call) and delegates to `compress_and_save_session(reason="postturn")` with the mid-turn `keep_recent_messages` — the same pipeline (retry, hard-truncate, safety net, archiving) as the pre/mid-turn paths.
+3. **Post-turn** (via `CompressionWorker`): after `StreamEnd`, the worker re-checks (using the real `ctx.context_tokens` from the last call and `WorkerContext.max_context_tokens` as the window) and delegates to `compress_and_save_session(reason="postturn", window=ctx.max_context_tokens)` with the mid-turn `keep_recent_messages` — the same pipeline (retry, hard-truncate, safety net, archiving) as the pre/mid-turn paths.
 
 A fourth, on-demand path is **forced `/compact`** (`compact_stream()`): the client sends `{"type":"compact"}`, bypassing the threshold entirely; emits `CompressionStarted` + `ContextCompressed`; raises `NothingToCompactError` when there is nothing to compress.
 
 Config values (`settings`):
 - `context_compression_enabled: bool = True`
-- `context_compression_threshold: float = 0.90` — trigger at 90% of `ollama_num_ctx`
-- `context_compression_target: float = 0.65` — hysteresis target: after compression the kept region should fit ~65% of the window, so the trigger doesn't re-fire a few messages later
+- `context_compression_threshold: float = 0.90` — trigger at 90% of the active window (real tokens)
+- `context_compression_target: float = 0.65` — hysteresis target: after compression the kept region should fit ~65% of the active window, so the trigger doesn't re-fire a few messages later
 - `context_keep_recent: int = 8` — keep last N conversational turns verbatim
 - `context_summary_temperature: float = 0.3`
 - `context_summary_max_tokens: int = 6000` — max output tokens for the summary LLM call
 - `output_reserve_tokens: int = 2048` — headroom reserved for the response in `check_context_size`
-- `context_message_token_budget: int = 0` — per-message view truncation budget (`0` = auto, `ollama_num_ctx // 6`)
+- `context_message_token_budget: int = 0` — per-message view truncation budget (`0` = auto, active window // 6)
 
-Per-profile overrides (`AgentProfile`): `compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file` — applied inside `compress_context` / `compress_session` / the summary system prompt. navi_code uses `compression_keep_recent=12`.
+Per-profile overrides (`AgentProfile`): `context_window` (the window itself), `compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file` — the compression ones are applied inside `compress_context` / `compress_session` / the summary system prompt. navi_code uses `compression_keep_recent=12`.
 
 ### Context size guard
 
-Before every LLM call, `check_context_size(built_ctx, session_context=session.context)` raises `ContextTooLargeError` when the estimated input exceeds `ollama_num_ctx - output_reserve_tokens`. The total uses `real_baseline_estimate()` when a baseline is available (real bulk + heuristic delta), falling back to the `chars // 3 + 500 per image` estimate. The error is surfaced to the user as a synthesized assistant response + `StreamEnd` rather than a raw system error.
+Before every LLM call, `check_context_size(built_ctx, session_context=session.context, model=profile.model, window=self._window_for(profile))` raises `ContextTooLargeError` when the estimated input exceeds `window - output_reserve_tokens`. The total uses `real_baseline_estimate()` when a baseline is available (real bulk + calibrated delta), falling back to `estimate_tokens()` (calibrated line, `chars / 1.8` without one, + 500 per image). `output_reserve_tokens` is capped at a quarter of the window, so a deliberately small test/profile window does not fail every check. The error is surfaced to the user as a synthesized assistant response + `StreamEnd` rather than a raw system error.
 
 ### Compression algorithm
 
-`compress_context(context, llm, model, temperature, keep_recent, *, max_tokens=None, keep_recent_messages=None, profile=None)`:
+`compress_context(context, llm, model, temperature, keep_recent, max_tokens=None, keep_recent_messages=None, profile=None, window=None, estimate=None)`:
 
-1. Resolve effective `keep_recent`/`max_tokens` from `profile.compression_*` overrides, then run `_plan_compression` — the single decision point shared with `would_compress()` (dry-run), so the prediction and the real compression can never drift apart. It partitions messages into `to_summarize` (old turns) and `to_keep` (recent `keep_recent` turns), shrinks `keep_recent` until the kept region fits the 65% target (turn-based mode), and retries with `keep_recent_messages=2` when the mid-turn partition found nothing.
+1. Resolve effective `keep_recent`/`max_tokens` from `profile.compression_*` overrides, then run `_plan_compression(context, keep_recent, keep_recent_messages, window=..., estimate=...)` — the single decision point shared with `would_compress()` (dry-run), so the prediction and the real compression can never drift apart. It partitions messages into `to_summarize` (old turns) and `to_keep` (recent `keep_recent` turns), shrinks `keep_recent` until the kept region fits `window × context_compression_target` measured with `estimate` in real tokens (turn-based mode), and retries with `keep_recent_messages=2` when the mid-turn partition found nothing. `window`/`estimate` default to `settings.ollama_num_ctx` and `estimate_context_tokens`.
    - A "turn" = one user message + all following assistant/tool messages up to the next user message.
    - Tool call groups (assistant + results) are never split across the partition.
    - **Adaptive partitioning:** `_turn_importance` scores each turn; an important old turn can be swapped into the kept set in place of a filler-recent one.
@@ -140,7 +152,9 @@
 
 `compress_session` wraps `compress_context` with retry + a **token-budget hard-truncate fallback** (`_hard_truncate`): if the LLM summarization fails twice, it drops oldest turns (keeping system + newest whole turns) until under `_HARD_TRUNCATE_TOKEN_FRAC = 0.5` of the window — a last resort with no LLM call.
 
-`compress_and_save_session` then mutates the session: replaces `session.context`, marks dropped messages `is_context=False`, appends the summary (`is_display=False`) and an `is_compression=True` system marker to `session.messages`, resets `context_token_count`, archives old messages when `session_messages_window` is exceeded, persists, and logs `compressor.compressed` with the trigger `reason` (`preturn` / `midturn` / `postturn` / `forced`). It also clears the real-token baseline (the context just shrank).
+`compress_and_save_session(session, session_store, ..., window=...)` then mutates the session: replaces `session.context`, marks dropped messages `is_context=False`, appends the summary (`is_display=False`) and an `is_compression=True` system marker to `session.messages`, resets `context_token_count` to the calibrated `estimate_tokens(new_context)`, archives old messages when `session_messages_window` is exceeded, persists, and logs `compressor.compressed` with the trigger `reason` (`preturn` / `midturn` / `postturn` / `forced`). It also clears the real-token baseline (the context just shrank).
+
+The demotion is persisted explicitly. `SessionStore.save()` only writes flags for rows present in `session.messages`; an older summary is context-only (`is_display=False`), so flipping `is_context` on it in memory never reached the database and the summary came back on the next load — on prod that left up to 15 live summaries in one session, ~80% of its context. `compress_and_save_session` therefore collects the demoted context-only rows (seq ≥ 0, absent from `session.messages`) and calls `session_store.set_context_flag(session_id, sequence_numbers, False)`, a narrow `UPDATE session_messages SET is_context = …` by sequence number that deliberately does not filter on `is_display`.
 
 All four trigger paths go through `compress_and_save_session`, so a summarizer LLM failure is always non-fatal: retry with a wider keep window, then hard-truncate. `CompressionWorker` only catches genuinely unexpected errors (e.g. a store failure) and logs a warning.
 
@@ -200,7 +214,7 @@
 
 Two tools expose session files to the user:
 - `share_file` copies an existing local file into the session directory and returns a download link. It requires an absolute source path and has its own size limit (`SHARE_FILE_MAX_SIZE_MB`, default 1024 MB).
-- `content_publish` registers a file that already exists in the session directory and exposes it as an inline viewer/card through `/sessions/{id}/files/{filename}`. It does not copy files.
+- `content_publish` registers a file that already exists in the session directory and exposes it as an inline viewer/card through `/sessions/{id}/files/{filename}`. It does not copy files. The card renders STL/glTF models, HTML, SVG, PDF, images, video, audio, and text/source files; the type is detected by `navi/content_types.py` from the extension and the file's signature (a PNG saved as `.txt` still opens as an image, a `.ts` that is not an MPEG transport stream opens as text).
 
 ---
 
diff --git a/docs/websocket.md b/docs/websocket.md
index e1eb827..87f0c30 100644
--- a/docs/websocket.md
+++ b/docs/websocket.md
@@ -75,6 +75,14 @@
 - `images`: optional list of base64-encoded images (data URIs accepted; the `data:...;base64,` prefix is stripped server-side). **Limits:** max 8 images per message, 50 MB total payload (across all images). Excess is rejected with a WebSocket error.
 - `files`: optional list of uploaded file references (appended to content as `[Uploaded files on disk: ...]`).
 
+### Liveness probe
+
+```json
+{"type": "ping", "ts": 1748000000000}
+```
+
+Answered with `{"type": "pong", "ts": ...}` (same `ts`, no other work). A `close` event never arrives for a half-open socket — a phone that slept, a suspended PWA or a dropped NAT mapping leaves `readyState` at `OPEN` while writes go nowhere — so the only evidence that the connection still works end to end is traffic coming back. The server also sends `heartbeat` after 20 s of its own silence, mid-run or idle, which serves the same purpose without a client probe.
+
 ### Forced compression
 
 ```json
@@ -159,7 +167,8 @@
 | `{"type": "terminal_closed", "terminal_name": "...", "reason": "explicit\|idle_timeout\|error\|shutdown\|session_ended"}` | A persistent terminal session ended |
 | `{"type": "task_update", "task_id": "bt-...", "session_id": "...", "tool": "...", "status": "running\|completed\|failed\|cancelled", "result_preview": "...", "parent_tool_call_id": "...", "started_at": "...", "finished_at": "...", "subagent_tokens": N}` | A background task (tool call detached with `background: true`, see `docs/tasks.md`) changed state. Out-of-band: NOT part of the run's replay buffer. `parent_tool_call_id` links it to the tool card that spawned it. |
 | `{"type": "message_queued", "position": N, "queue_len": N, "max": N, "dropped_total": N}` | The user's message arrived while a run was active and was queued (not an error). It will execute after the current run finishes; `dropped_total` counts messages dropped when the queue was full. |
-| `{"type": "heartbeat"}` | Periodic keepalive during long silent operations (every 20 s) |
+| `{"type": "heartbeat"}` | Periodic keepalive after 20 s of server silence — mid-run and on an idle socket alike. Also the client's liveness signal: silence past ~45 s means the socket is half-open |
+| `{"type": "pong", "ts": N}` | Answer to a client `ping`, `ts` echoed untouched |
 | `{"type": "session_sync", "session_id": "...", "profile_id": "..."}` | Client should reload session history from REST (`GET /sessions/{id}`) |
 
 `session_sync` is sent in three situations:
@@ -204,6 +213,8 @@
 
 If the client reconnects after the run has already finished, there is no active `_AgentRun`, so it receives only `session_sync` and must fetch history via REST.
 
+**On the client side**, a socket that dies mid-stream leaves a half-written assistant message behind. That message must be dropped as soon as a new socket opens, before `session_sync` is handled: the server either replays the still-live run from scratch or answers with `session_sync` for a finished one, and a client that keeps the stale message (or treats itself as "still streaming") ignores both. The web client does exactly this in `useWebSocket.js` (`chat.dropStaleStream()` on `open`) and `chat.js` (`dropStaleStream`).
+
 ---
 
 ## Run state management
diff --git a/manuals/content_publish.md b/manuals/content_publish.md
index 79bbe3c..8f6fdb8 100644
--- a/manuals/content_publish.md
+++ b/manuals/content_publish.md
@@ -2,7 +2,7 @@
 
 ## Что делает
 
-Публикует файл из директории текущей сессии для просмотра пользователем прямо в чате. Поддерживаются интерактивные viewer'ы для 3D-моделей (STL), HTML-страниц, SVG, PDF, изображений и видео.
+Публикует файл из директории текущей сессии для просмотра пользователем прямо в чате. Поддерживаются интерактивные viewer'ы для 3D-моделей (STL, glTF/GLB), HTML-страниц, SVG, PDF, изображений, видео, аудио и текстовых/исходных файлов.
 
 Важно: `workspace/` и директория сессии — разные места.
 
@@ -67,17 +67,30 @@
 
 ## Поддерживаемые типы
 
-| Расширение | Тип | Viewer |
+| Расширение | `content_type` | Viewer |
 |---|---|---|
-| `.stl` | 3D-модель | Three.js 3D viewer |
-| `.html`, `.htm` | HTML | iframe |
-| `.svg` | SVG | iframe |
-| `.pdf` | PDF | iframe |
-| `.png`, `.jpg`, `.jpeg`, `.gif`, `.webp` | Изображение | img |
-| `.mp4`, `.webm` | Видео | video |
+| `.stl` | `stl` | Three.js, по расширению файла |
+| `.glb`, `.gltf` | `model` | тот же Three.js viewer (GLTFLoader) |
+| `.html`, `.htm` | `html` | iframe |
+| `.svg` | `svg` | iframe |
+| `.pdf` | `pdf` | iframe |
+| `.png`, `.jpg`, `.jpeg`, `.gif`, `.webp`, `.bmp`, `.avif`, `.ico` | `image` | img, по клику — на весь экран |
+| `.mp4`, `.webm`, `.mov`, `.mkv`, `.m4v`, `.ogv` | `video` | video |
+| `.mp3`, `.wav`, `.ogg`, `.oga`, `.opus`, `.m4a`, `.flac`, `.aac`, `.weba` | `audio` | audio-плеер |
+| `.txt`, `.md`, `.log`, `.csv`, `.json`, `.yaml`, `.xml`, `.py`, `.js`, `.ts`, `.scad` и другие текстовые/исходные | `text` | подсвеченный `<pre>` (первые 1 МБ) |
 
 Для неизвестных типов показывается карточка со ссылкой на скачивание.
 
+### Как определяется тип
+
+Тип определяет `navi/content_types.py` — сначала по расширению, затем по сигнатуре файла. Поэтому файл с неверным расширением всё равно открывается правильным viewer'ом: PNG, сохранённый как `.txt`, — изображение; переименованный `.mp3` — аудио; `.ts`, который не является MPEG-транспортным потоком, — текст (TypeScript), а не видео.
+
+Параметр `content_type` нужен только если автоопределение ошиблось; значение вне поддерживаемого списка игнорируется, и тип определяется по файлу. Передавать `content_type="unknown"` бессмысленно.
+
+Графический контент (изображение, SVG, HTML, PDF, 3D-модель, видео) можно развернуть на весь экран кнопкой в шапке карточки и в панели Artifacts.
+
+**glTF:** надёжно работает `.glb` — буферы и текстуры внутри файла. `.gltf`, ссылающийся на отдельные `.bin` и изображения, отрендерится только если все эти файлы тоже лежат в директории сессии (публикуется при этом только сам `.gltf`).
+
 ## Просмотр исходников
 
 Для `.svg` и `.html` пользователь может открыть исходник прямо из карточки/панели Artifacts — отдельный параметр не нужен, потому что опубликованный файл сам является исходником.
diff --git a/manuals/share_file.md b/manuals/share_file.md
index 3d854af..826efb1 100644
--- a/manuals/share_file.md
+++ b/manuals/share_file.md
@@ -17,7 +17,7 @@
 - Built binaries or packages
 - Any output file created by terminal, code_exec, or filesystem
 
-Do not use this just to create an inline preview. For SVG/HTML/PDF/image/video/STL preview cards, put the file in the session directory and call `content_publish`.
+Do not use this just to create an inline preview. For SVG/HTML/PDF/image/video/audio/text/STL/glTF preview cards, put the file in the session directory and call `content_publish`.
 
 ## How it works
 
diff --git a/navi/api/routes/sessions.py b/navi/api/routes/sessions.py
index d2e1390..f927319 100644
--- a/navi/api/routes/sessions.py
+++ b/navi/api/routes/sessions.py
@@ -24,6 +24,7 @@
 from navi.auth import User
 from navi.auth.deps import check_session_access
 from navi.content_store import list_for_session
+from navi.content_types import detect, is_inline_mime
 from navi.config import settings
 from navi.core import BackendRegistry, ProfileRegistry, SessionStore
 from navi.core.scheduler import RecallScheduler
@@ -522,11 +523,16 @@
     if not file_path.exists() or not file_path.is_file():
         raise HTTPException(status_code=404, detail="File not found")
 
-    # Detect content type; use inline disposition for browser-renderable types
-    content_type, _ = mimetypes.guess_type(file_path.name)
-    content_type = content_type or "application/octet-stream"
-    inline_types = {"image/", "text/html", "text/plain", "application/pdf"}
-    inline = not download and any(content_type.startswith(t) for t in inline_types)
+    # The session's own registry decides the type — the host's /etc/mime.types
+    # is the fallback for formats it does not cover (docx, archives), where a
+    # sensible MIME still makes the saved file openable.
+    _, content_type = detect(file_path.name, file_path)
+    if content_type == "application/octet-stream":
+        content_type = mimetypes.guess_type(file_path.name)[0] or content_type
+    # Video and audio belong in this list as much as images do: served as an
+    # attachment they are downloaded instead of played, even when the client
+    # asks the browser to open the URL rather than the in-page player.
+    inline = not download and is_inline_mime(content_type)
 
     return FileResponse(
         path=file_path,
diff --git a/navi/api/websocket.py b/navi/api/websocket.py
index 392c229..6c227a1 100644
--- a/navi/api/websocket.py
+++ b/navi/api/websocket.py
@@ -4,6 +4,7 @@
   {"type": "message", "content": "...", "cwd": "..."}
   {"type": "compact"}                      # force context compression now
   {"type": "form_submit", "form_id": "...", "values": {...}}   # navi_ui form submit
+  {"type": "ping", "ts": 1234567890}       # liveness probe; answered with pong
 
   `cwd` is optional but recommended for terminal clients; when present the
   server treats it as the user's working directory and resolves relative paths
@@ -52,7 +53,11 @@
     return None
 
 
-_HEARTBEAT_INTERVAL = 20.0  # seconds — keeps the browser from dropping idle connections
+# Seconds between keepalive frames. Sent both while a run streams (below) and
+# while the connection sits idle between turns (the main receive loop): a silent
+# socket is dropped by browsers and proxies, and a client resumed from the
+# background cannot tell a half-open socket from a live one without traffic.
+_HEARTBEAT_INTERVAL = 20.0
 
 
 async def _stream_to_client(websocket: WebSocket, queue: asyncio.Queue) -> bool:
@@ -415,7 +420,19 @@
             orchestrator.schedule_queued_drain(session_id, session_store)
 
         while True:
-            raw = await websocket.receive_text()
+            try:
+                raw = await asyncio.wait_for(
+                    websocket.receive_text(), timeout=_HEARTBEAT_INTERVAL
+                )
+            except TimeoutError:
+                # Nothing to do and nothing arrived — keep the idle socket warm
+                # (see _HEARTBEAT_INTERVAL). A send that fails means the peer is
+                # gone even though no disconnect frame arrived: stop reading.
+                try:
+                    await websocket.send_json({"type": "heartbeat"})
+                except Exception:  # noqa: BLE001 — a dead peer never answers
+                    break
+                continue
 
             try:
                 data = json.loads(raw)
@@ -425,6 +442,14 @@
 
             msg_type = data.get("type")
 
+            if msg_type == "ping":
+                # Liveness probe: any answer proves the socket still carries
+                # traffic end to end, which onclose alone never shows for a
+                # half-open connection. The timestamp comes back untouched so
+                # the client can measure the round trip.
+                await websocket.send_json({"type": "pong", "ts": data.get("ts")})
+                continue
+
             if msg_type == "compact":
                 # Forced context compression (TUI /compact). Runs the real
                 # compressor (bypassing the token threshold) and streams
diff --git a/navi/content_store.py b/navi/content_store.py
index e46a6d5..169823a 100644
--- a/navi/content_store.py
+++ b/navi/content_store.py
@@ -14,6 +14,7 @@
 import structlog
 
 from navi.config import settings
+from navi.content_types import content_type_of, normalize_content_type
 from navi.session_files import ensure_session_dir, session_dir
 
 if TYPE_CHECKING:
@@ -21,25 +22,9 @@
 
 log = structlog.get_logger()
 
-_EXT_TO_TYPE = {
-    ".stl": "stl",
-    ".html": "html",
-    ".htm": "html",
-    ".svg": "svg",
-    ".pdf": "pdf",
-    ".png": "image",
-    ".jpg": "image",
-    ".jpeg": "image",
-    ".gif": "image",
-    ".webp": "image",
-    ".mp4": "video",
-    ".webm": "video",
-}
 
-
-def _detect_content_type(filename: str) -> str:
-    ext = Path(filename).suffix.lower()
-    return _EXT_TO_TYPE.get(ext, "unknown")
+def _detect_content_type(filename: str, path: Path | None = None) -> str:
+    return content_type_of(filename, path)
 
 
 def _file_url(session_id: str, filename: str, *, download: bool = False) -> str:
@@ -129,7 +114,9 @@
     if not src.is_file():
         raise IsADirectoryError(f"Path is a directory, not a file: {src}")
 
-    detected_type = content_type or _detect_content_type(filename)
+    # An explicit type is a hint, not an override: a value we have no viewer
+    # for (or a stale one) falls back to what the file itself says.
+    detected_type = normalize_content_type(content_type) or _detect_content_type(filename, src)
     source_filename = _existing_source_filename(session_id, source_filename)
     safe_filename = filename.replace("/", "_").replace("\\", "_")
     content_id = f"{session_id}_{safe_filename}"
diff --git a/navi/content_types.py b/navi/content_types.py
new file mode 100644
index 0000000..ea34ce3
--- /dev/null
+++ b/navi/content_types.py
@@ -0,0 +1,305 @@
+"""What a file in a session directory actually is.
+
+Five extension tables used to answer this question, and they disagreed: one in
+``content_store``, one in ``content_publish``'s schema, and three in the
+webclient. A ``.md`` report came back "unknown" and got a download link, while
+``mimetypes`` — which reads the host's ``/etc/mime.types`` — served a
+TypeScript ``.ts`` file as ``video/mp2t`` because the extension is shared with
+MPEG transport streams.
+
+Detection order:
+
+1. A strong binary signature in the file's opening kilobytes. Text cannot
+   start with the PNG or RIFF magic, so the signature wins over the extension:
+   a file saved under the wrong name still reaches the right viewer.
+2. The extension table, which answers everything without a telltale header.
+3. For a file with no usable extension: does it look like text?
+4. Otherwise unknown.
+
+The values in ``CONTENT_TYPES`` are what ``content_publish`` accepts and what
+the webclient branches on; adding one means adding a viewer branch there too.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+CONTENT_TYPES = (
+    "stl",
+    "model",
+    "html",
+    "svg",
+    "pdf",
+    "image",
+    "video",
+    "audio",
+    "text",
+    "unknown",
+)
+
+# Enough for every signature below, including a Matroska track entry.
+_SAMPLE_BYTES = 4096
+
+# (content type, MIME). `stl` and `model` share one viewer: the file's own
+# extension decides which three.js loader opens it.
+_EXTENSION_TYPES: dict[str, tuple[str, str]] = {
+    # 3D models
+    ".stl": ("stl", "model/stl"),
+    ".glb": ("model", "model/gltf-binary"),
+    ".gltf": ("model", "model/gltf+json"),
+    # Documents
+    ".html": ("html", "text/html"),
+    ".htm": ("html", "text/html"),
+    ".svg": ("svg", "image/svg+xml"),
+    ".pdf": ("pdf", "application/pdf"),
+    # Images
+    ".png": ("image", "image/png"),
+    ".jpg": ("image", "image/jpeg"),
+    ".jpeg": ("image", "image/jpeg"),
+    ".gif": ("image", "image/gif"),
+    ".webp": ("image", "image/webp"),
+    ".bmp": ("image", "image/bmp"),
+    ".avif": ("image", "image/avif"),
+    ".ico": ("image", "image/x-icon"),
+    # Video
+    ".mp4": ("video", "video/mp4"),
+    ".webm": ("video", "video/webm"),
+    ".mov": ("video", "video/quicktime"),
+    ".mkv": ("video", "video/x-matroska"),
+    ".m4v": ("video", "video/x-m4v"),
+    ".ogv": ("video", "video/ogg"),
+    # Audio
+    ".mp3": ("audio", "audio/mpeg"),
+    ".wav": ("audio", "audio/wav"),
+    ".ogg": ("audio", "audio/ogg"),
+    ".oga": ("audio", "audio/ogg"),
+    ".opus": ("audio", "audio/ogg"),
+    ".m4a": ("audio", "audio/mp4"),
+    ".flac": ("audio", "audio/flac"),
+    ".aac": ("audio", "audio/aac"),
+    ".weba": ("audio", "audio/webm"),
+    # Text and source. `.ts` belongs here, not to video: without an MPEG
+    # transport-stream signature a .ts file is TypeScript.
+    ".txt": ("text", "text/plain"),
+    ".text": ("text", "text/plain"),
+    ".log": ("text", "text/plain"),
+    ".md": ("text", "text/markdown"),
+    ".markdown": ("text", "text/markdown"),
+    ".csv": ("text", "text/csv"),
+    ".tsv": ("text", "text/tab-separated-values"),
+    ".json": ("text", "application/json"),
+    ".jsonl": ("text", "application/json"),
+    ".ndjson": ("text", "application/json"),
+    ".yaml": ("text", "application/yaml"),
+    ".yml": ("text", "application/yaml"),
+    ".xml": ("text", "text/xml"),
+    ".ini": ("text", "text/plain"),
+    ".cfg": ("text", "text/plain"),
+    ".conf": ("text", "text/plain"),
+    ".toml": ("text", "text/plain"),
+    ".py": ("text", "text/x-python"),
+    ".js": ("text", "text/javascript"),
+    ".mjs": ("text", "text/javascript"),
+    ".cjs": ("text", "text/javascript"),
+    ".ts": ("text", "text/plain"),
+    ".tsx": ("text", "text/plain"),
+    ".jsx": ("text", "text/javascript"),
+    ".css": ("text", "text/css"),
+    ".scss": ("text", "text/plain"),
+    ".sass": ("text", "text/plain"),
+    ".less": ("text", "text/plain"),
+    ".sh": ("text", "text/x-sh"),
+    ".bash": ("text", "text/x-sh"),
+    ".zsh": ("text", "text/x-sh"),
+    ".sql": ("text", "text/plain"),
+    ".scad": ("text", "text/plain"),
+    ".srt": ("text", "text/plain"),
+    ".vtt": ("text", "text/vtt"),
+    ".diff": ("text", "text/plain"),
+    ".patch": ("text", "text/plain"),
+    ".tex": ("text", "text/plain"),
+    ".c": ("text", "text/x-csrc"),
+    ".h": ("text", "text/x-chdr"),
+    ".cpp": ("text", "text/x-c++src"),
+    ".hpp": ("text", "text/x-c++hdr"),
+    ".java": ("text", "text/x-java"),
+    ".go": ("text", "text/plain"),
+    ".rs": ("text", "text/plain"),
+    ".rb": ("text", "text/plain"),
+    ".php": ("text", "text/plain"),
+    ".kt": ("text", "text/plain"),
+    ".swift": ("text", "text/plain"),
+    # Opaque containers: no viewer, but the MIME still matters for the download
+    # and keeps the text heuristic from calling a small archive "text".
+    ".zip": ("unknown", "application/zip"),
+    ".tar": ("unknown", "application/x-tar"),
+    ".gz": ("unknown", "application/gzip"),
+    ".tgz": ("unknown", "application/gzip"),
+    ".7z": ("unknown", "application/x-7z-compressed"),
+    ".rar": ("unknown", "application/vnd.rar"),
+    ".docx": ("unknown", "application/vnd.openxmlformats-officedocument.wordprocessingml.document"),
+    ".xlsx": ("unknown", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"),
+    ".pptx": ("unknown", "application/vnd.openxmlformats-officedocument.presentationml.presentation"),
+    ".odt": ("unknown", "application/vnd.oasis.opendocument.text"),
+    ".ods": ("unknown", "application/vnd.oasis.opendocument.spreadsheet"),
+    ".rtf": ("unknown", "application/rtf"),
+    ".epub": ("unknown", "application/epub+zip"),
+}
+
+# MIME families a browser can render itself. `video/` and `audio/` used to be
+# missing here, so every media file was sent as an attachment and the player
+# got a download instead of a stream.
+_INLINE_PREFIXES = ("image/", "audio/", "video/", "text/")
+_INLINE_MIMES = frozenset({"application/pdf", "application/json"})
+
+_VIDEO_CODECS = (b"V_VP8", b"V_VP9", b"V_AV1", b"V_MPEG4", b"V_MPEGH", b"V_THEORA")
+_AUDIO_CODECS = (b"A_OPUS", b"A_VORBIS", b"A_AAC", b"A_FLAC", b"A_MPEG")
+
+
+def _read_head(path: str | Path | None) -> bytes:
+    if path is None:
+        return b""
+    try:
+        with Path(path).open("rb") as handle:
+            return handle.read(_SAMPLE_BYTES)
+    except OSError:
+        return b""
+
+
+def _mpeg_audio_frame(head: bytes) -> bool:
+    """An MP3 without an ID3 tag starts with a frame header, not with a magic.
+
+    The check is deliberately narrow — reserved layer and invalid bitrate
+    are rejected — so ordinary binary data does not read as audio.
+    """
+    if len(head) < 4 or head[0] != 0xFF or (head[1] & 0xE0) != 0xE0:
+        return False
+    layer = (head[1] >> 1) & 0x03
+    bitrate = (head[2] >> 4) & 0x0F
+    return layer != 0 and bitrate != 0x0F
+
+
+def _mpeg_transport_stream(head: bytes) -> bool:
+    """MPEG-TS: 188-byte packets, each starting with 0x47."""
+    if len(head) < 377:
+        return False
+    return head[0] == 0x47 and head[188] == 0x47 and head[376] == 0x47
+
+
+def _matroska(head: bytes) -> tuple[str, str] | None:
+    """WebM/MKV, audio or video by the tracks it declares."""
+    is_webm = b"webm" in head[:64]
+    if any(codec in head for codec in _VIDEO_CODECS):
+        return "video", "video/webm" if is_webm else "video/x-matroska"
+    if any(codec in head for codec in _AUDIO_CODECS):
+        return "audio", "audio/webm" if is_webm else "audio/x-matroska"
+    return None
+
+
+def _signature(head: bytes) -> tuple[str, str] | None:
+    """A binary format the opening bytes give away, or None."""
+    if not head:
+        return None
+    if head.startswith(b"%PDF-"):
+        return "pdf", "application/pdf"
+    if head.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image", "image/png"
+    if head.startswith(b"\xff\xd8\xff"):
+        return "image", "image/jpeg"
+    if head.startswith((b"GIF87a", b"GIF89a")):
+        return "image", "image/gif"
+    if head.startswith(b"RIFF") and len(head) >= 12:
+        kind = head[8:12]
+        if kind == b"WEBP":
+            return "image", "image/webp"
+        if kind == b"WAVE":
+            return "audio", "audio/wav"
+        if kind == b"AVI ":
+            return "video", "video/x-msvideo"
+    if head.startswith(b"OggS"):
+        return ("video", "video/ogg") if b"theora" in head[:1024] else ("audio", "audio/ogg")
+    if head.startswith(b"fLaC"):
+        return "audio", "audio/flac"
+    if head.startswith(b"ID3"):
+        return "audio", "audio/mpeg"
+    if _mpeg_audio_frame(head):
+        return "audio", "audio/mpeg"
+    if head.startswith(b"\x1a\x45\xdf\xa3"):
+        return _matroska(head)
+    if len(head) >= 12 and head[4:8] == b"ftyp":
+        brand = head[8:12]
+        if brand in {b"M4A ", b"M4B ", b"M4P ", b"mp4a"}:
+            return "audio", "audio/mp4"
+        if brand in {b"avif", b"avis"}:
+            return "image", "image/avif"
+        return "video", "video/mp4"
+    if head.startswith(b"glTF"):
+        return "model", "model/gltf-binary"
+    if _mpeg_transport_stream(head):
+        return "video", "video/mp2t"
+    return None
+
+
+def _looks_like_text(head: bytes) -> bool:
+    """True when the sample decodes as text rather than as arbitrary bytes."""
+    if not head or b"\x00" in head:
+        return False
+    try:
+        text = head.decode("utf-8")
+    except UnicodeDecodeError:
+        # A UTF-8 sequence cut at the sample boundary, or a legacy encoding;
+        # latin-1 always decodes, and the printable ratio still separates
+        # prose from compressed data.
+        text = head.decode("latin-1")
+    if not text:
+        return False
+    printable = sum(1 for ch in text if ch.isprintable() or ch in "\n\r\t\f\v")
+    return printable / len(text) >= 0.9
+
+
+def detect(filename: str, path: str | Path | None = None) -> tuple[str, str]:
+    """Return ``(content_type, mime)`` for a file.
+
+    ``path`` is read for its signature; without it only the name is used.
+    """
+    suffix = Path(filename).suffix.lower()
+    head = _read_head(path)
+
+    found = _signature(head)
+    if found is not None:
+        return found
+
+    known = _EXTENSION_TYPES.get(suffix)
+    if known is not None:
+        return known
+
+    if _looks_like_text(head):
+        return "text", "text/plain"
+    return "unknown", "application/octet-stream"
+
+
+def content_type_of(filename: str, path: str | Path | None = None) -> str:
+    return detect(filename, path)[0]
+
+
+def mime_of(filename: str, path: str | Path | None = None) -> str:
+    return detect(filename, path)[1]
+
+
+def normalize_content_type(value: str | None) -> str | None:
+    """An explicitly requested type, or None when the caller should detect.
+
+    A model naming a type we do not have a viewer for is not an error; it is a
+    missing hint, so detection answers instead of the value being stored as is.
+    """
+    if not value:
+        return None
+    value = value.strip().lower()
+    if value in CONTENT_TYPES and value != "unknown":
+        return value
+    return None
+
+
+def is_inline_mime(mime: str) -> bool:
+    """Whether a browser can show this response in place instead of saving it."""
+    return mime.startswith(_INLINE_PREFIXES) or mime in _INLINE_MIMES
diff --git a/navi/core/agent.py b/navi/core/agent.py
index f0bd575..7a4a7c9 100644
--- a/navi/core/agent.py
+++ b/navi/core/agent.py
@@ -322,13 +322,15 @@
             if not settings.context_compression_enabled:
                 raise NothingToCompactError("Context compression is disabled.")
 
+            window = self._window_for(profile)
             yield CompressionStarted(
                 # Real-baseline estimate when available (see _compression_events_preturn);
-                # heuristic is only the fallback before the first LLM call of a session.
+                # the calibrated estimate is only the fallback before the first LLM
+                # call of a session.
                 context_tokens=self._compressor.real_baseline_estimate(
-                    session.context, session.context
+                    session.context, session.context, profile.model
                 ),
-                max_context_tokens=settings.ollama_num_ctx,
+                max_context_tokens=window,
             )
             event = await self._compressor.compress_and_save_session(
                 session=session,
@@ -347,6 +349,7 @@
                 # summarize when turns <= keep_recent and forced compact always
                 # reports "nothing to compact". Mirrors the midturn auto-compress.
                 keep_recent_messages=max(12, settings.context_keep_recent * 2),
+                window=window,
             )
             if event is None:
                 raise NothingToCompactError(
@@ -467,8 +470,10 @@
             images_for_context = images
             context_content = user_message
             if images:
-                current_tokens = self._compressor.estimate_context_tokens(session.context)
-                available_tokens = int(settings.ollama_num_ctx * 0.8) - current_tokens
+                current_tokens = self._compressor.estimate_tokens(
+                    session.context, profile.model
+                )
+                available_tokens = int(self._window_for(profile) * 0.8) - current_tokens
                 max_images = max(0, available_tokens // 500)
                 if max_images < len(images):
                     images_for_context = images[:max_images]
@@ -580,7 +585,12 @@
                 pending_final_nudge = None
 
             try:
-                self._compressor.check_context_size(built_ctx, session_context=session.context)
+                self._compressor.check_context_size(
+                    built_ctx,
+                    session_context=session.context,
+                    model=profile.model,
+                    window=self._window_for(profile),
+                )
             except ContextTooLargeError as e:
                 # Surface the error as a Navi response (not a raw system error) so the
                 # user sees a coherent message and the exchange is saved to history.
@@ -590,7 +600,13 @@
                 session.messages.append(assistant_msg)
                 await self._sessions.save(session)
                 yield TextDelta(delta=error_text)
-                yield StreamEnd(content=error_text)
+                # This path was never exercised (the old chars//3 check never
+                # fired), so it kept an invalid StreamEnd(content=...) and died
+                # with a TypeError instead of reporting the overflow.
+                yield StreamEnd(
+                    full_content=error_text,
+                    max_context_tokens=self._window_for(profile),
+                )
                 return
 
             async for _ev in self._consume_stream(
@@ -604,6 +620,9 @@
                         top_k=profile.top_k,
                         top_p=profile.top_p,
                         num_thread=profile.num_thread,
+                        # The window the compressor accounts against must be the
+                        # window the local model is actually loaded with.
+                        num_ctx=self._window_for(profile),
                     ),
                     stop_event=stop_event,
                     first_chunk_timeout=settings.llm_stream_first_chunk_timeout,
@@ -621,7 +640,12 @@
             # undercounts code-heavy tool output and fires compression too late).
             # len(session.context) here is exactly what was sent — the new
             # assistant/tool messages are appended only below.
-            self._compressor.record_real_baseline(len(session.context), state.context_tokens)
+            self._compressor.record_real_baseline(
+                len(session.context),
+                state.context_tokens,
+                chars=self._compressor.context_chars(built_ctx),
+                model=turn_ctx.resolved_model or profile.model,
+            )
 
             # Stopped mid-stream — save partial response and exit
             if stop_event and stop_event.is_set():
@@ -690,7 +714,7 @@
                 yield StreamEnd(
                     full_content=state.accumulated_text,
                     context_tokens=state.context_tokens,
-                    max_context_tokens=settings.ollama_num_ctx,
+                    max_context_tokens=self._window_for(profile),
                     elapsed_seconds=_elapsed,
                     tool_call_count=turn_ctx.tool_call_count,
                     token_count=_net_tokens if _net_tokens else None,
@@ -873,7 +897,7 @@
         ctx = WorkerContext(
             session_id=session.id,
             context_tokens=context_tokens,
-            max_context_tokens=settings.ollama_num_ctx,
+            max_context_tokens=self._window_for(profile),
             llm=llm,
             model=model,
             temperature=settings.context_summary_temperature,
@@ -911,6 +935,19 @@
     def _get_backend(self, backend_key: str) -> LLMBackend:
         return self._backends.get(backend_key)
 
+    @staticmethod
+    def _window_for(profile: "AgentProfile | None") -> int:
+        """Context window this profile's model is accounted against.
+
+        One global OLLAMA_NUM_CTX used to stand in for every model — a local
+        65k model and a cloud model fed 115k prompts were held to the same
+        number. A profile may now declare its own ``context_window``.
+        """
+        return (
+            getattr(profile, "context_window", None)
+            or settings.ollama_num_ctx
+        )
+
     async def _todo_updated(self, session) -> TodoUpdated:
         """Build a TodoUpdated event from the session's current todo state.
 
@@ -924,20 +961,21 @@
         return TodoUpdated(session_id=session.id, tasks=tasks)
 
     async def _compression_events_preturn(self, session, llm, profile, session_id):
-        # real_baseline_estimate, not the chars//3 heuristic: the baseline is
-        # valid here (the context only grew since the last LLM call), and the
-        # heuristic undercounts code-heavy tool output — firing compression
-        # too late is exactly what the baseline machinery was built to fix.
-        # The midturn gate below does the same.
+        # real_baseline_estimate, not a character heuristic: the baseline is
+        # valid here (the context only grew since the last LLM call), and a
+        # character count undercounts code-heavy tool output — firing
+        # compression too late is exactly what the baseline machinery was
+        # built to fix. The midturn gate below does the same.
         estimated_tokens = self._compressor.real_baseline_estimate(
-            session.context, session.context
+            session.context, session.context, profile.model
         )
+        window = self._window_for(profile)
         if (
             settings.context_compression_enabled
             and len(session.context) > 2
             and should_compress(
                 estimated_tokens,
-                settings.ollama_num_ctx,
+                window,
                 settings.context_compression_threshold,
             )
         ):
@@ -951,12 +989,14 @@
                 session.context,
                 keep_recent=settings.context_keep_recent,
                 threshold=settings.context_compression_threshold,
-                max_context_tokens=settings.ollama_num_ctx,
+                max_context_tokens=window,
+                current_tokens=estimated_tokens,
+                model=profile.model,
             ):
                 return
             yield CompressionStarted(
                 context_tokens=estimated_tokens,
-                max_context_tokens=settings.ollama_num_ctx,
+                max_context_tokens=window,
             )
             event = await self._compressor.compress_and_save_session(
                 session=session,
@@ -968,6 +1008,7 @@
                 reason="preturn",
                 keep_recent=settings.context_keep_recent,
                 max_tokens=settings.context_summary_max_tokens,
+                window=window,
             )
             if event:
                 yield event
@@ -984,11 +1025,12 @@
                 session_id=session_id,
             )
             estimated_tokens = self._compressor.real_baseline_estimate(
-                session.context, preflight_ctx
+                session.context, preflight_ctx, profile.model
             )
+            window = self._window_for(profile)
             if should_compress(
                 estimated_tokens,
-                settings.ollama_num_ctx,
+                window,
                 settings.context_compression_threshold,
             ):
                 # See _compression_events_preturn: only announce compression when
@@ -1002,12 +1044,14 @@
                     keep_recent=settings.context_keep_recent,
                     keep_recent_messages=max(12, settings.context_keep_recent * 2),
                     threshold=settings.context_compression_threshold,
-                    max_context_tokens=settings.ollama_num_ctx,
+                    max_context_tokens=window,
+                    current_tokens=estimated_tokens,
+                    model=profile.model,
                 ):
                     return
                 yield CompressionStarted(
                     context_tokens=estimated_tokens,
-                    max_context_tokens=settings.ollama_num_ctx,
+                    max_context_tokens=window,
                 )
                 event = await self._compressor.compress_and_save_session(
                     session=session,
@@ -1020,6 +1064,7 @@
                     keep_recent=settings.context_keep_recent,
                     max_tokens=settings.context_summary_max_tokens,
                     keep_recent_messages=max(12, settings.context_keep_recent * 2),
+                    window=window,
                 )
                 if event:
                     yield event
diff --git a/navi/core/compressor.py b/navi/core/compressor.py
index b6f55ea..95c35cf 100644
--- a/navi/core/compressor.py
+++ b/navi/core/compressor.py
@@ -19,11 +19,13 @@
 import structlog
 from datetime import datetime, timezone
 from pathlib import Path
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 from navi.llm.base import LLMBackend, Message
 from navi.config import settings
 from .events import ContextCompressed
+from .token_calibration import calibration_for
 
 if TYPE_CHECKING:
     from navi.profiles.base import AgentProfile
@@ -403,6 +405,9 @@
     context: list[Message],
     keep_recent: int,
     keep_recent_messages: int | None = None,
+    *,
+    window: int | None = None,
+    estimate: Callable[[list[Message]], int] | None = None,
 ) -> tuple[list[Message], list[Message], list[Message]] | None:
     """Decide what to summarize and what to keep — the single decision point
     shared by ``compress_context`` (real compression) and
@@ -433,15 +438,18 @@
     )
 
     if keep_recent_messages is None:
-        target_tokens = int(settings.ollama_num_ctx * settings.context_compression_target)
-        kept_tokens = ContextCompressor.estimate_context_tokens(system_msgs + to_keep)
+        est = estimate or ContextCompressor.estimate_context_tokens
+        target_tokens = int(
+            (window or settings.ollama_num_ctx) * settings.context_compression_target
+        )
+        kept_tokens = est(system_msgs + to_keep)
         _KEEP_RECENT_FLOOR = 2
         while kept_tokens > target_tokens and keep_recent > _KEEP_RECENT_FLOOR:
             keep_recent -= 1
             to_summarize, to_keep = partition_messages(
                 context, keep_recent, keep_recent_messages=keep_recent_messages
             )
-            kept_tokens = ContextCompressor.estimate_context_tokens(system_msgs + to_keep)
+            kept_tokens = est(system_msgs + to_keep)
 
     if len(to_summarize) < 2 and keep_recent_messages is not None and keep_recent_messages > 2:
         to_summarize, to_keep = partition_messages(
@@ -464,6 +472,8 @@
     max_tokens: int | None = None,
     keep_recent_messages: int | None = None,
     profile: "AgentProfile | None" = None,
+    window: int | None = None,
+    estimate: Callable[[list[Message]], int] | None = None,
 ) -> tuple[list[Message], str] | None:
     """
     Summarize old messages in the LLM context and return a shorter context list.
@@ -488,7 +498,11 @@
     effective_max_tokens = getattr(profile, "compression_max_tokens", None) or max_tokens
 
     planned = _plan_compression(
-        context, effective_keep_recent, keep_recent_messages=keep_recent_messages
+        context,
+        effective_keep_recent,
+        keep_recent_messages=keep_recent_messages,
+        window=window,
+        estimate=estimate,
     )
     if planned is None:
         return None  # nothing substantial to compress
@@ -554,17 +568,50 @@
     """
 
     @staticmethod
-    def estimate_context_tokens(context: list[Message]) -> int:
-        """Conservative local estimate used before the next LLM call returns real token counts.
+    def context_chars(context: list[Message]) -> int:
+        """Characters of message text — the input to every token estimate."""
+        return sum(len(m.content or "") for m in context)
 
-        Uses ~3 chars per token (more conservative than the naive 4) because code and
-        punctuation are often 1 token per character. Images counted at 500 tokens each
-        (rough vision-model estimate).
+    @staticmethod
+    def estimate_context_tokens(context: list[Message]) -> int:
+        """Raw chars // 3 heuristic, no calibration.
+
+        Kept as the reference unit for tests and for callers that want the
+        old, deliberately under-counting number. Window decisions must use
+        ``estimate_tokens`` instead — on prod this heuristic landed at ~56k
+        where the real prompt was 101k, and targets expressed in it were met
+        while the real prompt sat above the window.
         """
-        chars = sum(len(m.content or "") for m in context)
+        chars = ContextCompressor.context_chars(context)
         imgs = sum(500 * len(m.images) for m in context if m.images)
         return chars // 3 + imgs
 
+    @staticmethod
+    def estimate_tokens(context: list[Message], model=None) -> int:
+        """Token estimate in real-prompt units, used by every window decision.
+
+        ``estimate_context_tokens`` converted to real prompt tokens by the
+        calibration fitted for ``model`` (see ``navi/core/token_calibration.py``
+        for why chars // 3 is not usable as-is). Without observations the
+        calibration falls back to ``DEFAULT_CHARS_PER_TOKEN``.
+        """
+        chars = ContextCompressor.context_chars(context)
+        imgs = sum(500 * len(m.images) for m in context if m.images)
+        return calibration_for(model).tokens_for(chars) + imgs
+
+    @staticmethod
+    def estimate_tokens_delta(context: list[Message], model=None) -> int:
+        """Price messages *appended since* a real baseline, same units.
+
+        Uses the calibration slope without its intercept: the fixed prompt
+        overhead is already part of the baseline count. Pricing a delta with
+        the raw ``chars // 3`` left the pre/mid-turn gates in one unit and the
+        window in another for exactly the tool-heavy turns they exist for.
+        """
+        chars = ContextCompressor.context_chars(context)
+        imgs = sum(500 * len(m.images) for m in context if m.images)
+        return calibration_for(model).delta_for(chars) + imgs
+
     def __init__(self) -> None:
         self._profile: "AgentProfile | None" = None
         # Real prompt-token baseline from the last LLM call:
@@ -579,32 +626,53 @@
         """Tell the compressor which profile is active so it can use profile-specific settings."""
         self._profile = profile
 
-    def record_real_baseline(self, context_len: int, real_tokens: int | None) -> None:
+    def record_real_baseline(
+        self,
+        context_len: int,
+        real_tokens: int | None,
+        *,
+        chars: int | None = None,
+        model=None,
+    ) -> None:
         """Cache the real prompt-token count returned by the last LLM call.
 
         ``context_len`` is ``len(session.context)`` at the time of that call
         (before the new assistant/tool messages are appended). The next
         estimate reuses these real tokens for the bulk of the context and only
         heuristics the new messages appended since.
+
+        ``chars`` is the character count of the exact prompt that was sent
+        (the built context, not just session.context). Together with
+        ``real_tokens`` it feeds the per-model calibration line, which is what
+        converts hypothetical contexts (the proposed kept region) into real
+        tokens.
         """
         if real_tokens and real_tokens > 0:
             self._real_baseline = (context_len, int(real_tokens))
+            if chars:
+                calibration_for(model).observe(chars, real_tokens)
 
     def real_baseline_estimate(
-        self, session_context: list[Message], built_ctx: list[Message]
+        self,
+        session_context: list[Message],
+        built_ctx: list[Message],
+        model=None,
     ) -> int:
         """Estimate context tokens using the real baseline when available.
 
-        Falls back to the chars//3 heuristic on ``built_ctx`` when no baseline
-        is recorded or it is stale (the session shrank, e.g. right after
-        compression — ``base_len`` exceeds the current length).
+        Falls back to the calibrated ``estimate_tokens`` on ``built_ctx`` when
+        no baseline is recorded or it is stale (the session shrank, e.g. right
+        after compression — ``base_len`` exceeds the current length). Every
+        branch is in real prompt tokens, the unit the window is measured in.
         """
         if self._real_baseline is not None:
             base_len, base_tokens = self._real_baseline
             if base_len <= len(session_context):
-                delta = self.estimate_context_tokens(session_context[base_len:])
+                delta = self.estimate_tokens_delta(
+                    session_context[base_len:], model
+                )
                 return base_tokens + delta
-        return self.estimate_context_tokens(built_ctx)
+        return self.estimate_tokens(built_ctx, model)
 
     async def compress_session(
         self,
@@ -615,6 +683,7 @@
         keep_recent: int,
         max_tokens: int | None = None,
         keep_recent_messages: int | None = None,
+        window: int | None = None,
     ) -> tuple[list[Message], str] | None:
         """Compress context with retry + hard-truncate fallback.
 
@@ -624,6 +693,13 @@
         """
         effective_keep_recent = getattr(self._profile, "compression_keep_recent", None) or keep_recent
         effective_max_tokens = getattr(self._profile, "compression_max_tokens", None) or max_tokens
+        window_tokens = window or settings.ollama_num_ctx
+        # Every size decision in this pass — the keep-region target, the
+        # post-check, the truncate budgets — must be in real-token units, the
+        # same units the trigger fired in. Mixed units are what let a
+        # compression pass finish at "65% of the window" while the real prompt
+        # it produced was above the window.
+        est = lambda msgs: self.estimate_tokens(msgs, model)
 
         # Attempt 1: normal compression
         try:
@@ -636,6 +712,8 @@
                 max_tokens=effective_max_tokens,
                 keep_recent_messages=keep_recent_messages,
                 profile=self._profile,
+                window=window_tokens,
+                estimate=est,
             )
         except Exception:
             # Attempt 2: keep more recent turns verbatim
@@ -651,10 +729,12 @@
                     if keep_recent_messages is not None
                     else None,
                     profile=self._profile,
+                    window=window_tokens,
+                    estimate=est,
                 )
             except Exception:
                 # Attempt 3: hard-truncate fallback
-                return self._hard_truncate(context)
+                return self._hard_truncate(context, window=window_tokens, estimate=est)
 
         if result is None:
             # Partition found nothing to summarize (e.g. few very large messages
@@ -664,14 +744,14 @@
             # the window overflows. (The per-message truncation in
             # context_builder.build caps single huge messages for the LLM view;
             # this is the stored-context safety net.)
-            from navi.config import settings
-
             if should_compress(
-                self.estimate_context_tokens(context),
-                settings.ollama_num_ctx,
+                est(context),
+                window_tokens,
                 settings.context_compression_threshold,
             ):
-                return self._token_budget_truncate(context)
+                return self._token_budget_truncate(
+                    context, estimate=est, window=window_tokens
+                )
             return None
 
         # Target hysteresis safety net: if the compressed context is still over
@@ -683,13 +763,11 @@
         # turns were kept verbatim (not summarized) and are lost; this only
         # triggers when the verbatim kept region alone exceeds the target.
         new_context, summary_text = result
-        # Access settings via the module (a local `from navi.config import settings`
-        # above shadows the module-level binding inside this function).
-        import navi.config as _cfg
-
-        target_tokens = int(_cfg.settings.ollama_num_ctx * _cfg.settings.context_compression_target)
-        if self.estimate_context_tokens(new_context) > target_tokens:
-            truncated = self._token_budget_truncate(new_context, budget_tokens=target_tokens)
+        target_tokens = int(window_tokens * settings.context_compression_target)
+        if est(new_context) > target_tokens:
+            truncated = self._token_budget_truncate(
+                new_context, budget_tokens=target_tokens, estimate=est, window=window_tokens
+            )
             if truncated is not None:
                 new_context = truncated[0]
         return new_context, summary_text
@@ -701,6 +779,8 @@
         keep_recent_messages: int | None = None,
         threshold: float | None = None,
         max_context_tokens: int | None = None,
+        current_tokens: int | None = None,
+        model=None,
     ) -> bool:
         """True if ``compress_session`` would actually shrink the context.
 
@@ -712,8 +792,14 @@
         (no misleading "compression" status on a no-op).
         """
         effective_keep_recent = getattr(self._profile, "compression_keep_recent", None) or keep_recent
+        window_tokens = max_context_tokens or settings.ollama_num_ctx
+        est = lambda msgs: self.estimate_tokens(msgs, model)
         if _plan_compression(
-            context, effective_keep_recent, keep_recent_messages=keep_recent_messages
+            context,
+            effective_keep_recent,
+            keep_recent_messages=keep_recent_messages,
+            window=window_tokens,
+            estimate=est,
         ) is not None:
             return True
         # Partition no-op: compression can still shrink via the token-budget
@@ -722,15 +808,23 @@
         # split by message-boundary truncation, and is handled instead by the
         # per-message view truncation in context_builder.build). Mirror that
         # decision so would_compress and compress_session agree exactly.
+        # The size compared against the threshold is the caller's real-token
+        # estimate (`current_tokens`), not a fresh heuristic: the trigger fired
+        # on real tokens, and judging it again in chars // 3 units made the two
+        # disagree by ~1.8x.
         if threshold is not None and max_context_tokens is not None:
-            if should_compress(
-                self.estimate_context_tokens(context), max_context_tokens, threshold
-            ):
-                return self._token_budget_truncate(context) is not None
+            total = current_tokens if current_tokens is not None else est(context)
+            if should_compress(total, window_tokens, threshold):
+                return self._token_budget_truncate(
+                    context, estimate=est, window=window_tokens
+                ) is not None
         return False
 
     def _hard_truncate(
-        self, context: list[Message]
+        self,
+        context: list[Message],
+        window: int | None = None,
+        estimate: Callable[[list[Message]], int] | None = None,
     ) -> tuple[list[Message], str] | None:
         """Last-resort fallback: drop oldest non-system messages without summarizing.
 
@@ -740,13 +834,17 @@
         messages — the fixed-count floor used to no-op on <= 6 messages even
         when they were huge.
         """
-        from navi.config import settings
-
-        budget = max(1, int(settings.ollama_num_ctx * _HARD_TRUNCATE_TOKEN_FRAC))
-        return self._token_budget_truncate(context, budget_tokens=budget)
+        budget = max(1, int((window or settings.ollama_num_ctx) * _HARD_TRUNCATE_TOKEN_FRAC))
+        return self._token_budget_truncate(
+            context, budget_tokens=budget, estimate=estimate, window=window
+        )
 
     def _token_budget_truncate(
-        self, context: list[Message], budget_tokens: int | None = None
+        self,
+        context: list[Message],
+        budget_tokens: int | None = None,
+        estimate: Callable[[list[Message]], int] | None = None,
+        window: int | None = None,
     ) -> tuple[list[Message], str] | None:
         """Drop oldest non-system messages (by turn groups) until under budget.
 
@@ -755,10 +853,12 @@
         budget or there is nothing to drop. ``budget_tokens`` defaults to half
         the context window.
         """
-        from navi.config import settings
+        est = estimate or ContextCompressor.estimate_context_tokens
 
         if budget_tokens is None:
-            budget_tokens = max(1, int(settings.ollama_num_ctx * _HARD_TRUNCATE_TOKEN_FRAC))
+            budget_tokens = max(
+                1, int((window or settings.ollama_num_ctx) * _HARD_TRUNCATE_TOKEN_FRAC)
+            )
 
         system_msgs = [m for m in context if m.role == "system"]
         non_system = [m for m in context if m.role != "system"]
@@ -783,7 +883,7 @@
         kept_turns: list[list[Message]] = []
         used = 0
         for turn in reversed(turns):
-            turn_tokens = self.estimate_context_tokens(turn)
+            turn_tokens = est(turn)
             if kept_turns and used + turn_tokens > budget_tokens:
                 break
             kept_turns.insert(0, turn)
@@ -802,34 +902,47 @@
         return new_context, summary_text
 
     def check_context_size(
-        self, context: list[Message], session_context: list[Message] | None = None
+        self,
+        context: list[Message],
+        session_context: list[Message] | None = None,
+        model=None,
+        window: int | None = None,
     ) -> None:
         """Raise ContextTooLargeError before an LLM call if the context is dangerously large."""
-        from navi.config import settings
+        # Local import: the window and the reserve are read at call time, so a
+        # runtime settings swap (tests, operator reload) reaches this check.
+        from navi.config import settings as _settings
         from navi.exceptions import ContextTooLargeError
 
         if not context:
             return
 
-        output_reserve = settings.output_reserve_tokens
-        # When the caller supplies the raw session context, prefer the real-token
-        # baseline (real bulk + heuristic delta) over the pure chars//3 estimate,
-        # so a code-heavy context is caught before it overflows the window.
-        total = (
-            self.real_baseline_estimate(session_context, context)
-            if session_context is not None
-            else self.estimate_context_tokens(context)
+        window_tokens = window or _settings.ollama_num_ctx
+        # A reserve bigger than the window turns `available` negative and makes
+        # every check fail, however small the context — a 500-token model still
+        # has to be able to answer. Cap the reserve at a quarter of the window.
+        output_reserve = min(
+            _settings.output_reserve_tokens, max(1, window_tokens // 4)
         )
-        available = settings.ollama_num_ctx - output_reserve
+        est = lambda msgs: self.estimate_tokens(msgs, model)
+        # When the caller supplies the raw session context, prefer the real-token
+        # baseline (real bulk + calibrated delta) over the calibrated estimate;
+        # both are in the same units as the real counts the window is measured in.
+        total = (
+            self.real_baseline_estimate(session_context, context, model)
+            if session_context is not None
+            else est(context)
+        )
+        available = window_tokens - output_reserve
 
         if total > available:
-            existing = self.estimate_context_tokens(context[:-1])
-            new = self.estimate_context_tokens(context[-1:])
+            existing = est(context[:-1])
+            new = est(context[-1:])
             remaining = available - existing
             raise ContextTooLargeError(
                 f"Context too large: new content is ~{new:,} estimated tokens, "
                 f"but only ~{max(0, remaining):,} tokens are available "
-                f"(window {settings.ollama_num_ctx:,}, already used ~{existing:,}, "
+                f"(window {window_tokens:,}, already used ~{existing:,}, "
                 f"output_reserve {output_reserve:,}). "
                 "Split the file into smaller parts or delegate to a subagent."
             )
@@ -846,9 +959,11 @@
         keep_recent: int,
         max_tokens: int | None = None,
         keep_recent_messages: int | None = None,
+        window: int | None = None,
     ) -> ContextCompressed | None:
         """Compresses the session context and persists the changes to the session store."""
         count_before = len(session.context)
+        window_tokens = window or settings.ollama_num_ctx
         result = await self.compress_session(
             context=session.context,
             llm=llm,
@@ -857,6 +972,7 @@
             keep_recent=keep_recent,
             max_tokens=max_tokens,
             keep_recent_messages=keep_recent_messages,
+            window=window_tokens,
         )
         if result is None:
             return None
@@ -871,9 +987,26 @@
         # summaries with is_context=True in the DB — they piled up in the
         # context at every reload (the half-the-context-is-summaries symptom).
         new_context_ids = {id(m) for m in new_context}
+        demoted: list[Message] = []
         for msg in [*session.messages, *session.context]:
             if msg.role != "system" and id(msg) not in new_context_ids:
                 msg.is_context = False
+                demoted.append(msg)
+
+        # Rows that live only in session.context — a summary is is_display=False,
+        # so after a reload it is absent from session.messages — are invisible to
+        # save(), which updates flags for display history only. Their demotion
+        # used to stay in memory: every reload brought the stale summary back
+        # with is_context=True, and one prod session ended up carrying 15 live
+        # summaries, 80% of its context. Persist those flips explicitly.
+        in_messages = {id(m) for m in session.messages}
+        context_only = [
+            m.sequence_number
+            for m in demoted
+            if m.sequence_number >= 0 and id(m) not in in_messages
+        ]
+        if context_only:
+            await session_store.set_context_flag(session_id, context_only, False)
 
         # Add the summary message to session.messages if it's not already there
         summary_msg = next((m for m in new_context if m.is_summary), None)
@@ -893,7 +1026,7 @@
         )
 
         session.context = new_context
-        session.context_token_count = self.estimate_context_tokens(new_context)
+        session.context_token_count = self.estimate_tokens(new_context, model)
         # The context just shrank, so the recorded real-token baseline is stale.
         # Drop it — the next LLM call re-establishes a fresh one.
         self._real_baseline = None
@@ -922,5 +1055,5 @@
             messages_after=len(new_context),
             summary=summary_text,
             context_tokens=session.context_token_count,
-            max_context_tokens=settings.ollama_num_ctx,
+            max_context_tokens=window_tokens,
         )
diff --git a/navi/core/pg_session_store.py b/navi/core/pg_session_store.py
index 72a09dd..a9bfe2f 100644
--- a/navi/core/pg_session_store.py
+++ b/navi/core/pg_session_store.py
@@ -646,6 +646,31 @@
                     removed += 1
         return removed
 
+    async def set_context_flag(
+        self, session_id: str, sequence_numbers: list[int], is_context: bool
+    ) -> int:
+        """Persist a context-flag flip for rows save() cannot see.
+
+        ``save()`` updates flags only for rows in ``session.messages``. A
+        summary is ``is_display=False``: after a reload it lives in
+        ``session.context`` alone, so compression's demotion of a superseded
+        summary never reached the database — every reload resurrected it, and
+        prod sessions accumulated up to 15 live summaries (80% of the context).
+        """
+        if not sequence_numbers:
+            return 0
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            result = await conn.execute(
+                "UPDATE session_messages SET is_context = $3 "
+                "WHERE session_id = $1 AND sequence_number = ANY($2::int[])",
+                session_id,
+                list(sequence_numbers),
+                is_context,
+            )
+        parts = result.split()
+        return int(parts[-1]) if len(parts) >= 2 else 0
+
     async def archive_old_messages(self, session_id: str, keep_seq_threshold: int) -> int:
         """Move messages older than keep_seq_threshold from hot to archive table.
 
diff --git a/navi/core/session.py b/navi/core/session.py
index ce0641b..3e77e6b 100644
--- a/navi/core/session.py
+++ b/navi/core/session.py
@@ -119,6 +119,19 @@
         """
 
     @abstractmethod
+    async def set_context_flag(
+        self, session_id: str, sequence_numbers: list[int], is_context: bool
+    ) -> int:
+        """Flip ``is_context`` on already-persisted rows, whatever their is_display.
+
+        ``save()`` only writes flags for rows in ``session.messages`` (display
+        history). A summary is deliberately ``is_display=False``: after a
+        reload it exists only in ``session.context``, so a demotion made by
+        compression never reached the database and the stale summary came back
+        on the next load. Returns the number of rows changed.
+        """
+
+    @abstractmethod
     async def archive_old_messages(self, session_id: str, keep_seq_threshold: int) -> int: ...
 
     @abstractmethod
@@ -282,6 +295,20 @@
             return None
         return max(candidates, key=lambda s: s.last_active)
 
+    async def set_context_flag(
+        self, session_id: str, sequence_numbers: list[int], is_context: bool
+    ) -> int:
+        session = self._sessions.get(session_id)
+        if session is None:
+            return 0
+        wanted = set(sequence_numbers)
+        changed = 0
+        for msg in [*session.messages, *session.context]:
+            if msg.sequence_number in wanted and msg.is_context != is_context:
+                msg.is_context = is_context
+                changed += 1
+        return changed
+
     async def archive_old_messages(self, session_id: str, keep_seq_threshold: int) -> int:
         # In-memory store: no-op, everything stays in RAM
         return 0
diff --git a/navi/core/token_calibration.py b/navi/core/token_calibration.py
new file mode 100644
index 0000000..23d7245
--- /dev/null
+++ b/navi/core/token_calibration.py
@@ -0,0 +1,143 @@
+"""Map local char counts onto real prompt tokens, per model.
+
+``ContextCompressor.estimate_context_tokens`` (chars // 3) is the only token
+count available *before* an LLM call, and on this workload it is off by a
+factor of ~1.8. Prod sessions measured 101 134 real prompt tokens against a
+~56 400 estimate (tool output, JSON schemas and Russian text are token-dense),
+so every window decision taken on the estimate — the compression target, the
+hard-truncate budgets, the pre-call size check — was taken in a unit roughly
+half the size of the real one. Compression aimed at "65% of the window" and
+landed above it, the next two tool results re-triggered it, and a session could
+run 48 passes in a day.
+
+Every LLM call reports ``prompt_eval_count``, and the char count of the exact
+prompt that produced it is known too. Two observations of different size pin a
+line::
+
+    real_tokens ≈ overhead + chars / chars_per_token
+
+The line predicts the real cost of any hypothetical prompt assembled from the
+same pieces (system text + tool schemas + messages), whichever part of it
+changes — a smaller kept region after compression included. Observations are
+kept per model: tokenizers differ enough that one line does not cover glm and
+gemma.
+
+The fit needs size spread to separate overhead from slope; until it has that,
+``DEFAULT_CHARS_PER_TOKEN`` is used. It is deliberately more conservative than
+chars // 3 (which was measured at ~1.8x below reality) but not as aggressive as
+the worst observed ratio.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+# Fallback until a fitted line exists. The old estimator used 3 chars/token;
+# prod measurements sat at 1.67–1.79. 1.8 is the safe middle: closer to truth
+# on message text without assuming the token-densest content everywhere.
+DEFAULT_CHARS_PER_TOKEN = 1.8
+
+_MAX_OBSERVATIONS = 8
+# The two observations must differ this much in size, or the fitted slope is
+# noise over a narrow band.
+_MIN_SPREAD = 1.25
+# Sanity clamps on the fitted slope (tokens per char). A fit outside this band
+# means the observations are not comparable (e.g. one prompt was truncated by
+# the server) — keep the default instead.
+_MIN_TOKENS_PER_CHAR = 1 / 4.0
+_MAX_TOKENS_PER_CHAR = 1 / 1.0
+
+
+@dataclass(frozen=True)
+class _Observation:
+    chars: int
+    tokens: int
+
+
+class TokenCalibration:
+    """Linear fit of prompt tokens against prompt chars for one model."""
+
+    def __init__(self) -> None:
+        self._observations: list[_Observation] = []
+
+    def observe(self, chars: int, tokens: int | None) -> None:
+        """Record one (prompt chars, real prompt tokens) pair."""
+        if not chars or not tokens or tokens <= 0:
+            return
+        self._observations.append(_Observation(int(chars), int(tokens)))
+        del self._observations[: -_MAX_OBSERVATIONS]
+
+    def _line(self) -> tuple[float, float] | None:
+        """(overhead, tokens_per_char) or None when the observations can't pin one."""
+        n = len(self._observations)
+        if n < 2:
+            return None
+        xs = [o.chars for o in self._observations]
+        if max(xs) < min(xs) * _MIN_SPREAD:
+            return None
+
+        mean_x = sum(xs) / n
+        mean_y = sum(o.tokens for o in self._observations) / n
+        sxx = sum((x - mean_x) ** 2 for x in xs)
+        if sxx <= 0:
+            return None
+        sxy = sum((o.chars - mean_x) * (o.tokens - mean_y) for o in self._observations)
+        slope = sxy / sxx
+        if not (_MIN_TOKENS_PER_CHAR <= slope <= _MAX_TOKENS_PER_CHAR):
+            return None
+        intercept = mean_y - slope * mean_x
+        return max(0.0, intercept), slope
+
+    def tokens_for(self, chars: int) -> int:
+        """Estimated real prompt tokens for a prompt of ``chars`` characters."""
+        line = self._line()
+        if line is None:
+            return int(chars / DEFAULT_CHARS_PER_TOKEN)
+        intercept, slope = line
+        return max(0, round(intercept + chars * slope))
+
+    def delta_for(self, chars: int) -> int:
+        """Price *appended* content with the slope only.
+
+        The fitted intercept is the fixed prompt overhead (system prompt and
+        tool schemas). That overhead is already inside a real baseline count,
+        so charging it again to the delta added on top of that count would
+        double-count it.
+        """
+        line = self._line()
+        if line is None:
+            return int(chars / DEFAULT_CHARS_PER_TOKEN)
+        return max(0, round(chars * line[1]))
+
+    @property
+    def observations(self) -> int:
+        return len(self._observations)
+
+
+_registry: dict[str, TokenCalibration] = {}
+
+
+def _key(model: str | list[str] | None) -> str:
+    if isinstance(model, (list, tuple)):
+        model = model[0] if model else None
+    return (model or "").strip().lower()
+
+
+def calibration_for(model: str | list[str] | None = None) -> TokenCalibration:
+    key = _key(model)
+    calibration = _registry.get(key)
+    if calibration is None:
+        calibration = _registry[key] = TokenCalibration()
+    return calibration
+
+
+def observe_prompt(
+    model: str | list[str] | None, chars: int, tokens: int | None
+) -> None:
+    """Record a real prompt-token count for ``model``."""
+    calibration_for(model).observe(chars, tokens)
+
+
+def reset_calibration() -> None:
+    """Drop every observation — tests, and any operator-driven reset."""
+    _registry.clear()
diff --git a/navi/llm/fallback.py b/navi/llm/fallback.py
index 94e294a..7d4d17c 100644
--- a/navi/llm/fallback.py
+++ b/navi/llm/fallback.py
@@ -134,6 +134,7 @@
         top_k: int | None = None,
         top_p: float | None = None,
         num_thread: int | None = None,
+        num_ctx: int | None = None,
     ) -> LLMResponse:
         models = self._model_list(model)
         last_err: Exception = LLMBackendError("No backends configured")
@@ -157,6 +158,7 @@
                             messages, tools=tools, temperature=temperature,
                             model=m, think=think, max_tokens=max_tokens,
                             top_k=top_k, top_p=top_p, num_thread=num_thread,
+                            num_ctx=num_ctx,
                         )
                     except LLMConnectionError as e:
                         last_err = e
@@ -249,6 +251,7 @@
         top_k: int | None = None,
         top_p: float | None = None,
         num_thread: int | None = None,
+        num_ctx: int | None = None,
     ) -> AsyncGenerator[LLMChunk, None]:
         models = self._model_list(model)
         last_err: Exception = LLMBackendError("No backends configured")
@@ -266,6 +269,7 @@
                         gen = self._get_client(server).stream_complete(
                             messages, tools=tools, temperature=temperature, model=m, think=think,
                             top_k=top_k, top_p=top_p, num_thread=num_thread,
+                            num_ctx=num_ctx,
                         )
                         first = await asyncio.wait_for(
                             gen.__anext__(), timeout=settings.llm_stream_first_chunk_timeout
diff --git a/navi/llm/ollama.py b/navi/llm/ollama.py
index 4ffe6c4..09c921f 100644
--- a/navi/llm/ollama.py
+++ b/navi/llm/ollama.py
@@ -52,8 +52,9 @@
     top_k: int | None = None,
     top_p: float | None = None,
     num_thread: int | None = None,
+    num_ctx: int | None = None,
 ) -> dict:
-    opts: dict = {"temperature": temperature, "num_ctx": settings.ollama_num_ctx}
+    opts: dict = {"temperature": temperature, "num_ctx": num_ctx or settings.ollama_num_ctx}
     if max_tokens is not None:
         opts["num_predict"] = max_tokens
     if top_k is not None:
@@ -121,13 +122,21 @@
         top_k: int | None = None,
         top_p: float | None = None,
         num_thread: int | None = None,
+        num_ctx: int | None = None,
     ) -> LLMResponse:
         resolved = _resolve_model(model, self.model)
         try:
             kwargs: dict = {
                 "model": resolved,
                 "messages": _to_ollama_messages(messages),
-                "options": _base_options(temperature, max_tokens=max_tokens, top_k=top_k, top_p=top_p, num_thread=num_thread),
+                "options": _base_options(
+                    temperature,
+                    max_tokens=max_tokens,
+                    top_k=top_k,
+                    top_p=top_p,
+                    num_thread=num_thread,
+                    num_ctx=num_ctx,
+                ),
                 "stream": False,
                 "think": _resolve_think(think),
             }
@@ -182,13 +191,17 @@
         top_k: int | None = None,
         top_p: float | None = None,
         num_thread: int | None = None,
+        num_ctx: int | None = None,
     ) -> AsyncGenerator[LLMChunk, None]:
         resolved = _resolve_model(model, self.model)
         try:
             kwargs: dict = {
                 "model": resolved,
                 "messages": _to_ollama_messages(messages),
-                "options": _base_options(temperature, top_k=top_k, top_p=top_p, num_thread=num_thread),
+                "options": _base_options(
+                    temperature, top_k=top_k, top_p=top_p, num_thread=num_thread,
+                    num_ctx=num_ctx,
+                ),
                 "stream": True,
                 "think": _resolve_think(think),
             }
diff --git a/navi/profiles/base.py b/navi/profiles/base.py
index d5a4b4e..684a5bb 100644
--- a/navi/profiles/base.py
+++ b/navi/profiles/base.py
@@ -145,6 +145,13 @@
     # compressor uses these values as defaults instead of global settings.
     compression_keep_recent: int | None = None
     compression_max_tokens: int | None = None
+    # Context window (in real prompt tokens) the compressor accounts this
+    # profile's model against. ``None`` = OLLAMA_NUM_CTX. Set it when the
+    # profile's models do not share the global default — a cloud model fed
+    # 115k-token prompts was being compressed against a 65k window, which cost
+    # a summary pass every ~35 messages. The same number is sent to Ollama as
+    # ``num_ctx`` for local models, so it must be what the model can hold.
+    context_window: int | None = None
     # Optional path (relative to profile dir) of a plain-text file containing an
     # additional system prompt appended to the compressor's summary instructions.
     compression_prompt_file: str | None = None
diff --git a/navi/session_files.py b/navi/session_files.py
index 69db724..23ab0e8 100644
--- a/navi/session_files.py
+++ b/navi/session_files.py
@@ -18,6 +18,7 @@
 from PIL import Image
 
 from navi.config import settings
+from navi.content_types import content_type_of
 
 log = structlog.get_logger()
 
@@ -201,10 +202,14 @@
         try:
             stat = entry.stat()
             rel = entry.relative_to(root).as_posix()
+            is_dir = entry.is_dir()
             out.append({
                 "path": rel,
+                # The file list is also what the Files tab previews from, so
+                # each entry carries the type the viewers branch on.
+                "content_type": "unknown" if is_dir else content_type_of(rel, entry),
                 "size": stat.st_size if entry.is_file() else 0,
-                "is_dir": entry.is_dir(),
+                "is_dir": is_dir,
                 "modified_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(),
             })
             if entry.is_dir():
diff --git a/navi/tools/content_publish.py b/navi/tools/content_publish.py
index 560fa6d..e0041a9 100644
--- a/navi/tools/content_publish.py
+++ b/navi/tools/content_publish.py
@@ -10,6 +10,7 @@
 
 from navi.content_store import publish
 from navi.config import settings
+from navi.content_types import CONTENT_TYPES
 from navi.session_files import session_dir
 
 from ._internal.base import Tool, ToolContext, ToolResult, current_session_id
@@ -19,7 +20,8 @@
     name = "content_publish"
     description = (
         "Publish a file from this session's directory for inline viewing in the chat — HTML, "
-        "SVG, PDF, images, video, STL models — as an interactive card.\n\n"
+        "SVG, PDF, images, video, audio, text and source files, STL/glTF models — as an "
+        "interactive card.\n\n"
         "The file MUST already be in the session directory: this tool registers it and does "
         "NOT copy from workspace/. Write, copy or move it there first. Editing it afterwards "
         "updates the card in place, so publish once.\n\n"
@@ -43,8 +45,12 @@
             },
             "content_type": {
                 "type": "string",
-                "enum": ["stl", "html", "svg", "pdf", "image", "video", "unknown"],
-                "description": "Content type for viewer selection. Auto-detected from extension if omitted.",
+                "enum": list(CONTENT_TYPES),
+                "description": (
+                    "Content type for viewer selection. Auto-detected from the file's "
+                    "extension and signature if omitted — pass it only when the detection "
+                    "would pick the wrong viewer."
+                ),
             },
             "source_filename": {
                 "type": "string",
diff --git a/navi/workers/compressor.py b/navi/workers/compressor.py
index 395dfb1..4e773eb 100644
--- a/navi/workers/compressor.py
+++ b/navi/workers/compressor.py
@@ -55,6 +55,9 @@
                 keep_recent=settings.context_keep_recent,
                 max_tokens=settings.context_summary_max_tokens,
                 keep_recent_messages=max(12, settings.context_keep_recent * 2),
+                # The window the gate above just fired against — the profile's
+                # own model window, not the global default.
+                window=ctx.max_context_tokens,
             )
         except Exception:
             # compress_and_save_session already retries and hard-truncates
diff --git a/tests/conftest.py b/tests/conftest.py
index 97d3e06..5b96459 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -26,8 +26,17 @@
     import navi.config as _config_mod
 
     _config_mod.settings = fresh
+
+    # The token calibration is fitted from real prompt counts and lives in a
+    # module-level registry (it has to outlive a ContextCompressor instance —
+    # the compression worker builds a fresh one per pass). Clear it per test so
+    # one test's observations cannot move another's token estimates.
+    from navi.core.token_calibration import reset_calibration
+
+    reset_calibration()
     yield
     _config_mod.settings = fresh
+    reset_calibration()
 
 
 @pytest.fixture
diff --git a/tests/integration/test_websocket.py b/tests/integration/test_websocket.py
index aeaa684..dfeaa97 100644
--- a/tests/integration/test_websocket.py
+++ b/tests/integration/test_websocket.py
@@ -203,3 +203,26 @@
             with pytest.raises(WebSocketDisconnect) as exc_info:
                 ws.receive_json()
         assert exc_info.value.code == 4003
+
+
+class TestLiveness:
+    @pytest.mark.anyio
+    async def test_idle_heartbeat(self, client, make_session, monkeypatch):
+        # A socket nobody writes to still has to carry traffic: that is the only
+        # evidence a sleeping phone, a suspended PWA or a dropped NAT mapping
+        # cannot hide. Twenty seconds is too long for a test, so shrink it.
+        monkeypatch.setattr("navi.api.websocket._HEARTBEAT_INTERVAL", 0.05)
+        session = await make_session("secretary")
+        with client.websocket_connect(f"/ws/sessions/{session.id}") as ws:
+            assert ws.receive_json()["type"] == "session_sync"
+            assert ws.receive_json()["type"] == "heartbeat"
+
+    @pytest.mark.anyio
+    async def test_ping_answered_with_pong(self, client, make_session):
+        session = await make_session("secretary")
+        with client.websocket_connect(f"/ws/sessions/{session.id}") as ws:
+            assert ws.receive_json()["type"] == "session_sync"
+            ws.send_json({"type": "ping", "ts": 1700000000000})
+            # The timestamp comes back untouched so the client can measure the
+            # round trip and time out a socket that never answers.
+            assert ws.receive_json() == {"type": "pong", "ts": 1700000000000}
diff --git a/tests/unit/api/test_session_files.py b/tests/unit/api/test_session_files.py
index 14a71bc..46ac221 100644
--- a/tests/unit/api/test_session_files.py
+++ b/tests/unit/api/test_session_files.py
@@ -128,6 +128,62 @@
     assert "attachment" in response.headers["content-disposition"]
 
 
+async def test_download_audio_is_inline(store):
+    # Served as an attachment an audio file is downloaded instead of played.
+    session_id = await _session_id(store)
+    dest = session_files_mod.ensure_session_dir(session_id) / "voice.mp3"
+    dest.write_bytes(b"ID3\x04\x00" + b"\x00" * 16)
+
+    response = await sessions_mod.download_file(session_id, "voice.mp3", store)
+
+    assert response.media_type == "audio/mpeg"
+    assert "inline" in response.headers["content-disposition"]
+
+
+async def test_download_video_is_inline(store):
+    session_id = await _session_id(store)
+    dest = session_files_mod.ensure_session_dir(session_id) / "clip.mp4"
+    dest.write_bytes(b"\x00\x00\x00\x18ftypisom" + b"\x00" * 16)
+
+    response = await sessions_mod.download_file(session_id, "clip.mp4", store)
+
+    assert response.media_type == "video/mp4"
+    assert "inline" in response.headers["content-disposition"]
+
+
+async def test_download_markdown_is_inline_with_its_own_mime(store):
+    session_id = await _session_id(store)
+    dest = session_files_mod.ensure_session_dir(session_id) / "report.md"
+    dest.write_text("# Report\n")
+
+    response = await sessions_mod.download_file(session_id, "report.md", store)
+
+    assert response.media_type == "text/markdown"
+    assert "inline" in response.headers["content-disposition"]
+
+
+async def test_download_flag_forces_attachment_for_media(store):
+    session_id = await _session_id(store)
+    dest = session_files_mod.ensure_session_dir(session_id) / "voice.mp3"
+    dest.write_bytes(b"ID3\x04\x00" + b"\x00" * 16)
+
+    response = await sessions_mod.download_file(session_id, "voice.mp3", store, download=True)
+
+    assert "attachment" in response.headers["content-disposition"]
+
+
+async def test_file_listing_carries_content_type(store):
+    session_id = await _session_id(store)
+    directory = session_files_mod.ensure_session_dir(session_id)
+    (directory / "voice.mp3").write_bytes(b"ID3\x04\x00" + b"\x00" * 16)
+    (directory / "notes.md").write_text("# Notes\n")
+
+    files = {item["path"]: item for item in session_files_mod.list_session_files(session_id)}
+
+    assert files["voice.mp3"]["content_type"] == "audio"
+    assert files["notes.md"]["content_type"] == "text"
+
+
 async def test_get_session_content_returns_store_items(store, monkeypatch):
     session_id = await _session_id(store)
 
diff --git a/tests/unit/core/test_compressor.py b/tests/unit/core/test_compressor.py
index f09b217..0914bdc 100644
--- a/tests/unit/core/test_compressor.py
+++ b/tests/unit/core/test_compressor.py
@@ -1088,16 +1088,18 @@
     context + a heuristic delta for messages appended since, instead of the
     chars//3 estimate that undercounts code-heavy tool output."""
 
-    def test_no_baseline_falls_back_to_heuristic(self):
+    def test_no_baseline_falls_back_to_the_calibrated_estimate(self):
         c = ContextCompressor()
-        ctx = [Message(role="user", content="hello world")]  # 11 chars -> 3 tokens
-        assert c.real_baseline_estimate(ctx, ctx) == 3
+        # 11 chars -> int(11 / 1.8) = 6 tokens, the calibration default.
+        ctx = [Message(role="user", content="hello world")]
+        assert c.real_baseline_estimate(ctx, ctx) == 6
 
     def test_uses_real_baseline_plus_delta_for_new_messages(self):
         c = ContextCompressor()
         # Simulate: last LLM call saw 4 session messages and reported 5000 real
-        # prompt tokens. Since then 2 new messages (12 chars -> 4 tokens) were
-        # appended. Estimate should be 5000 + 4, NOT the chars//3 of the whole.
+        # prompt tokens. Since then 2 new messages (13 chars -> int(13 / 1.8) = 7
+        # tokens, no fitted line yet) were appended. Estimate should be
+        # 5000 + 7, NOT a character heuristic over the whole context.
         base = [
             Message(role="user", content="do the task"),
             Message(role="assistant", content="step"),
@@ -1110,7 +1112,7 @@
             Message(role="assistant", content="done"),
         ]
         built = [Message(role="system", content="sys")] + session_context
-        assert c.real_baseline_estimate(session_context, built) == 5004
+        assert c.real_baseline_estimate(session_context, built) == 5007
 
     def test_stale_baseline_after_shrink_falls_back_to_heuristic(self):
         """After compression the context shrank below baseline_len -> the
@@ -1118,8 +1120,8 @@
         c = ContextCompressor()
         c.record_real_baseline(10, 5000)  # baseline recorded at len 10
         ctx = [Message(role="user", content="hi")]  # now len 1 < 10
-        # falls back to heuristic on built_ctx
-        assert c.real_baseline_estimate(ctx, ctx) == 0  # 2 chars // 3 = 0
+        # falls back to the calibrated estimate on built_ctx
+        assert c.real_baseline_estimate(ctx, ctx) == 1  # int(2 / 1.8) = 1
 
     def test_record_real_baseline_ignores_none_and_zero(self):
         c = ContextCompressor()
@@ -1156,3 +1158,199 @@
                 c.check_context_size(built, session_context=session_context)
         finally:
             _config.settings = prev_settings
+
+
+class _RecordingStore:
+    """SessionStore that records flag flips the way Postgres needs them.
+
+    ``InMemorySessionStore.save`` keeps the same objects, so an in-memory
+    demotion is visible without any write — exactly the blind spot that hid
+    the stale-summary bug. This store stands in for ``PgSessionStore``: it
+    remembers explicit ``set_context_flag`` calls and nothing else.
+    """
+
+    def __init__(self):
+        self.context_flips: list[tuple[str, tuple[int, ...], bool]] = []
+        self.saved = 0
+
+    async def save(self, session):
+        self.saved += 1
+
+    async def set_context_flag(self, session_id, sequence_numbers, is_context):
+        self.context_flips.append((session_id, tuple(sequence_numbers), is_context))
+        return len(sequence_numbers)
+
+
+class TestCalibratedEstimates:
+    """Window decisions must be taken in real prompt-token units.
+
+    Prod measured 101k real prompt tokens against a 56k chars//3 estimate; a
+    target of "65% of the window" was met in the small unit while the real
+    prompt sat above the window, and compression re-fired every few messages.
+    """
+
+    def test_estimate_tokens_uses_the_fitted_line(self):
+        from navi.core.token_calibration import observe_prompt
+
+        # overhead 10k tokens, 2 chars per token
+        observe_prompt("glm", 100_000, 60_000)
+        observe_prompt("glm", 40_000, 30_000)
+        ctx = [Message(role="user", content="x" * 10_000)]
+        assert ContextCompressor.estimate_tokens(ctx, "glm") == pytest.approx(15_000, rel=0.01)
+        # A different model keeps its own line (none yet -> default).
+        raw = ContextCompressor.estimate_context_tokens(ctx)
+        assert ContextCompressor.estimate_tokens(ctx, "gemma") > raw
+
+    def test_record_real_baseline_feeds_the_calibration(self):
+        c = ContextCompressor()
+        c.record_real_baseline(2, 60_000, chars=100_000, model="glm")
+        c.record_real_baseline(2, 30_000, chars=40_000, model="glm")
+        ctx = [Message(role="user", content="x" * 200_000)]
+        assert c.estimate_tokens(ctx, "glm") == pytest.approx(110_000, rel=0.01)
+
+    def test_would_compress_trusts_the_callers_real_token_count(self):
+        """The gate fired on real prompt tokens; judging it again with the
+        chars//3 heuristic made the two disagree by ~1.8x. Four huge turns and
+        keep_recent=12: the partition finds nothing, so the decision is purely
+        the token gate — which must use the caller's real count."""
+        c = ContextCompressor()
+        ctx = []
+        for i in range(4):
+            ctx.append(Message(role="user", content=str(i)))
+            ctx.append(Message(role="assistant", content="x" * 120_000))
+        # The heuristic alone says "way over the threshold" and would truncate.
+        assert c.would_compress(
+            ctx, keep_recent=12, keep_recent_messages=16,
+            threshold=0.90, max_context_tokens=65_536,
+        ) is True
+        # The real prompt was small: nothing to do.
+        assert c.would_compress(
+            ctx, keep_recent=12, keep_recent_messages=16,
+            threshold=0.90, max_context_tokens=65_536, current_tokens=1_000,
+        ) is False
+
+    def test_appended_messages_are_priced_at_the_slope_not_the_intercept(self):
+        """The pre/mid-turn gates add a delta to a real baseline count. The
+        fitted overhead is already inside that count, so the delta must use
+        the slope alone — or every appended tool result is charged twice."""
+        from navi.core.token_calibration import observe_prompt
+
+        observe_prompt("glm", 100_000, 60_000)   # overhead 10k, 0.5 tok/char
+        observe_prompt("glm", 40_000, 30_000)
+        c = ContextCompressor()
+        c.record_real_baseline(1, 50_000)
+        appended = [Message(role="tool", name="fs", content="x" * 20_000)]
+        # 50_000 real + 20_000 chars at 0.5 tok/char = 60_000, not 70_000.
+        assert c.real_baseline_estimate(
+            [Message(role="user", content="u"), *appended],
+            appended,
+            "glm",
+        ) == pytest.approx(60_000, rel=0.01)
+
+    @pytest.mark.asyncio
+    async def test_target_is_measured_in_real_tokens(self):
+        """After compression the kept region must fit 65% of the window in the
+        same unit the trigger used, not in the half-sized heuristic unit."""
+        backend = FakeLLMBackend(responses=["summary"])
+        c = ContextCompressor()
+        ctx = [Message(role="system", content="sys")]
+        for i in range(20):
+            ctx.append(Message(role="user", content=f"u{i}"))
+            ctx.append(Message(role="assistant", content="x" * 2_000))
+
+        new_context, _ = await c.compress_session(
+            context=ctx, llm=backend, model="m", temperature=0.3,
+            keep_recent=12, window=10_000,
+        )
+        # target = 0.65 * 10_000 = 6_500 calibrated tokens
+        assert c.estimate_tokens(new_context, "m") <= 6_500
+        raw = ContextCompressor.estimate_context_tokens(new_context)
+        # ...which is stricter than the raw heuristic would have allowed.
+        assert raw < 6_500
+
+    def test_window_comes_from_the_caller_not_the_global_default(self):
+        """A cloud model fed 115k-token prompts was compressed against a 65k
+        window. The keep-region target follows the window it is handed."""
+        from navi.core.compressor import _plan_compression
+
+        c = ContextCompressor()
+        ctx = [Message(role="system", content="sys")]
+        for i in range(20):
+            ctx.append(Message(role="user", content=f"u{i}"))
+            ctx.append(Message(role="assistant", content="x" * 2_000))
+
+        def kept(window: int) -> int:
+            planned = _plan_compression(
+                ctx, 12, window=window,
+                estimate=lambda msgs: c.estimate_tokens(msgs, "m"),
+            )
+            assert planned is not None
+            return len(planned[2])
+
+        # 20 turns x ~1.1k calibrated tokens. Target 65% of 65k keeps the whole
+        # 12-turn window; the same target on a 12k window has to drop turns.
+        assert kept(65_536) > kept(12_000)
+
+
+class TestContextOnlyDemotionIsPersisted:
+    """A summary is is_display=False, so after a reload it lives only in
+    session.context and save() never sees it. Its demotion has to be written
+    explicitly — otherwise every reload resurrected it (prod: 15 live
+    summaries, 80% of one session's context)."""
+
+    @pytest.mark.asyncio
+    async def test_stale_context_only_summary_is_flipped_in_the_store(self):
+        from navi.core.session import Session
+
+        store = _RecordingStore()
+        c = ContextCompressor()
+        backend = FakeLLMBackend(responses=["fresh summary"])
+
+        session = Session(profile_id="test")
+        old_summary = Message(
+            role="user", content="stale summary", is_summary=True,
+            is_display=False, is_context=True,
+        )
+        old_summary.sequence_number = 7
+        session.context = [old_summary]
+        for i in range(10):
+            user = Message(role="user", content=f"task {i}")
+            answer = Message(role="assistant", content=f"answer {i}")
+            session.context.extend([user, answer])
+        # Reload shape: display history carries no context-only rows.
+        session.messages = [m for m in session.context if not m.is_summary]
+
+        event = await c.compress_and_save_session(
+            session=session,
+            session_store=store,
+            llm=backend,
+            model="m",
+            temperature=0.3,
+            session_id=session.id,
+            reason="pytest",
+            keep_recent=8,
+            window=65_536,
+        )
+        assert event is not None
+        assert store.context_flips == [(session.id, (7,), False)]
+        assert old_summary.is_context is False
+
+    @pytest.mark.asyncio
+    async def test_no_write_when_nothing_is_demoted(self):
+        from navi.core.session import Session
+
+        store = _RecordingStore()
+        c = ContextCompressor()
+        session = Session(profile_id="test")
+        session.context = [
+            Message(role="user", content="hi"),
+            Message(role="assistant", content="hello"),
+        ]
+        session.messages = list(session.context)
+        event = await c.compress_and_save_session(
+            session=session, session_store=store, llm=FakeLLMBackend(responses=["s"]),
+            model="m", temperature=0.3, session_id=session.id, reason="pytest",
+            keep_recent=8, window=65_536,
+        )
+        assert event is None
+        assert store.context_flips == []
diff --git a/tests/unit/core/test_pg_session_store.py b/tests/unit/core/test_pg_session_store.py
index b99f21a..d84dba0 100644
--- a/tests/unit/core/test_pg_session_store.py
+++ b/tests/unit/core/test_pg_session_store.py
@@ -153,6 +153,38 @@
 
 
 @pytest.mark.asyncio
+async def test_set_context_flag_updates_rows_save_cannot_see():
+    """A summary is is_display=False: after a reload it is absent from
+    session.messages, so save() never writes its is_context flip. This one
+    narrow UPDATE is what stops stale summaries from coming back on load."""
+    conn = FakeConnection()
+    conn.enqueue("UPDATE 2")
+
+    pool = FakePool(conn)
+    store = PgSessionStore(pool)
+    store._initialized = True
+
+    changed = await store.set_context_flag("sid-1", [7, 12], False)
+
+    assert changed == 2
+    query, args = conn.calls[0][1], conn.calls[0][2]
+    assert "UPDATE session_messages SET is_context" in query
+    assert "is_display" not in query  # deliberately not filtered by display flag
+    assert args == ("sid-1", [7, 12], False)
+
+
+@pytest.mark.asyncio
+async def test_set_context_flag_skips_the_db_when_there_is_nothing_to_flip():
+    conn = FakeConnection()
+    pool = FakePool(conn)
+    store = PgSessionStore(pool)
+    store._initialized = True
+
+    assert await store.set_context_flag("sid-1", [], False) == 0
+    assert conn.calls == []
+
+
+@pytest.mark.asyncio
 async def test_archive_old_messages_sends_correct_sql():
     conn = FakeConnection()
     conn.enqueue("INSERT 0 3")  # copy to archive
diff --git a/tests/unit/test_content_store.py b/tests/unit/test_content_store.py
index 8b235e2..5ce80d2 100644
--- a/tests/unit/test_content_store.py
+++ b/tests/unit/test_content_store.py
@@ -149,6 +149,41 @@
         assert len(execute_calls) == 2
         assert all("ON CONFLICT (session_id, filename) DO UPDATE" in call[1] for call in execute_calls)
 
+    async def test_detection_reads_the_file_not_only_its_name(self, monkeypatch, tmp_path):
+        # A PNG saved under a .txt name is an image: the signature wins.
+        conn = FakeConnection()
+        conn.enqueue("INSERT 0 1")
+
+        async def fake_pool():
+            return FakePool(conn)
+
+        monkeypatch.setattr(content_store, "_get_db_pool", fake_pool)
+        sess_dir = tmp_path / "sessions" / "sess-1"
+        sess_dir.mkdir(parents=True)
+        (sess_dir / "picture.txt").write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 16)
+
+        info = await content_store.publish(session_id="sess-1", filename="picture.txt")
+
+        assert info["content_type"] == "image"
+
+    async def test_a_type_without_a_viewer_falls_back_to_detection(self, monkeypatch, tmp_path):
+        conn = FakeConnection()
+        conn.enqueue("INSERT 0 1")
+
+        async def fake_pool():
+            return FakePool(conn)
+
+        monkeypatch.setattr(content_store, "_get_db_pool", fake_pool)
+        sess_dir = tmp_path / "sessions" / "sess-1"
+        sess_dir.mkdir(parents=True)
+        (sess_dir / "voice.mp3").write_bytes(b"ID3\x04\x00" + b"\x00" * 16)
+
+        info = await content_store.publish(
+            session_id="sess-1", filename="voice.mp3", content_type="sound"
+        )
+
+        assert info["content_type"] == "audio"
+
     async def test_missing_file_raises(self, monkeypatch, tmp_path):
         async def fake_pool():
             return FakePool(FakeConnection())
diff --git a/tests/unit/test_content_types.py b/tests/unit/test_content_types.py
new file mode 100644
index 0000000..21391cd
--- /dev/null
+++ b/tests/unit/test_content_types.py
@@ -0,0 +1,169 @@
+"""Unit tests for content-type detection — the one table the viewers trust."""
+
+import pytest
+
+from navi.content_types import (
+    CONTENT_TYPES,
+    content_type_of,
+    detect,
+    is_inline_mime,
+    mime_of,
+    normalize_content_type,
+)
+
+_PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
+_JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 16
+_PDF = b"%PDF-1.7\n" + b"\x00" * 16
+_WAV = b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x00" * 16
+_MP3_WITH_ID3 = b"ID3\x04\x00" + b"\x00" * 16
+_MP3_BARE_FRAME = b"\xff\xfb\x90\x64" + b"\x00" * 16
+_OGG_VORBIS = b"OggS\x00\x02" + b"\x00" * 8 + b"\x01vorbis"
+_OGG_THEORA = b"OggS\x00\x02" + b"\x00" * 8 + b"\x80theora"
+_WEBM_AUDIO = b"\x1a\x45\xdf\xa3\x01\x00\x00\x00webm" + b"\x00" * 16 + b"A_OPUS"
+_WEBM_VIDEO = b"\x1a\x45\xdf\xa3\x01\x00\x00\x00webm" + b"\x00" * 16 + b"V_VP9"
+_GLB = b"glTF\x02\x00\x00\x00" + b"\x00" * 16
+
+
+def _mp4(brand: bytes) -> bytes:
+    return b"\x00\x00\x00\x18ftyp" + brand + b"\x00" * 16
+
+
+def _mpeg_ts() -> bytes:
+    head = bytearray(377)
+    head[0] = head[188] = head[376] = 0x47
+    return bytes(head)
+
+
+def _write(tmp_path, name: str, data: bytes):
+    path = tmp_path / name
+    path.write_bytes(data)
+    return path
+
+
+class TestExtensionTable:
+    @pytest.mark.parametrize(("name", "expected"), [
+        ("chart.svg", "svg"),
+        ("report.pdf", "pdf"),
+        ("photo.png", "image"),
+        ("clip.mp4", "video"),
+        ("track.mp3", "audio"),
+        ("voice.m4a", "audio"),
+        ("stream.opus", "audio"),
+        ("part.stl", "stl"),
+        ("part.glb", "model"),
+        ("part.gltf", "model"),
+        ("notes.md", "text"),
+        ("data.csv", "text"),
+        ("config.json", "text"),
+        ("model.scad", "text"),
+        ("script.py", "text"),
+        ("app.ts", "text"),
+    ])
+    def test_extension_decides_without_reading_the_file(self, name, expected):
+        assert content_type_of(name) == expected
+
+    def test_unknown_extension_without_a_file_is_unknown(self):
+        assert content_type_of("mystery.dat") == "unknown"
+
+    def test_mime_comes_from_the_table_not_the_host(self):
+        # The host's /etc/mime.types calls .ts an MPEG transport stream; the
+        # registry knows a .ts that is not one is TypeScript.
+        assert mime_of("app.ts") == "text/plain"
+        assert mime_of("track.mp3") == "audio/mpeg"
+        assert mime_of("photo.png") == "image/png"
+
+    def test_every_registered_type_is_known_to_the_clients(self):
+        # Guards the contract with the webclient: a new type needs a viewer there.
+        assert "audio" in CONTENT_TYPES
+        assert "text" in CONTENT_TYPES
+        assert "model" in CONTENT_TYPES
+        assert "unknown" in CONTENT_TYPES
+
+
+class TestSignature:
+    @pytest.mark.parametrize(("data", "expected"), [
+        (_PNG, ("image", "image/png")),
+        (_JPEG, ("image", "image/jpeg")),
+        (_PDF, ("pdf", "application/pdf")),
+        (_WAV, ("audio", "audio/wav")),
+        (_MP3_WITH_ID3, ("audio", "audio/mpeg")),
+        (_MP3_BARE_FRAME, ("audio", "audio/mpeg")),
+        (_OGG_VORBIS, ("audio", "audio/ogg")),
+        (_OGG_THEORA, ("video", "video/ogg")),
+        (_WEBM_AUDIO, ("audio", "audio/webm")),
+        (_WEBM_VIDEO, ("video", "video/webm")),
+        (_GLB, ("model", "model/gltf-binary")),
+        (_mp4(b"isom"), ("video", "video/mp4")),
+        (_mp4(b"M4A "), ("audio", "audio/mp4")),
+    ])
+    def test_signature_identifies_the_format(self, tmp_path, data, expected):
+        path = _write(tmp_path, "sample.bin", data)
+
+        assert detect(path.name, path) == expected
+
+    def test_signature_beats_a_wrong_extension(self, tmp_path):
+        # A PNG saved as .txt still reaches the image viewer.
+        path = _write(tmp_path, "picture.txt", _PNG)
+
+        assert detect(path.name, path) == ("image", "image/png")
+
+    def test_audio_saved_without_an_extension_is_still_audio(self, tmp_path):
+        path = _write(tmp_path, "recording", _MP3_WITH_ID3)
+
+        assert detect(path.name, path) == ("audio", "audio/mpeg")
+
+    def test_mpeg_transport_stream_is_not_typescript(self, tmp_path):
+        path = _write(tmp_path, "stream.ts", _mpeg_ts())
+
+        assert detect(path.name, path) == ("video", "video/mp2t")
+
+    def test_typescript_is_not_a_transport_stream(self, tmp_path):
+        path = _write(tmp_path, "app.ts", b"export const answer: number = 42\n")
+
+        assert detect(path.name, path) == ("text", "text/plain")
+
+
+class TestTextAndOpaque:
+    def test_extensionless_text_is_text(self, tmp_path):
+        path = _write(tmp_path, "NOTES", b"line one\nline two\n")
+
+        assert detect(path.name, path) == ("text", "text/plain")
+
+    def test_binary_without_an_extension_stays_unknown(self, tmp_path):
+        path = _write(tmp_path, "blob", b"\x00\x01\x02\x03\xff\xfe")
+
+        assert detect(path.name, path) == ("unknown", "application/octet-stream")
+
+    def test_archive_keeps_its_mime_and_gets_no_viewer(self, tmp_path):
+        path = _write(tmp_path, "bundle.zip", b"PK\x03\x04rest")
+
+        assert detect(path.name, path) == ("unknown", "application/zip")
+
+
+class TestHelpers:
+    @pytest.mark.parametrize(("mime", "expected"), [
+        ("audio/mpeg", True),
+        ("video/mp4", True),
+        ("image/png", True),
+        ("text/html", True),
+        ("text/plain", True),
+        ("application/pdf", True),
+        ("application/json", True),
+        ("application/zip", False),
+        ("application/octet-stream", False),
+        ("model/stl", False),
+    ])
+    def test_inline_mimes(self, mime, expected):
+        assert is_inline_mime(mime) is expected
+
+    @pytest.mark.parametrize(("value", "expected"), [
+        ("audio", "audio"),
+        ("  VIDEO  ", "video"),
+        ("model", "model"),
+        ("unknown", None),
+        ("sound", None),
+        (None, None),
+        ("", None),
+    ])
+    def test_normalize_content_type(self, value, expected):
+        assert normalize_content_type(value) == expected
diff --git a/tests/unit/test_token_calibration.py b/tests/unit/test_token_calibration.py
new file mode 100644
index 0000000..eb6b275
--- /dev/null
+++ b/tests/unit/test_token_calibration.py
@@ -0,0 +1,113 @@
+"""Tests for the per-model prompt-token calibration."""
+
+from __future__ import annotations
+
+import pytest
+
+from navi.core.token_calibration import (
+    DEFAULT_CHARS_PER_TOKEN,
+    TokenCalibration,
+    calibration_for,
+    observe_prompt,
+    reset_calibration,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clean_registry():
+    reset_calibration()
+    yield
+    reset_calibration()
+
+
+class TestDefault:
+    def test_without_observations_uses_the_conservative_default(self):
+        cal = TokenCalibration()
+        # chars // 3 would say 1000; the default is deliberately higher because
+        # chars // 3 was measured ~1.8x below the real prompt size on prod.
+        assert cal.tokens_for(1800) == int(1800 / DEFAULT_CHARS_PER_TOKEN)
+        assert cal.tokens_for(1800) > 1800 // 3
+
+    def test_one_observation_is_not_enough_for_a_line(self):
+        cal = TokenCalibration()
+        cal.observe(100_000, 60_000)
+        assert cal._line() is None
+        assert cal.tokens_for(50_000) == int(50_000 / DEFAULT_CHARS_PER_TOKEN)
+
+
+class TestLine:
+    def test_two_observations_pin_overhead_and_slope(self):
+        cal = TokenCalibration()
+        # overhead 10k tokens, 2 chars per token
+        cal.observe(100_000, 60_000)
+        cal.observe(40_000, 30_000)
+        intercept, slope = cal._line()
+        assert intercept == pytest.approx(10_000, rel=0.01)
+        assert slope == pytest.approx(0.5, rel=0.01)
+        assert cal.tokens_for(200_000) == pytest.approx(110_000, rel=0.01)
+
+    def test_predicts_a_smaller_kept_region_in_real_units(self):
+        """The point of the line: a hypothetical post-compression context has to
+        be priced in real tokens, overhead included. A pure ratio would call
+        10k chars '5.6k tokens' where the real prompt is 15k."""
+        cal = TokenCalibration()
+        cal.observe(100_000, 60_000)
+        cal.observe(40_000, 30_000)
+        assert cal.tokens_for(10_000) == pytest.approx(15_000, rel=0.01)
+
+    def test_delta_pricing_drops_the_fixed_overhead(self):
+        """Appended content is priced at the slope: the 10k overhead is already
+        inside the real baseline count the delta is added to."""
+        cal = TokenCalibration()
+        cal.observe(100_000, 60_000)
+        cal.observe(40_000, 30_000)
+        assert cal.delta_for(10_000) == pytest.approx(5_000, rel=0.01)
+        assert cal.delta_for(10_000) < cal.tokens_for(10_000)
+
+    def test_delta_without_a_line_uses_the_default_ratio(self):
+        cal = TokenCalibration()
+        assert cal.delta_for(1800) == int(1800 / DEFAULT_CHARS_PER_TOKEN)
+
+    def test_observations_with_equal_size_do_not_fit(self):
+        cal = TokenCalibration()
+        cal.observe(50_000, 30_000)
+        cal.observe(51_000, 30_500)  # 2% apart — slope would be noise
+        assert cal._line() is None
+
+    def test_an_implausible_slope_is_rejected(self):
+        cal = TokenCalibration()
+        # 10 tokens per char can only mean one prompt was truncated server-side.
+        cal.observe(10_000, 100_000)
+        cal.observe(20_000, 200_000)
+        assert cal._line() is None
+        assert cal.tokens_for(1800) == int(1800 / DEFAULT_CHARS_PER_TOKEN)
+
+    def test_a_negative_intercept_is_clamped_to_zero(self):
+        cal = TokenCalibration()
+        cal.observe(100_000, 40_000)
+        cal.observe(50_000, 20_500)
+        intercept, _ = cal._line()
+        assert intercept >= 0
+
+    def test_only_the_last_observations_are_kept(self):
+        cal = TokenCalibration()
+        for i in range(20):
+            cal.observe(10_000 + i * 10_000, 5_000 + i * 5_000)
+        assert cal.observations == 8
+
+
+class TestRegistry:
+    def test_calibration_is_per_model(self):
+        observe_prompt("glm", 100_000, 60_000)
+        observe_prompt("glm", 40_000, 30_000)
+        assert calibration_for("glm").observations == 2
+        assert calibration_for("gemma").observations == 0
+
+    def test_model_may_be_the_profile_priority_list(self):
+        observe_prompt(["glm-5.3-flash:cloud", "gemma4:31b-cloud"], 1000, 500)
+        assert calibration_for("GLM-5.3-FLASH:cloud").observations == 1
+
+    def test_reset_drops_every_model(self):
+        observe_prompt("glm", 1000, 500)
+        reset_calibration()
+        assert calibration_for("glm").observations == 0
diff --git a/tests/unit/tools/test_content_publish.py b/tests/unit/tools/test_content_publish.py
index 980d1f7..83ea5df 100644
--- a/tests/unit/tools/test_content_publish.py
+++ b/tests/unit/tools/test_content_publish.py
@@ -20,6 +20,11 @@
         monkeypatch.setattr(content_store_mod, "settings", _test_settings)
         yield ContentPublishTool()
 
+    def test_schema_offers_every_type_the_viewers_have(self):
+        enum = ContentPublishTool.parameters["properties"]["content_type"]["enum"]
+
+        assert {"audio", "text", "model", "stl", "video"} <= set(enum)
+
     async def test_requires_active_session(self):
         result = await ContentPublishTool().execute({"filename": "logo.svg"}, ctx=ToolContext(session_id=None))
 
diff --git a/tests/unit/workers/test_compression_worker.py b/tests/unit/workers/test_compression_worker.py
index ed3ad10..8522d5d 100644
--- a/tests/unit/workers/test_compression_worker.py
+++ b/tests/unit/workers/test_compression_worker.py
@@ -177,4 +177,48 @@
 
     src = inspect.getsource(CompressionWorker.run)
     assert "keep_recent_messages" in src
-    assert "max(12" in src
\ No newline at end of file
+    assert "max(12" in src
+
+class _RecordingStore(InMemorySessionStore):
+    """In-memory store plus a log of explicit context-flag writes.
+
+    In-memory ``save()`` keeps the same objects, so a demotion is visible
+    without any write — the blind spot that hid the stale-summary bug. The
+    flip has to reach the store explicitly for Postgres to keep it.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.context_flips: list[tuple[str, tuple[int, ...], bool]] = []
+
+    async def set_context_flag(self, session_id, sequence_numbers, is_context):
+        self.context_flips.append((session_id, tuple(sequence_numbers), is_context))
+        return await super().set_context_flag(session_id, sequence_numbers, is_context)
+
+
+@pytest.mark.asyncio
+async def test_worker_persists_the_demotion_of_a_context_only_summary():
+    """Reload shape: the stale summary is is_display=False, so save() cannot
+    reach it. Without the explicit flip the row keeps is_context=True and every
+    load brings the whole pile of old summaries back (prod: 15 at once)."""
+    worker = CompressionWorker()
+    backend = FakeLLMBackend(responses=["fresh summary"])
+    store = _RecordingStore()
+    session = Session(profile_id="test")
+    store._sessions[session.id] = session
+
+    old_summary = Message(
+        role="user", content="old summary text", is_summary=True,
+        is_display=False, is_context=True,
+    )
+    old_summary.sequence_number = 41  # already persisted, loaded from the DB
+    context = [old_summary]
+    for i in range(10):
+        context.append(Message(role="user", content=f"task {i}"))
+        context.append(Message(role="assistant", content=f"answer {i}"))
+    session.context = list(context)
+    session.messages = [m for m in context if not m.is_summary]
+
+    result = await worker.run(session, _ctx(session.id, backend, store))
+    assert any(isinstance(ev, ContextCompressed) for ev in result.events)
+    assert store.context_flips == [(session.id, (41,), False)]