diff --git a/CHANGELOG.md b/CHANGELOG.md
index f0fb354..60abf5b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,7 @@
 # Changelog
 
 ## Unreleased (master, since v0.1.38)
+- **feat(rollover): 批量 rollover 压缩 — Prompt Cache 稳定性主杠杆 (closes #241)** — 就地压缩的真实大头是 Prompt Cache 失效（#80 提出、#240 确认）：历史中间改写把压缩点之后的整个后缀踢出缓存前缀，后续每轮全价重算。rollover 模式（**默认开启**，`"rollover": false` 恢复旧行为）让 model-visible history 阶段内 append-only：`compress` 立即校验范围但只记录 pending（原文保持可见）；新 `absorb` 工具把大型 tool output 蒸馏成模型自写的摘要、原文延迟到批量时移除；用量越过 `rollover.threshold`（默认 70%，低于 75% 强制 nudge 带）时一次性批量应用全部 pending（一次缓存失效摊薄整段），追加一次性 `▣ ACP rollover` 报告；`decompress`/`search_context` 结果只追加到尾部。pending 跨重启持久化（`.acp.json` 的 `rolloverPending` + `absorbed` 记录修复：`mergeInitialState` 原本丢弃 `absorbed` 字段）、`acp_status` 显示 pending 行、`/acp-rollover` 命令强制立即应用。kernel 侧零改动（全部 adapter 侧：pending 状态、延迟 apply、隐藏 pending compress 调用的 restore 修复——KEEP_LAST_ORPHANED=2 会把第 3 个 pending 调用从历史中间隐藏、破坏前缀稳定）。测试：新增 tests/rollover.test.ts 5 个（阶段内字节稳定含 ≥3 pending、阈值触发恰好一次改写后重新锚定、absorb 原文可见直到批量、检索结果纯尾部追加、手动 /acp-rollover）；既有 60 处 createAcpExtension 调用显式 `rollover: false` 保留旧行为覆盖；prefix-stab 回归测试同处理
 - **fix(nudge): 移除瞬态 compress 重试提示注入（closes #223，取代 #217）** — compress 失败后每次 LLM 调用重注入的 `compressRetryMessage` 瞬态 user 提示整体移除：对从不重试的模型，该提示无限追加（用户日志 ~400 次/小时、emergency pct 95→127%），即 #223 的"永远追加失败标记"。失败信息本身仍以 toolResult 形式持久留在 session 日志中（模型可见、可自我纠正，随正常压缩流程淘汰）；issue #6 的 nudge 断路器保留：每用户轮 MAX_COMPRESS_ATTEMPTS=3 次失败/no-op 后 emergency nudge 停止重注入（kernel 紧急截断仍机械兜底），UI 提示改为 "nudge paused until the next user message (emergency truncation still active)"。`noteCompressOutcomes` 返回值去掉 `retryFor`
 - **fix(tokens): 图片 token 计入发送视图估算 (closes #200)** — `extractText` 只投影 `type:"text"` 块，图片在 sent-view 估算中计 0 token：含图会话的 nudge/truncation/compress 仲裁系统性偏晚（只等真实 400 后 overflow-selfheal 被动触发），且 density 校准被 phantom gap 污染（provider 真实 usage 含图、估算不含 → 图片轮 dReal/dEst 爆表被 clamp 到 2.5×，纯文本轮又拉回 1.0，density 振荡且仍低估 5-10× → 过早/过晚压缩交替）。现在 `collectImageTokens` 按 `IMAGE_TOKEN_COST=1600`/张计入（仅视觉模型，`model.input` 含 `image`；非视觉模型 pi-ai 静默丢图、计 0），density 校准环自动收敛真实成本（A/B 实测收敛 ~0.98）；出站 payload 字节不变（sha256 一致），前缀缓存不受影响 (#201)
 - **fix(delegate): 并发多 agent 时失败必达，不再静默 (#16)** — async delegate 此前有三条失败路径完全不通知主模型(spawn error、结果持久化 error、`sendUserMessage` 注入丢失)，模型未挂在 `acp_delegate_wait` 上时失败被吞，直到收尾汇总才发现少了结果。现在：所有终止路径 best-effort 注入 `FAILED ⚠️` 通知(带错误摘录，与 sync 路径对齐，明确提示"该任务结果缺失、收尾前决定是否重派")；注入失败的 run 进入未送达集，随**下一个** delegate 通知或任何 delegate 工具结果(`acp_delegate`/`wait`/`cancel`)捎带 Recovery notice 补投；Recovery notice 的 delivered 标记改为 carrier 发送成功后才提交（发送抛错不再永久吞掉其他 run 的结果）；system prompt 补充 FAILED/Recovery 通知说明
diff --git a/CONFIGURATION.md b/CONFIGURATION.md
index 60e9bc3..715062f 100644
--- a/CONFIGURATION.md
+++ b/CONFIGURATION.md
@@ -125,6 +125,14 @@ All keys below are currently **ACTIVE**.
 | `compress.emergencyThresholdPercent` | number \| string | `"95%"` | 🟢 ACTIVE | Context threshold that triggers emergency truncation. |
 | `compress.nudgeGrowthTokens` | number | `50000` | 🟢 ACTIVE | Token growth step for soft compression nudges. |
 
+**Rollover keys**
+
+| Key | Type | Default | Status | Description |
+|-----|------|---------|--------|-------------|
+| `rollover` | boolean \| object | `true` | 🟢 ACTIVE | Batch rollover mode: `compress`/`absorb` are deferred and applied in one batch rewrite when context pressure crosses the threshold. |
+| `rollover.enabled` | boolean | `true` | 🟢 ACTIVE | Enable batch rollover mode. `false` restores the legacy immediate in-place compression. |
+| `rollover.threshold` | number \| string | `"70%"` | 🟢 ACTIVE | Context-usage threshold at which pending compressions/absorbs are applied in one batch. |
+
 **Prompts keys**
 
 | Key | Type | Default | Status | Description |
@@ -340,6 +348,53 @@ On `anthropic` / `claude-sonnet-4-5` the effective thresholds become `maxContext
 
 ---
 
+## Rollover (Prompt Cache Stability)
+
+Batch rollover mode — **on by default** — trades a little temporary context for dramatically fewer prompt-cache invalidations. In legacy mode every `compress` call rewrites the model-visible history in place, evicting the entire suffix after the compression point from the provider's cache prefix; every subsequent round re-pays full price for that suffix. Rollover mode makes history **append-only within a phase**:
+
+1. **Deferred compress** — a `compress` call validates its ranges immediately (bad ranges still fail now, with errors) but only *records* them as pending. The range stays visible until the batch applies.
+2. **`absorb` tool** — distills a large tool result into a compact summary you write; the original output is marked pending drop and stays visible until the batch applies (the summary is the durable record).
+3. **Batch rollover** — when context usage crosses `rollover.threshold` (default 70%), all pending compressions and absorbs are applied in **one** rewrite: one cache invalidation, amortized over the whole phase. A one-shot `▣ ACP rollover | ...` report is appended to that round.
+4. **Retrieval appends to the tail** — `decompress` / `search_context` results are tool results at the end of the history; the prefix is never touched.
+
+Pending work is visible in `acp_status` (`Rollover: N pending ...` line) and survives restarts (persisted alongside the ACP state). To force the batch early, run `/acp-rollover`.
+
+### Trade-off
+
+Pending content occupies context until the rollover fires — that is the price of a stable cache prefix. The default threshold (70%) sits **below** the forced-nudge band (`compress.maxContextLimit`, 75%) so the rollover always applies before the nudge escalation, and the reclaimed tokens (typically tens of percent of the window) drop usage well back below the band in one step.
+
+### `rollover`
+
+- **Type:** `boolean | object`
+- **Default:** `true`
+- **Status:** 🟢 ACTIVE
+- **Description:** Enable batch rollover mode. `false` (or `{"enabled": false}`) restores the legacy behavior where every `compress` call rewrites history in place immediately.
+
+### `rollover.enabled`
+
+- **Type:** `boolean`
+- **Default:** `true`
+- **Status:** 🟢 ACTIVE
+- **Description:** Same as the `rollover` shorthand. `false` disables deferred compression: `compress` applies immediately, the `absorb` tool is not registered, and the rollover system-prompt section is omitted.
+
+### `rollover.threshold`
+
+- **Type:** `number | string`
+- **Default:** `0.70` (or `"70%"`)
+- **Status:** 🟢 ACTIVE
+- **Description:** Context-usage threshold at which pending compressions/absorbs are applied in one batch. Accepts a ratio (`0.70`) or a percent string (`"70%"`). Keep it **below** `compress.maxContextLimit` so the rollover fires before forced nudges start. A higher value keeps the prefix stable longer at the cost of carrying more pending context; a lower value reclaims sooner.
+
+```json
+{
+  "rollover": {
+    "enabled": true,
+    "threshold": "70%"
+  }
+}
+```
+
+---
+
 ## Prompts Customization
 
 The `prompts` object overrides acp-kernel's **load-bearing** compression prompt rules — the verbatim instructions the model receives about *how* to write summaries (keep full file paths, function signatures, decisions and rationale; drop verbose logs, etc.). These four fields are embedded into the system prompt and the compression nudge text:
diff --git a/CONFIGURATION.zh-CN.md b/CONFIGURATION.zh-CN.md
index 02d4783..77c1fe9 100644
--- a/CONFIGURATION.zh-CN.md
+++ b/CONFIGURATION.zh-CN.md
@@ -125,6 +125,14 @@
 | `compress.emergencyThresholdPercent` | number \| string | `"95%"` | 🟢 ACTIVE | 触发紧急截断的上下文阈值。 |
 | `compress.nudgeGrowthTokens` | number | `50000` | 🟢 ACTIVE | 软压缩 nudge 的 token 增长步长。 |
 
+**Rollover 键**
+
+| 键 | 类型 | 默认值 | 状态 | 说明 |
+|----|------|--------|------|------|
+| `rollover` | boolean \| object | `true` | 🟢 ACTIVE | 批量 rollover 模式:`compress`/`absorb` 延迟生效,在上下文压力越过阈值时一次性批量应用。 |
+| `rollover.enabled` | boolean | `true` | 🟢 ACTIVE | 启用批量 rollover 模式。`false` 恢复旧的就地立即压缩行为。 |
+| `rollover.threshold` | number \| string | `"70%"` | 🟢 ACTIVE | pending 压缩/吸收一次性批量应用的上下文用量阈值。 |
+
 **prompts 键**
 
 | 键 | 类型 | 默认值 | 状态 | 说明 |
@@ -340,6 +348,53 @@ provider 的 key 是 **Pi provider 名**(如 `"anthropic"`、`"openai"`、`"zhip
 
 ---
 
+## Rollover（Prompt Cache 稳定性）
+
+批量 rollover 模式 —— **默认开启** —— 用一点临时 context 换取大幅减少的 prompt-cache 失效次数。旧模式下每次 `compress` 都在历史中间就地改写,把压缩点之后的整个后缀踢出 provider 缓存前缀;之后每一轮都按全价重算这段 input。rollover 模式让 model-visible history 在阶段内 **append-only**:
+
+1. **延迟压缩** —— `compress` 调用立即校验范围(坏范围仍当场报错),但只把范围*记录*为 pending。原文保持可见,直到批量应用。
+2. **`absorb` 工具** —— 把大型工具输出蒸馏成你写的紧凑摘要;原文标记为 pending drop,保持可见直到批量应用(摘要是持久记录)。
+3. **批量 rollover** —— 当上下文用量越过 `rollover.threshold`(默认 70%)时,所有 pending 压缩与吸收**一次性**应用:一次缓存失效,摊薄到整个阶段。该轮追加一条一次性 `▣ ACP rollover | ...` 报告。
+4. **检索结果追加到尾部** —— `decompress` / `search_context` 的结果是历史末尾的 tool result,前缀永不被触碰。
+
+pending 工作显示在 `acp_status`(`Rollover: N pending ...` 行)中,并跨重启持久化(与 ACP state 一起保存)。想提前强制批量,运行 `/acp-rollover`。
+
+### 权衡
+
+pending 内容在 rollover 触发前一直占用 context —— 这是换取稳定缓存前缀的代价。默认阈值(70%)位于强制 nudge 带(`compress.maxContextLimit`,75%)**之下**,保证 rollover 总在 nudge 升级之前应用;回收的 token(通常是窗口的百分之几十)一步就把用量拉回带内。
+
+### `rollover`
+
+- **类型:** `boolean | object`
+- **默认值:** `true`
+- **状态:** 🟢 ACTIVE
+- **说明:** 启用批量 rollover 模式。`false`(或 `{"enabled": false}`)恢复旧行为:每次 `compress` 立即就地改写历史。
+
+### `rollover.enabled`
+
+- **类型:** `boolean`
+- **默认值:** `true`
+- **状态:** 🟢 ACTIVE
+- **说明:** 与 `rollover` 简写相同。`false` 禁用延迟压缩:`compress` 立即生效,不注册 `absorb` 工具,系统提示词中也不含 rollover 段落。
+
+### `rollover.threshold`
+
+- **类型:** `number | string`
+- **默认值:** `0.70`(或 `"70%"`)
+- **状态:** 🟢 ACTIVE
+- **说明:** pending 压缩/吸收一次性批量应用的上下文用量阈值。接受比例(`0.70`)或百分号字符串(`"70%"`)。保持**低于** `compress.maxContextLimit`,让 rollover 在强制 nudge 开始之前触发。值越高,前缀稳定时间越长,但携带的 pending context 越多;值越低,回收越早。
+
+```json
+{
+  "rollover": {
+    "enabled": true,
+    "threshold": "70%"
+  }
+}
+```
+
+---
+
 ## 提示词自定义
 
 `prompts` 对象覆盖 acp-kernel 的**承重**压缩提示词规则——即模型收到的关于*如何*写摘要的逐字指令（保留完整文件路径、函数签名、决策与理由；丢弃冗长日志等）。这四个字段被嵌入系统提示词和压缩 nudge 文本：
diff --git a/README.md b/README.md
index 0344dff..78c921a 100644
--- a/README.md
+++ b/README.md
@@ -62,6 +62,19 @@ Each message gets an invisible `<acp>` ref tag (`m00001`, `m00002`, ...) visible
 
 Pi's built-in auto-compaction is cancelled — billion-context is the sole context manager.
 
+## Batch rollover — Prompt Cache stability
+
+The dominant cost of in-place compression is not the summary's output tokens — it is the **prompt-cache invalidation**: rewriting history mid-stream evicts the entire suffix after the compression point from the provider's cache prefix, and every later round re-pays full price for it.
+
+Batch rollover mode (on by default) makes the model-visible history **append-only within a phase**:
+
+- `compress` validates its ranges immediately but only **records** them as pending — the range stays visible.
+- `absorb` distills a large tool result into a summary you write; the original stays visible until the batch applies.
+- When context usage crosses the rollover threshold (default **70%**, below the 75% forced-nudge band), all pending work is applied in **one** rewrite — one cache invalidation, amortized over the whole phase — and a one-shot `▣ ACP rollover | ...` report is appended.
+- `decompress` / `search_context` results land at the tail of the history; the prefix is never touched.
+
+Pending work shows up in `acp_status` and survives restarts; `/acp-rollover` forces the batch early. Set `"rollover": false` in `acp.json` to restore the legacy immediate-compression behavior. See [CONFIGURATION.md](./CONFIGURATION.md#rollover-prompt-cache-stability) for the trade-off and thresholds.
+
 ## Plugin compatibility & ordering
 
 billion-context takes over context management by intercepting Pi's `context` event. **Pi has no plugin priority mechanism** — when multiple extensions register handlers for the same event, they run in a fixed sequence (load order), with no `priority`/`weight` field and no way for the user to control the order. The `context` event specifically is a *pipeline*: every handler receives the previous handler's output, there is no short-circuit, and the **last** handler has the final say over what reaches the model.
@@ -76,7 +89,8 @@ This has two practical implications:
 
 | Tool | What it does |
 |------|-------------|
-| `compress` | Replace a contiguous message range with a detailed summary |
+| `compress` | Replace a contiguous message range with a detailed summary (deferred to the next rollover in batch mode) |
+| `absorb` | Distill a large tool result into a compact summary; the original is dropped at the next rollover |
 | `decompress` | Restore a previously compressed block's content |
 | `search_context` | Search compressed block summaries (and visible messages) by keyword |
 | `acp_status` | Show context usage, compressed blocks, compressible ranges |
@@ -134,6 +148,8 @@ Blocks: 3 active (3.7K summary, 15.2K original compressed)
   b3 (T2)  3.3K→1.0K  age=1m  "Architecture review"
 ```
 
+In batch rollover mode the status also shows pending work (`Rollover: N pending compression(s) + M absorb(s) — ~X tokens pending (threshold 70%, current Y%)`), and `/acp-rollover` applies the pending batch immediately instead of waiting for the threshold.
+
 ## `/acp-subagents` command
 
 **Optional, one-time setup — only if you also use [pi-subagents](https://github.com/nicobailon/pi-subagents).**
diff --git a/README.zh-CN.md b/README.zh-CN.md
index 04f1a81..72bdf9e 100644
--- a/README.zh-CN.md
+++ b/README.zh-CN.md
@@ -61,6 +61,19 @@ assign refs → sync blocks → prune → filter → hide calls → recommend 
 
 Pi 内置的自动压缩会被取消 —— billion-context 是唯一的上下文管理者。
 
+## 批量 rollover —— Prompt Cache 稳定性
+
+就地压缩的主要成本不是摘要的 output tokens,而是 **Prompt Cache 失效**:在历史中间改写会把压缩点之后的整个后缀踢出 provider 的缓存前缀,之后每一轮都要按全价重算这段 input。
+
+批量 rollover 模式(默认开启)让 model-visible history 在阶段内 **append-only**:
+
+- `compress` 立即校验范围(坏范围仍然当场报错),但只把范围**记录**为 pending —— 原文保持可见。
+- `absorb` 把大型工具输出蒸馏成你写的摘要;原文保持可见,直到批量生效。
+- 当上下文用量越过 rollover 阈值(默认 **70%**,低于 75% 强制 nudge 带)时,所有 pending 工作**一次性**应用 —— 一次缓存失效,摊薄到整个阶段 —— 并追加一条一次性的 `▣ ACP rollover | ...` 报告。
+- `decompress` / `search_context` 的结果落在历史尾部,前缀永不被触碰。
+
+pending 工作会显示在 `acp_status` 中,并跨重启持久化;`/acp-rollover` 可立即强制批量生效。在 `acp.json` 中设置 `"rollover": false` 可恢复旧的就地立即压缩行为。权衡与阈值详见 [CONFIGURATION.zh-CN.md](./CONFIGURATION.zh-CN.md#rollover-prompt-cache-稳定性)。
+
 ## 插件兼容性与排序
 
 billion-context 通过拦截 Pi 的 `context` 事件接管上下文管理。**Pi 没有插件优先级机制** —— 当多个扩展为同一个事件注册 handler 时,它们按固定顺序(加载顺序)执行,没有 `priority`/`weight` 字段,用户也无法控制顺序。`context` 事件尤其是一个*管线*:每个 handler 都接收上一个 handler 的输出,没有短路,**最后一个** handler 对发给模型的内容拥有最终决定权。
@@ -75,7 +88,8 @@ billion-context 通过拦截 Pi 的 `context` 事件接管上下文管理。**Pi
 
 | 工具 | 作用 |
 |------|------|
-| `compress` | 用详细摘要替换连续的消息范围 |
+| `compress` | 用详细摘要替换连续的消息范围(批量模式下延迟到下一次 rollover 生效) |
+| `absorb` | 把大型工具输出蒸馏成你写的紧凑摘要;原文在下一次 rollover 时移除 |
 | `decompress` | 恢复之前压缩的块内容 |
 | `search_context` | 按关键词搜索已压缩块摘要(及可见消息) |
 | `acp_status` | 显示上下文用量、已压缩块、可压缩范围 |
@@ -132,6 +146,8 @@ Blocks: 3 active (3.7K summary, 15.2K original compressed)
   b3 (T2)  3.3K→1.0K  age=1m  "Architecture review"
 ```
 
+批量 rollover 模式下,状态面板还会显示 pending 工作(`Rollover: N pending compression(s) + M absorb(s) — ~X tokens pending (threshold 70%, current Y%)`),`/acp-rollover` 可立即应用 pending 批量,而不必等待阈值。
+
 ## `/acp-subagents` 命令
 
 **可选、一次性设置——仅当你同时使用 [pi-subagents](https://github.com/nicobailon/pi-subagents) 时需要。**
diff --git a/src/absorb-tool.ts b/src/absorb-tool.ts
new file mode 100644
index 0000000..a587588
--- /dev/null
+++ b/src/absorb-tool.ts
@@ -0,0 +1,84 @@
+import { Type, type Static } from "typebox";
+import type {
+  AgentToolResult,
+  ExtensionContext,
+  ToolDefinition,
+} from "@earendil-works/pi-coding-agent";
+import { applyAbsorb, defaultCountTokens } from "acp-kernel";
+import type { AcpRuntime } from "./runtime.js";
+import { logThrow } from "./log.js";
+import { estimateTokens, collectCoveredMessageIds, calibrateTokens } from "./tokens.js";
+import { getSystemPromptText } from "./compat.js";
+import { emptyPending } from "./rollover.js";
+
+const AbsorbParams = Type.Object({
+  ref: Type.String({ description: 'Message ref of the tool result to distill, e.g. "m00012" (from its acp tag).' }),
+  summary: Type.String({ description: "Distilled essentials of the tool result: outcome, key values, paths:lines, errors, decisions. Dense and self-contained." }),
+});
+
+type AbsorbArgs = Static<typeof AbsorbParams>;
+
+export function makeAbsorbTool(runtime: AcpRuntime): ToolDefinition<typeof AbsorbParams> {
+  return {
+    name: "absorb",
+    label: "Absorb",
+    description:
+      "Distill a large tool result into a compact summary you write. The original output is marked pending drop and removed from context at the next rollover (when usage crosses the rollover threshold, or via /acp rollover) — until then it stays visible, keeping the prompt-cache prefix stable. Use for large tool outputs you have already used.",
+    promptSnippet: 'absorb({ ref: "m00012", summary: "..." })',
+    promptGuidelines: [
+      "Only tool results are absorbable (not user or assistant messages).",
+      "The original stays visible until the next rollover — absorb when you are done with the output.",
+      "Write dense summaries: keep paths, exact values, errors, and decisions.",
+    ],
+    parameters: AbsorbParams,
+    async execute(toolCallId, params, _signal, _onUpdate, ctx): Promise<AgentToolResult<unknown>> {
+      let result: string;
+      try {
+        result = await handleAbsorb(params as AbsorbArgs, runtime, ctx, toolCallId);
+      } catch (e) {
+        logThrow("absorb", e, { sid: ctx.sessionManager.getSessionId() });
+        throw e;
+      }
+      return { details: undefined, content: [{ type: "text", text: result }] };
+    },
+  };
+}
+
+async function handleAbsorb(args: AbsorbArgs, runtime: AcpRuntime, ctx: ExtensionContext, toolCallId?: string): Promise<string> {
+  const { state: initialState, coreMessages } = await runtime.stateFor(ctx);
+  const config = runtime.configFor(ctx);
+  const modelId = (ctx.model as { id?: string } | undefined)?.id ?? "default";
+  const systemPromptText = getSystemPromptText(ctx);
+  const systemPromptTokens = systemPromptText ? defaultCountTokens(systemPromptText) : 0;
+  const sentTokens = estimateTokens(coreMessages, collectCoveredMessageIds(initialState)) + systemPromptTokens;
+  const turn = runtime.core.processTurn({
+    messages: coreMessages,
+    state: initialState,
+    config,
+    tokenCount: calibrateTokens(sentTokens, runtime.density.densityFor(modelId)),
+  });
+  const ref = args.ref.trim();
+  const pending = runtime.getRolloverPending(ctx);
+  const rawId = turn.state.messageRefs.byRef[ref];
+  if (rawId && pending?.absorbs.some((a) => a.resultMessageId === rawId)) {
+    return `already recorded for rollover (${ref}) — no change.`;
+  }
+  const outcome = applyAbsorb({
+    ref,
+    summary: args.summary,
+    absorbCallId: toolCallId,
+    messages: turn.messages,
+    state: turn.state,
+    config,
+    countTokens: defaultCountTokens,
+  });
+  if (!outcome.ok) throw new Error(outcome.resultText);
+  const prevCount = turn.state.absorbed?.length ?? 0;
+  const newCount = outcome.state.absorbed?.length ?? 0;
+  if (newCount <= prevCount) return outcome.resultText;
+  const newRecord = outcome.state.absorbed![newCount - 1]!;
+  const existing = pending ?? emptyPending();
+  runtime.setRolloverPending(ctx, { compressions: existing.compressions, absorbs: [...existing.absorbs, newRecord] });
+  await runtime.save(turn.state, ctx);
+  return outcome.resultText.replace("is now hidden", "is recorded for the next rollover and stays visible until then");
+}
diff --git a/src/commands.ts b/src/commands.ts
index 3b00445..e8c12cf 100644
--- a/src/commands.ts
+++ b/src/commands.ts
@@ -6,6 +6,8 @@ import { collectCoveredMessageIds, estimateTokens, calibrateTokens, collectImage
 import { buildStatusPanel } from "acp-kernel/panel";
 import { getDelegateUsage } from "./delegate-tool.js";
 import { ensureSubagentAcpTools } from "./setup-subagent-tools.js";
+import { resolveRollover } from "./config.js";
+import { pendingHasWork, runRollover, rolloverReportText } from "./rollover.js";
 
 declare const CURRENT_VERSION: string;
 
@@ -41,6 +43,38 @@ export function makeCommands(runtime: AcpRuntime): Array<{ name: string; options
         handler: async (_args, ctx) => ctx.ui.notify(await statusReport(runtime, ctx)),
       },
     },
+    {
+      name: "acp-rollover",
+      options: {
+        description: "Apply all pending rollover work now (batch-compress + absorb pending ranges in one rewrite). Usage: /acp-rollover",
+        handler: async (_args, ctx) => {
+          const rollover = resolveRollover(runtime.adapter);
+          if (!rollover.enabled) {
+            ctx.ui.notify("Rollover mode is disabled (rollover: false). Nothing to do.");
+            return;
+          }
+          if (!pendingHasWork(runtime.getRolloverPending(ctx))) {
+            ctx.ui.notify("No pending rollover work (no recorded compressions or absorbs).");
+            return;
+          }
+          const release = await runtime.acquireLock(ctx.sessionManager.getSessionId());
+          try {
+            const { state, coreMessages, entries } = await runtime.stateFor(ctx);
+            const config = runtime.configFor(ctx);
+            const modelId = (ctx.model as { id?: string } | undefined)?.id ?? "default";
+            const systemPromptText = getSystemPromptText(ctx);
+            const systemPromptTokens = systemPromptText ? defaultCountTokens(systemPromptText) : 0;
+            const imageTokens = collectImageTokens(entries, modelSupportsImages(ctx.model));
+            const sentTokens = estimateTokens(coreMessages, collectCoveredMessageIds(state), imageTokens) + systemPromptTokens;
+            const turn = runtime.core.processTurn({ messages: coreMessages, state, config, tokenCount: calibrateTokens(sentTokens, runtime.density.densityFor(modelId)) });
+            const result = await runRollover({ runtime, ctx, config, coreMessages, turn, modelId, imageTokens, systemPromptTokens });
+            ctx.ui.notify(result ? rolloverReportText(result) : "No pending rollover work.");
+          } finally {
+            release();
+          }
+        },
+      },
+    },
     {
       name: "acp-decompress",
       options: {
diff --git a/src/compress-tool.ts b/src/compress-tool.ts
index 9f0b579..8ffe2de 100644
--- a/src/compress-tool.ts
+++ b/src/compress-tool.ts
@@ -9,6 +9,8 @@ import { debug, logError, logInfo, logThrow, logWarn } from "./log.js";
 import { estimateTokens, collectCoveredMessageIds, calibrateTokens, collectImageTokens, modelSupportsImages } from "./tokens.js";
 import { defaultCountTokens, parseCompressArgs, type CompressionBlock, type CompressParseDiagnostics } from "acp-kernel";
 import { getSystemPromptText } from "./compat.js";
+import { resolveRollover } from "./config.js";
+import { emptyPending, pendingOverlaps, rangeTokenEstimate } from "./rollover.js";
 
 function formatK(n: number): string {
   return n >= 1000 ? `${(n / 1000).toFixed(1)}K` : String(n);
@@ -105,11 +107,12 @@ function compressPanelBlocks(text: string): number {
 }
 
 /** Success = completed run that created >= 1 block (partial range errors
- *  still count: progress was made). A 0-block panel must NOT be success —
- *  it would reset the retry counter while the emergency nudge re-fires,
- *  looping no-op compressions (issue #6). */
+ *  still count: progress was made), OR a rollover-mode panel that recorded
+ *  ranges as pending (work was accepted — the retry counter resets). A
+ *  0-block panel must NOT be success — it would reset the retry counter
+ *  while the emergency nudge re-fires, looping no-op compressions (issue #6). */
 export function isCompressSuccessText(text: string): boolean {
-  return compressPanelBlocks(text) > 0;
+  return compressPanelBlocks(text) > 0 || text.includes("recorded for next rollover");
 }
 
 /** No-op = completed run that compressed nothing (0-block panel: every
@@ -183,6 +186,62 @@ async function handleCompress(args: CompressArgs, runtime: AcpRuntime, ctx: Exte
     beforeMsgCount: messages.length,
     beforeTokens,
   });
+  // Batch rollover mode (#241): validate the ranges against a scratch state
+  // (the kernel clones internally — `state` is untouched), then record them
+  // as pending. Nothing in the model-visible history changes until the next
+  // rollover, so the prompt-cache prefix stays stable inside the phase.
+  const rollover = resolveRollover(runtime.adapter);
+  if (rollover.enabled) {
+    const scratch = runtime.core.applyCompression({
+      ranges: ranges.map((r) => ({ startRef: r.startId, endRef: r.endId, summary: r.summary, topic: r.topic ?? topLevelTopic, summaryMaxChars, compressCallId: toolCallId })),
+      messages,
+      state,
+      config,
+    });
+    const rewriteSpans = scratch.result.blocksCreated > 0
+      ? tier3OnlyRewrite(scratch.state.blocks.slice(-scratch.result.blocksCreated), scratch.state.blocks)
+      : null;
+    if (rewriteSpans) {
+      throw new Error(
+        `Range ${rewriteSpans.join(", ")} only re-condenses terminal tier-3 block(s) — T3 is the highest tier, so rewriting it reclaims nothing and can repeat forever (dog/billion-context-pi#3). Nothing was compressed. ` +
+          `Use search_context or decompress to retrieve details, or pick a range containing uncompressed messages (acp_status lists compressible ranges).`,
+      );
+    }
+    if (scratch.result.errors.length > 0) {
+      throw new Error("Errors: " + scratch.result.errors.join("; "));
+    }
+    const pending = runtime.getRolloverPending(ctx) ?? emptyPending();
+    for (const r of ranges) {
+      if (pendingOverlaps(pending, r.startId, r.endId)) {
+        throw new Error(`Range ${r.startId}..${r.endId} overlaps a range already recorded for the next rollover — it will be compressed once, in batch.`);
+      }
+    }
+    const newOnes = ranges.map((r) => ({
+      startRef: r.startId,
+      endRef: r.endId,
+      summary: r.summary,
+      topic: r.topic ?? topLevelTopic,
+      summaryMaxChars,
+      callId: toolCallId ?? "",
+      createdAt: Date.now(),
+      estTokens: rangeTokenEstimate(messages, state, r.startId, r.endId),
+    }));
+    runtime.setRolloverPending(ctx, { compressions: [...pending.compressions, ...newOnes], absorbs: pending.absorbs });
+    await runtime.save(state, ctx);
+    const pendingTokens =
+      pending.compressions.reduce((s, c) => s + c.estTokens, 0) +
+      newOnes.reduce((s, c) => s + c.estTokens, 0) +
+      pending.absorbs.reduce((s, a) => s + a.tokensReclaimed, 0);
+    logInfo("compress", {
+      sid: ctx.sessionManager.getSessionId(),
+      event: "recorded-pending",
+      ranges: ranges.length,
+      pendingCompressions: pending.compressions.length + newOnes.length,
+      pendingTokens,
+    });
+    return `▣ ACP | ${ranges.length} range${ranges.length > 1 ? "s" : ""} recorded for next rollover — ~${formatK(pendingTokens)} tokens pending (applied when usage crosses ${Math.round(rollover.threshold * 100)}%)`;
+  }
+
   const applied = runtime.core.applyCompression({
     ranges: ranges.map((r) => ({ startRef: r.startId, endRef: r.endId, summary: r.summary, topic: r.topic ?? topLevelTopic, summaryMaxChars, compressCallId: toolCallId })),
     messages,
diff --git a/src/config.ts b/src/config.ts
index e19e411..2400c28 100644
--- a/src/config.ts
+++ b/src/config.ts
@@ -34,6 +34,21 @@ export interface CompressSettings {
   nudgeGrowthTokens?: number;
 }
 
+/** Batch rollover tuning (Prompt Cache stability, #241). Deferred compress /
+ *  absorb work is applied in one batch when the calibrated sent-view usage
+ *  crosses `threshold` (or via `/acp rollover`), so the model-visible history
+ *  stays append-only within a phase and the cache prefix is rewritten once
+ *  per rollover instead of once per compression. */
+export interface RolloverConfig {
+  /** Enable batch rollover mode. Default: true. Set `false` (or
+   *  `rollover: false`) to restore the legacy immediate-compression behavior. */
+  enabled?: boolean;
+  /** Context usage percentage that triggers a rollover. Accepts a ratio (0.7)
+   *  or percent string ("70%"). Default: 0.70 — below the 0.75 forced-nudge
+   *  band so pending work is reclaimed before nudges start. */
+  threshold?: number | string;
+}
+
 /** Per-provider compression overrides. Carries the same tuning fields as the
  *  global level, plus an optional per-model map keyed by model id. */
 export interface ProviderCompress extends CompressSettings {
@@ -111,6 +126,10 @@ export interface AdapterConfig {
    *  replacing the kernel's tuned compression rules may reduce summary quality
    *  (lost paths/signatures/decisions → worse retrieval). */
   acknowledgePromptsRisk?: boolean;
+  /** Batch rollover mode (Prompt Cache stability, #241). Accepts a boolean
+   *  shorthand (`false` disables) or a RolloverConfig object. Default: enabled
+   *  at a 0.70 usage threshold. */
+  rollover?: boolean | RolloverConfig;
   coreOverrides?: Partial<Config>;
 }
 
@@ -133,6 +152,22 @@ export function resolveDelegate(adapter: AdapterConfig): { enabled: boolean; dis
   };
 }
 
+/** Resolve rollover config from the adapter, handling the boolean shorthand.
+ *  Default: enabled at a 0.70 usage threshold. */
+export function resolveRollover(adapter: AdapterConfig): { enabled: boolean; threshold: number } {
+  const r = adapter.rollover;
+  if (typeof r === "object" && r !== null) {
+    return {
+      enabled: r.enabled !== false,
+      threshold: r.threshold !== undefined ? parsePercent(r.threshold) : 0.7,
+    };
+  }
+  return {
+    enabled: r !== false,
+    threshold: 0.7,
+  };
+}
+
 /** Per-field deepest-wins merge of the three compression levels (global →
  *  provider → model). An undefined field at a deeper level does NOT clear a
  *  value set at a shallower level — only a defined value overrides. */
diff --git a/src/index.ts b/src/index.ts
index 6464fdc..8d4ad2e 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -6,20 +6,22 @@ import type {
 } from "@earendil-works/pi-coding-agent";
 import type { CoreMessage, NudgeDecision, CompressionBlock, Prompts } from "acp-kernel";
 import { renderNudgeText, resolvePrompts, defaultPrompts, viableRanges } from "acp-kernel";
-import { type AdapterConfig, resolveDelegate } from "./config.js";
+import { type AdapterConfig, resolveDelegate, resolveRollover } from "./config.js";
 import { createRuntime, type AcpRuntime } from "./runtime.js";
 import { makeCompressTool, isCompressSuccessText, isCompressNoopText } from "./compress-tool.js";
+import { makeAbsorbTool } from "./absorb-tool.js";
 import { makeDecompressTool } from "./decompress-tool.js";
 import { makeSearchTool } from "./search-tool.js";
 import { makeStatusTool } from "./status-tool.js";
 import { makeDelegateTool, makeDelegateWaitTool, makeDelegateCancelTool, runningRunsSnapshot, resetDelegateUsage, setDelegateDisplayUsage } from "./delegate-tool.js";
 import { makeCommands } from "./commands.js";
 import { coreOutToAgentMessages, extractText } from "./messages.js";
-import { buildAcpSystemPrompt, ACP_DELEGATE_PROMPT } from "./system-prompt.js";
+import { buildAcpSystemPrompt, ACP_DELEGATE_PROMPT, ROLLOVER_PROMPT_SECTION } from "./system-prompt.js";
 import { delegateStatusWidget } from "./fleet-widget.js";
 import { wireToolGuardrails } from "./tool-guardrails.js";
 import { debug, logError, logInfo, logWarn, logThrow, closeLogStream } from "./log.js";
 import { collectCoveredMessageIds, estimateTokens, lastUserMessageId, calibrateTokens, collectImageTokens, modelSupportsImages } from "./tokens.js";
+import { findHiddenPendingCompressCalls, mergeRestoredMessages, pendingHasWork, rolloverReportText, runRollover, shouldRollover, type RolloverResult } from "./rollover.js";
 import { checkForUpdate } from "./update.js";
 import {
   THROTTLE_RETRY_ERROR_MESSAGE,
@@ -110,6 +112,13 @@ function wireSessionLifecycle(pi: ExtensionAPI, runtime: AcpRuntime): void {
       pi.registerTool(makeDelegateWaitTool(pi));
       pi.registerTool(makeDelegateCancelTool(pi));
     }
+    // Absorb is a rollover-mode tool (#241): it defers removal to the next
+    // rollover, so registering it when rollover is off would let the model
+    // record pending drops that never apply. Resolved AFTER reloadConfig so
+    // user-config rollover:false wins over the factory-time default.
+    if (resolveRollover(runtime.adapter).enabled) {
+      pi.registerTool(makeAbsorbTool(runtime));
+    }
     // Headless hosts exit as soon as the turn ends; awaiting the check keeps
     // the process alive until a running install finishes. TUI stays
     // fire-and-forget so interactive startup is never blocked by npm.
@@ -235,7 +244,7 @@ function wireContextTransform(pi: ExtensionAPI, runtime: AcpRuntime): void {
         activeBefore: state.blocks.filter((b) => b.active).length,
       });
 
-      const turn = runtime.core.processTurn({ messages: coreMessages, state, config, tokenCount });
+      let turn = runtime.core.processTurn({ messages: coreMessages, state, config, tokenCount });
       await runtime.save(turn.state, ctx);
       // 密度校准（Phase 2）：processTurn 后调用，countTokens 用上一轮 density（1 轮延迟可忽略）。
       // real 侧 = provider 锚定 usage（缺失时锚点冻结，§5.9）；est 侧 = 发送视图估算
@@ -274,8 +283,34 @@ function wireContextTransform(pi: ExtensionAPI, runtime: AcpRuntime): void {
       activeAfter: turn.state.blocks.filter((b) => b.active).length,
     });
 
+    // Batch rollover (#241): inside a phase the model-visible history stays
+    // append-only — compress/absorb results are only PENDING markers — so the
+    // prompt-cache prefix is stable. When context pressure crosses the
+    // threshold (or /acp rollover forces it), apply ALL pending work in one
+    // batch: a single cache invalidation amortized over the whole phase.
+    const rollover = resolveRollover(runtime.adapter);
+    let rolloverResult: RolloverResult | null = null;
+    if (rollover.enabled) {
+      const pending = runtime.getRolloverPending(ctx);
+      if (shouldRollover({ enabled: rollover.enabled, tokenCount, limit: config.modelContextLimit, threshold: rollover.threshold, hasPending: pendingHasWork(pending), manual: false })) {
+        rolloverResult = await runRollover({ runtime, ctx, config, coreMessages, turn, modelId, imageTokens: collectImageTokens(entries, modelSupportsImages(ctx.model)), systemPromptTokens });
+        if (rolloverResult) {
+          turn = rolloverResult.turn;
+          logInfo("rollover", { sid, compressions: rolloverResult.compressionsApplied, absorbs: rolloverResult.absorbsApplied, before: rolloverResult.beforeTokens, after: rolloverResult.afterTokens, reclaimed: rolloverResult.reclaimed, errors: rolloverResult.errors.length });
+          debug.event("rollover", { sid, compressions: rolloverResult.compressionsApplied, absorbs: rolloverResult.absorbsApplied, before: rolloverResult.beforeTokens, after: rolloverResult.afterTokens });
+        }
+      }
+    }
+
     const originalById = collectOriginals(entries);
-    const rebuilt = coreOutToAgentMessages(turn.messages, originalById);
+    // Deferred (pending) compress calls are orphans to the kernel's
+    // hideConsumedCompressCalls: beyond the last two, older ones get hidden
+    // mid-history — an in-place rewrite that busts the cache prefix. Re-insert
+    // them at their original positions so the phase view stays append-only.
+    const pendingAfter = runtime.getRolloverPending(ctx);
+    const hiddenIds = findHiddenPendingCompressCalls(coreMessages, turn.messages, pendingAfter);
+    const merged = mergeRestoredMessages(turn.messages, coreMessages, hiddenIds, turn.state);
+    const rebuilt = coreOutToAgentMessages(merged, originalById);
     const debugOn = debug.enabled;
 
     const turnKey = lastUserMessageId(entries) ?? sid;
@@ -338,6 +373,10 @@ function wireContextTransform(pi: ExtensionAPI, runtime: AcpRuntime): void {
       }
     }
 
+    if (rolloverResult) {
+      rebuilt.push(rolloverReportMessage(rolloverResult));
+    }
+
     if (outcome !== null && outcome.cappedNow) {
       logWarn("nudge", { sid, event: "compress-retry-capped", failures: outcome.count });
       debug.event("compress-retry-capped", { sid, turnKey, failures: outcome.count });
@@ -372,7 +411,10 @@ function wireContextTransform(pi: ExtensionAPI, runtime: AcpRuntime): void {
 function wireSystemPrompt(pi: ExtensionAPI, runtime: AcpRuntime): void {
   pi.on("before_agent_start", (event) => {
     const delegate = runtime.adapter.delegate !== false;
-    const acp = buildAcpSystemPrompt(runtime.prompts);
+    let acp = buildAcpSystemPrompt(runtime.prompts);
+    if (resolveRollover(runtime.adapter).enabled) {
+      acp = `${acp}\n${ROLLOVER_PROMPT_SECTION}`;
+    }
     const prompt = delegate ? `${acp}\n${ACP_DELEGATE_PROMPT}` : acp;
     return { systemPrompt: formatSystemPromptForEvent(event.systemPrompt, prompt) };
   });
@@ -560,3 +602,14 @@ function nudgeMessage(nudge: NudgeDecision, blocks: CompressionBlock[], prompts:
     timestamp: Date.now(),
   } as AgentMessage;
 }
+
+// One-shot tail report after a rollover fires (same transient-tail pattern as
+// nudgeMessage): the model sees what was applied this round; the next context
+// event rebuilds the array from scratch, so it does not pollute context.
+function rolloverReportMessage(r: RolloverResult): AgentMessage {
+  return {
+    role: "user",
+    content: [{ type: "text", text: rolloverReportText(r) }],
+    timestamp: Date.now(),
+  } as AgentMessage;
+}
diff --git a/src/rollover.ts b/src/rollover.ts
new file mode 100644
index 0000000..89f7639
--- /dev/null
+++ b/src/rollover.ts
@@ -0,0 +1,248 @@
+import {
+  defaultCountTokens,
+  type AbsorbRecord,
+  type CompressionState,
+  type Config,
+  type CoreMessage,
+  type ProcessTurnResult,
+} from "acp-kernel";
+import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
+import type { AcpRuntime } from "./runtime.js";
+import { estimateTokens, collectCoveredMessageIds, calibrateTokens } from "./tokens.js";
+import { formatTokens } from "./tag-tokens.js";
+
+export interface PendingCompression {
+  startRef: string;
+  endRef: string;
+  summary: string;
+  topic?: string;
+  summaryMaxChars?: number;
+  callId: string;
+  createdAt: number;
+  estTokens: number;
+}
+
+export interface RolloverPending {
+  compressions: PendingCompression[];
+  absorbs: AbsorbRecord[];
+}
+
+export const DEFAULT_ROLLOVER_THRESHOLD = 0.7;
+
+export function emptyPending(): RolloverPending {
+  return { compressions: [], absorbs: [] };
+}
+
+export function pendingHasWork(p: RolloverPending | null): boolean {
+  return !!p && (p.compressions.length > 0 || p.absorbs.length > 0);
+}
+
+export function shouldRollover(opts: {
+  enabled: boolean;
+  tokenCount: number;
+  limit: number;
+  threshold: number;
+  hasPending: boolean;
+  manual: boolean;
+}): boolean {
+  return opts.enabled && opts.hasPending && opts.limit > 0 && (opts.manual || opts.tokenCount >= opts.threshold * opts.limit);
+}
+
+function refNum(ref: string): number | null {
+  const m = /^m(\d+)$/.exec(ref);
+  return m ? Number(m[1]) : null;
+}
+
+export function pendingOverlaps(p: RolloverPending | null, startRef: string, endRef: string): boolean {
+  if (!p) return false;
+  const s = refNum(startRef);
+  const e = refNum(endRef);
+  if (s === null || e === null) return false;
+  for (const c of p.compressions) {
+    const ps = refNum(c.startRef);
+    const pe = refNum(c.endRef);
+    if (ps === null || pe === null) continue;
+    if (s <= pe && e >= ps) return true;
+  }
+  return false;
+}
+
+export function rangeTokenEstimate(messages: CoreMessage[], state: CompressionState, startRef: string, endRef: string): number {
+  const s = refNum(startRef);
+  const e = refNum(endRef);
+  if (s === null || e === null) return 0;
+  let total = 0;
+  for (const m of messages) {
+    const ref = state.messageRefs.byRaw[m.id];
+    if (!ref) continue;
+    const n = refNum(ref);
+    if (n !== null && n >= s && n <= e) total += defaultCountTokens(m.text ?? "");
+  }
+  return total;
+}
+
+export interface RolloverInput {
+  runtime: AcpRuntime;
+  ctx: ExtensionContext;
+  config: Config;
+  coreMessages: CoreMessage[];
+  turn: ProcessTurnResult;
+  modelId: string;
+  imageTokens: Map<string, number>;
+  systemPromptTokens: number;
+}
+
+export interface RolloverResult {
+  turn: ProcessTurnResult;
+  compressionsApplied: number;
+  absorbsApplied: number;
+  errors: string[];
+  warnings: string[];
+  tokensCompressed: number;
+  beforeTokens: number;
+  afterTokens: number;
+  reclaimed: number;
+}
+
+export async function runRollover(input: RolloverInput): Promise<RolloverResult | null> {
+  const { runtime, ctx, config, coreMessages, turn, modelId, imageTokens, systemPromptTokens } = input;
+  const pending = runtime.getRolloverPending(ctx);
+  if (!pending || !pendingHasWork(pending)) return null;
+  const p = pending;
+
+  let state = turn.state;
+  const errors: string[] = [];
+  const warnings: string[] = [];
+  let compressionsApplied = 0;
+  let tokensCompressed = 0;
+  if (p.compressions.length > 0) {
+    const applied = runtime.core.applyCompression({
+      ranges: p.compressions.map((c) => ({
+        startRef: c.startRef,
+        endRef: c.endRef,
+        summary: c.summary,
+        topic: c.topic,
+        summaryMaxChars: c.summaryMaxChars,
+        compressCallId: c.callId,
+      })),
+      messages: turn.messages,
+      state,
+      config,
+    });
+    state = applied.state;
+    compressionsApplied = applied.result.blocksCreated;
+    tokensCompressed = applied.result.tokensCompressed;
+    errors.push(...applied.result.errors);
+    warnings.push(...applied.result.warnings);
+  }
+  if (p.absorbs.length > 0) {
+    state = {
+      ...state,
+      absorbed: [...(state.absorbed ?? []), ...p.absorbs],
+      stats: {
+        ...state.stats,
+        absorbedTokens: (state.stats.absorbedTokens ?? 0) + p.absorbs.reduce((sum, a) => sum + a.tokensReclaimed, 0),
+      },
+    };
+  }
+  runtime.setRolloverPending(ctx, null);
+  await runtime.save(state, ctx);
+
+  const probe = runtime.core.processTurn({ messages: coreMessages, state, config, tokenCount: 0 });
+  const newSentTokens = estimateTokens(probe.messages, collectCoveredMessageIds(state), imageTokens) + systemPromptTokens;
+  const newTokenCount = calibrateTokens(newSentTokens, runtime.density.densityFor(modelId));
+  const afterTurn = runtime.core.processTurn({ messages: coreMessages, state, config, tokenCount: newTokenCount });
+
+  const density = runtime.density.densityFor(modelId);
+  const beforeTokens = calibrateTokens(estimateTokens(turn.messages, collectCoveredMessageIds(turn.state), imageTokens), density);
+  const afterTokens = calibrateTokens(estimateTokens(afterTurn.messages, collectCoveredMessageIds(state), imageTokens), density);
+
+  return {
+    turn: afterTurn,
+    compressionsApplied,
+    absorbsApplied: p.absorbs.length,
+    errors,
+    warnings,
+    tokensCompressed,
+    beforeTokens,
+    afterTokens,
+    reclaimed: Math.max(0, beforeTokens - afterTokens),
+  };
+}
+
+function formatK(n: number): string {
+  return n >= 1000 ? `${(n / 1000).toFixed(1)}K` : String(n);
+}
+
+export function rolloverReportText(r: RolloverResult): string {
+  const applied: string[] = [];
+  if (r.compressionsApplied > 0) applied.push(`${r.compressionsApplied} compression(s)`);
+  if (r.absorbsApplied > 0) applied.push(`${r.absorbsApplied} absorb(s)`);
+  const lines = [
+    `\u25a3 ACP rollover | ${applied.join(" + ")} applied: ${formatK(r.beforeTokens)} \u2192 ${formatK(r.afterTokens)} tokens (~${formatK(r.reclaimed)} reclaimed)`,
+  ];
+  for (const w of r.warnings) lines.push(`Warning: ${w}`);
+  for (const e of r.errors) lines.push(`Error: ${e}`);
+  return lines.join("\n");
+}
+
+export function findHiddenPendingCompressCalls(input: CoreMessage[], output: CoreMessage[], pending: RolloverPending | null): Set<string> {
+  const hidden = new Set<string>();
+  if (!pending || pending.compressions.length === 0) return hidden;
+  const outputIds = new Set(output.map((m) => m.id));
+  for (const c of pending.compressions) {
+    for (const m of input) {
+      if (m.toolCallId !== c.callId) continue;
+      if (m.contentType === "tool-call" && m.toolName !== "compress") continue;
+      if (!outputIds.has(m.id)) hidden.add(m.id);
+    }
+  }
+  return hidden;
+}
+
+function renderRestored(m: CoreMessage, state: CompressionState): CoreMessage {
+  const ref = state.messageRefs.byRaw[m.id];
+  if (!ref) return m;
+  const body = m.text ?? "";
+  const tokens = state.tokenSnapshot[ref] ?? defaultCountTokens(body);
+  const type = m.contentType === "tool-call" || m.contentType === "tool-result" ? (m.toolName || "tool") : m.contentType;
+  const tag = `\x3cacp tokens="${formatTokens(tokens)}" type="${type}"\x3e${ref}\x3c/acp\x3e`;
+  return { ...m, text: body ? `${tag}\n${body}` : tag };
+}
+
+export function mergeRestoredMessages(coreOut: CoreMessage[], input: CoreMessage[], hiddenIds: Set<string>, state: CompressionState): CoreMessage[] {
+  if (hiddenIds.size === 0) return coreOut;
+  const outIds = new Set(coreOut.map((m) => m.id));
+  const restored = new Map<string, CoreMessage>();
+  for (const m of input) {
+    if (hiddenIds.has(m.id) && !outIds.has(m.id)) restored.set(m.id, renderRestored(m, state));
+  }
+  if (restored.size === 0) return coreOut;
+  const inputPos = new Map<string, number>();
+  input.forEach((m, i) => {
+    if (!inputPos.has(m.id)) inputPos.set(m.id, i);
+  });
+  const merged: CoreMessage[] = [];
+  const emitted = new Set<string>();
+  let lastPos = -1;
+  for (const cm of coreOut) {
+    const pos = inputPos.get(cm.id);
+    const anchor = pos !== undefined ? pos : lastPos;
+    if (pos !== undefined) lastPos = pos;
+    for (const [rid, rm] of restored) {
+      if (emitted.has(rid)) continue;
+      if ((inputPos.get(rid) ?? -1) < anchor) {
+        merged.push(rm);
+        emitted.add(rid);
+      }
+    }
+    merged.push(cm);
+  }
+  for (const [rid, rm] of restored) {
+    if (!emitted.has(rid)) {
+      merged.push(rm);
+      emitted.add(rid);
+    }
+  }
+  return merged;
+}
diff --git a/src/runtime.ts b/src/runtime.ts
index 908b5bb..c2af645 100644
--- a/src/runtime.ts
+++ b/src/runtime.ts
@@ -12,6 +12,7 @@ import { resolveConfig, type AdapterConfig } from "./config.js";
 import { DensityEstimator } from "./density.js";
 import { entriesToCoreMessages, extractText, matchesStoredText, messageIdentity, messageRef } from "./messages.js";
 import { SessionStateStore, type LiveRefOrigin } from "./state.js";
+import type { RolloverPending } from "./rollover.js";
 import { loadUserConfig, applyUserConfig } from "./user-config.js";
 import { ThrottleEpisode } from "./throttle-retry.js";
 import { logInfo, logWarn, setDebugEnabled } from "./log.js";
@@ -89,6 +90,14 @@ export interface AcpRuntime {
   reloadConfig(cwd: string): Promise<void>;
   stateFor(ctx: ExtensionContext, liveMessages?: AgentMessage[]): Promise<{ state: CompressionState; coreMessages: ReturnType<typeof entriesToCoreMessages>; entries: SessionEntry[] }>;
   save(state: CompressionState, ctx: ExtensionContext): Promise<void>;
+  /** Batch-rollover pending work (#241): compressions recorded by the compress
+   *  tool and absorbs recorded by the absorb tool. Applied in one batch at
+   *  rollover (usage ≥ threshold or /acp rollover) so the model-visible
+   *  history stays append-only within a phase. */
+  getRolloverPending(ctx: ExtensionContext): RolloverPending | null;
+  /** Synchronous on purpose: callers do a read-modify-write of the pending
+   *  list and need it atomic under parallel tool calls (no await between). */
+  setRolloverPending(ctx: ExtensionContext, pending: RolloverPending | null): void;
   acquireLock(sid: string): Promise<() => void>;
   /** Per-session overflow self-heal state (learned window + armed emergency).
    *  Keyed by session id so concurrent sessions cannot share an episode. */
@@ -392,6 +401,15 @@ export function createRuntime(adapter: AdapterConfig): AcpRuntime {
     await store.save(state, sm.getSessionFile() ?? undefined, sm.getSessionId());
   }
 
+  function getRolloverPending(ctx: ExtensionContext): RolloverPending | null {
+    const sm = ctx.sessionManager;
+    return store.getRolloverPending(sm.getSessionFile() ?? undefined, sm.getSessionId());
+  }
+  async function setRolloverPending(ctx: ExtensionContext, pending: RolloverPending | null): Promise<void> {
+    const sm = ctx.sessionManager;
+    await store.setRolloverPending(sm.getSessionFile() ?? undefined, sm.getSessionId(), pending);
+  }
+
   function noteActiveBlocks(sid: string, activeBlockIds: string[]): boolean {
     const current = new Set(activeBlockIds);
     const prev = lastActiveBlockIds.get(sid);
@@ -403,4 +421,4 @@ export function createRuntime(adapter: AdapterConfig): AcpRuntime {
     lastActiveBlockIds.delete(sid);
   }
 
-  return { core, store, density, setCountModel: (m) => { countModelId = m; }, noteActiveBlocks, clearSessionTracking, get adapter() { return adapterRef; }, setAdapter: (a) => { adapterRef = a; }, get prompts() { return promptsRef; }, setPrompts: (p) => { promptsRef = p; }, markNudgeShown: (k) => { nudgeShownTurns.add(k); }, nudgeShownFor: (k) => nudgeShownTurns.has(k), clearNudgeTracking: () => { nudgeShownTurns.clear(); }, noteCompressOutcomes, compressRetryCappedFor, clearCompressRetryTracking, liveContextLimit, configFor, reloadConfig, stateFor, save, acquireLock, overflowFor, overflowDrop, throttleFor, throttleDrop };}
+  return { core, store, density, setCountModel: (m) => { countModelId = m; }, noteActiveBlocks, clearSessionTracking, get adapter() { return adapterRef; }, setAdapter: (a) => { adapterRef = a; }, get prompts() { return promptsRef; }, setPrompts: (p) => { promptsRef = p; }, markNudgeShown: (k) => { nudgeShownTurns.add(k); }, nudgeShownFor: (k) => nudgeShownTurns.has(k), clearNudgeTracking: () => { nudgeShownTurns.clear(); }, noteCompressOutcomes, compressRetryCappedFor, clearCompressRetryTracking, liveContextLimit, configFor, reloadConfig, stateFor, save, getRolloverPending, setRolloverPending, acquireLock, overflowFor, overflowDrop, throttleFor, throttleDrop };}
diff --git a/src/state.ts b/src/state.ts
index becd87d..0a97b0a 100644
--- a/src/state.ts
+++ b/src/state.ts
@@ -1,7 +1,8 @@
 import { promises as fs } from "node:fs";
 import * as path from "node:path";
-import { createInitialState, type CompressionState } from "acp-kernel";
+import { createInitialState, type AbsorbRecord, type CompressionState } from "acp-kernel";
 import { logError, logInfo, logWarn } from "./log.js";
+import type { RolloverPending } from "./rollover.js";
 
 const STATE_SUFFIX = ".acp.json";
 
@@ -13,6 +14,7 @@ export interface LiveRefOrigin {
 interface StateCacheSlot {
   state: CompressionState;
   liveRefOrigins: LiveRefOrigin[];
+  rolloverPending: RolloverPending | null;
 }
 
 function stateFileFor(sessionFile: string | undefined): string | null {
@@ -57,13 +59,15 @@ export class SessionStateStore {
     if (cached) return cached.state;
     let state = createInitialState();
     let liveRefOrigins: LiveRefOrigin[] = [];
+    let rolloverPending: RolloverPending | null = null;
     if (file) {
       try {
         const raw = await fs.readFile(file, "utf8");
-        const parsed = JSON.parse(raw) as CompressionState & { liveRefOrigins?: unknown };
+        const parsed = JSON.parse(raw) as CompressionState & { liveRefOrigins?: unknown; rolloverPending?: unknown };
         if (parsed && Array.isArray(parsed.blocks)) {
           state = mergeInitialState(parsed);
           liveRefOrigins = parseLiveRefOrigins(parsed.liveRefOrigins);
+          rolloverPending = parseRolloverPending(parsed.rolloverPending);
         }
       } catch (e) {
         const code = (e as NodeJS.ErrnoException).code;
@@ -80,7 +84,7 @@ export class SessionStateStore {
         if (parentState) state = parentState;
       }
     }
-    this.cache.set(key, { state, liveRefOrigins });
+    this.cache.set(key, { state, liveRefOrigins, rolloverPending });
     return state;
   }
 
@@ -88,15 +92,17 @@ export class SessionStateStore {
     const file = stateFileFor(sessionFile);
     if (!file) return;
     const key = cacheKey(sessionFile, sessionId);
-    const liveRefOrigins = this.cache.get(key)?.liveRefOrigins ?? [];
-    this.cache.set(key, { state, liveRefOrigins });
+    const slot = this.cache.get(key);
+    const liveRefOrigins = slot?.liveRefOrigins ?? [];
+    const rolloverPending = slot?.rolloverPending ?? null;
+    this.cache.set(key, { state, liveRefOrigins, rolloverPending });
     const dir = path.dirname(file);
     await fs.mkdir(dir, { recursive: true }).catch((e: unknown) => {
       logError("state", { event: "save-mkdir-failed", dir, error: e instanceof Error ? e.message : String(e) });
     });
     const tmp = path.join(dir, `.acp-tmp-${path.basename(file)}`);
     try {
-      await fs.writeFile(tmp, JSON.stringify({ ...state, liveRefOrigins }), "utf8");
+      await fs.writeFile(tmp, JSON.stringify({ ...state, liveRefOrigins, rolloverPending }), "utf8");
       await fs.rename(tmp, file);
     } catch (e) {
       logError("state", { event: "save-failed", file, error: e instanceof Error ? e.message : String(e) });
@@ -110,7 +116,17 @@ export class SessionStateStore {
   setLiveRefOrigins(sessionFile: string | undefined, sessionId: string, origins: LiveRefOrigin[]): void {
     const key = cacheKey(sessionFile, sessionId);
     const slot = this.cache.get(key);
-    if (slot) this.cache.set(key, { state: slot.state, liveRefOrigins: [...origins] });
+    if (slot) this.cache.set(key, { state: slot.state, liveRefOrigins: [...origins], rolloverPending: slot.rolloverPending });
+  }
+
+  getRolloverPending(sessionFile: string | undefined, sessionId: string): RolloverPending | null {
+    return this.cache.get(cacheKey(sessionFile, sessionId))?.rolloverPending ?? null;
+  }
+
+  setRolloverPending(sessionFile: string | undefined, sessionId: string, pending: RolloverPending | null): void {
+    const key = cacheKey(sessionFile, sessionId);
+    const slot = this.cache.get(key);
+    if (slot) this.cache.set(key, { state: slot.state, liveRefOrigins: slot.liveRefOrigins, rolloverPending: pending });
   }
 
   invalidate(): void {
@@ -155,6 +171,27 @@ function parseLiveRefOrigins(value: unknown): LiveRefOrigin[] {
   });
 }
 
+function parseRolloverPending(value: unknown): RolloverPending | null {
+  if (!value || typeof value !== "object") return null;
+  const p = value as { compressions?: unknown; absorbs?: unknown };
+  const compressions = Array.isArray(p.compressions)
+    ? p.compressions.filter((item): item is RolloverPending["compressions"][number] => {
+        if (!item || typeof item !== "object") return false;
+        const c = item as Record<string, unknown>;
+        return typeof c.startRef === "string" && typeof c.endRef === "string" && typeof c.summary === "string" && typeof c.callId === "string";
+      })
+    : [];
+  const absorbs = Array.isArray(p.absorbs)
+    ? p.absorbs.filter((item): item is RolloverPending["absorbs"][number] => {
+        if (!item || typeof item !== "object") return false;
+        const a = item as Record<string, unknown>;
+        return typeof a.toolCallId === "string" && typeof a.resultMessageId === "string" && typeof a.summary === "string";
+      })
+    : [];
+  if (compressions.length === 0 && absorbs.length === 0) return null;
+  return { compressions, absorbs };
+}
+
 function mergeInitialState(parsed: CompressionState): CompressionState {
   const fresh = createInitialState();
   return {
@@ -163,7 +200,18 @@ function mergeInitialState(parsed: CompressionState): CompressionState {
     tokenSnapshot: parsed.tokenSnapshot ?? fresh.tokenSnapshot,
     nudge: { ...fresh.nudge, ...(parsed.nudge ?? {}) },
     stats: { ...fresh.stats, ...(parsed.stats ?? {}) },
+    absorbed: parseAbsorbedRecords(parsed.absorbed),
     nextBlockId: parsed.nextBlockId ?? fresh.nextBlockId,
     nextRunId: parsed.nextRunId ?? fresh.nextRunId,
   };
 }
+
+function parseAbsorbedRecords(value: unknown): AbsorbRecord[] | undefined {
+  if (!Array.isArray(value)) return undefined;
+  const records = value.filter((item): item is AbsorbRecord => {
+    if (!item || typeof item !== "object") return false;
+    const a = item as Record<string, unknown>;
+    return typeof a.toolCallId === "string" && typeof a.resultMessageId === "string" && typeof a.summary === "string";
+  });
+  return records.length > 0 ? records : undefined;
+}
diff --git a/src/status-tool.ts b/src/status-tool.ts
index 74b6b78..b8539a5 100644
--- a/src/status-tool.ts
+++ b/src/status-tool.ts
@@ -6,7 +6,8 @@ import { estimateTokens, collectCoveredMessageIds, calibrateTokens, collectImage
 import { getSystemPromptText } from "./compat.js";
 import { logThrow } from "./log.js";
 import { getDelegateUsage } from "./delegate-tool.js";
-import { resolveDelegate } from "./config.js";
+import { resolveDelegate, resolveRollover } from "./config.js";
+import { pendingHasWork } from "./rollover.js";
 
 const StatusParams = Type.Object({
   scope: Type.Optional(Type.Union([Type.Literal("compressed"), Type.Literal("uncompressed")], { description: '"compressed" = drill into blocks; "uncompressed" = show visible messages/ranges. Default: overview.' })),
@@ -60,11 +61,12 @@ async function handleStatus(args: StatusArgs, runtime: AcpRuntime, ctx: Extensio
   const systemPromptText = getSystemPromptText(ctx);
   const systemPromptTokens = systemPromptText ? defaultCountTokens(systemPromptText) : 0;
   const sentTokens = estimateTokens(coreMessages, coveredIds, collectImageTokens(entries, modelSupportsImages(ctx.model))) + systemPromptTokens;
+  const tokenCount = calibrateTokens(sentTokens, runtime.density.densityFor(modelId));
   const turn = runtime.core.processTurn({
     messages: coreMessages,
     state,
     config,
-    tokenCount: calibrateTokens(sentTokens, runtime.density.densityFor(modelId)),
+    tokenCount,
   });
   const processed = turn.messages;
 
@@ -94,6 +96,20 @@ async function handleStatus(args: StatusArgs, runtime: AcpRuntime, ctx: Extensio
         : `Nudge: idle — ${nudge.reason}`,
     );
   }
+  const rollover = resolveRollover(runtime.adapter);
+  if (rollover.enabled) {
+    const pending = runtime.getRolloverPending(ctx);
+    if (pending && pendingHasWork(pending)) {
+      const pendingTokens =
+        pending.compressions.reduce((s, c) => s + c.estTokens, 0) +
+        pending.absorbs.reduce((s, a) => s + a.tokensReclaimed, 0);
+      const usagePct = config.modelContextLimit > 0 ? Math.round((tokenCount / config.modelContextLimit) * 100) : 0;
+      extra.push("");
+      extra.push(
+        `Rollover: ${pending.compressions.length} pending compression(s) + ${pending.absorbs.length} absorb(s) — ~${pendingTokens.toLocaleString()} tokens pending (threshold ${Math.round(rollover.threshold * 100)}%, current ${usagePct}%)`,
+      );
+    }
+  }
   if (ranges.length > 0 || protectedRanges.length > 0) {
     extra.push("");
     // Reuse the kernel's merged range formatter so acp_status, the nudge,
diff --git a/src/system-prompt.ts b/src/system-prompt.ts
index 78ce8a1..8d06f7b 100644
--- a/src/system-prompt.ts
+++ b/src/system-prompt.ts
@@ -71,6 +71,19 @@ Retries are capped; when the cap is reached the error is surfaced to the user un
 `;
 }
 
+export const ROLLOVER_PROMPT_SECTION = `
+ROLLOVER MODE (deferred compression)
+
+This session runs in batch-rollover mode: visible history is append-only within a phase, so the provider's prompt cache stays warm. Compression is therefore DEFERRED:
+- A successful compress records the range as pending — it does NOT rewrite history yet. The range stays visible in your context until the next rollover. The panel says "recorded for next rollover".
+- A rollover applies ALL pending compressions and absorbs in ONE batch — a single history rewrite, one cache invalidation, amortized over the phase. It fires automatically when context usage crosses the rollover threshold (acp_status shows the threshold and what is pending), or immediately via /acp-rollover.
+- Refs stay stable while ranges are pending (nothing is rewritten yet). They renumber only when a rollover applies. After a rollover, pre-rollover refs are stale — run acp_status before compressing again.
+- Do NOT re-compress a range that is already pending — overlapping ranges are rejected.
+- absorb({ ref, summary }) distills a large tool result into a summary you write. The original output stays visible until the next rollover, then is removed in the same batch rewrite.
+- Pending content is still visible, so you can keep using it; the cost is temporary extra context. Record compressions as soon as content is consumed — don't wait for the threshold.
+- When a rollover fires, a one-shot "▣ ACP rollover" report is appended at the end of context: what was applied and the token delta.
+`;
+
 export const ACP_DELEGATE_PROMPT = `
 ACP_DELEGATE NOTIFICATIONS
 
diff --git a/tests/compress-retry.test.ts b/tests/compress-retry.test.ts
index b507e93..a70476d 100644
--- a/tests/compress-retry.test.ts
+++ b/tests/compress-retry.test.ts
@@ -143,7 +143,7 @@ test("noteCompressOutcomes: counts, caps, resets on success, resets per turn, ne
 
 test("compress tool accepts JSON-encoded string content (non-strict-tool providers)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-retry-str.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const entries = [userMsg("e1", ZH)];
@@ -164,7 +164,7 @@ test("compress tool accepts JSON-encoded string content (non-strict-tool provide
 
 test("compress tool THROWS on garbage string content (isError:true → counted by the outcome tracker)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-retry-str2.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const entries = [userMsg("e1", ZH)];
@@ -186,7 +186,7 @@ test("compress tool THROWS on garbage string content (isError:true → counted b
 
 test("failed compress toolResults never inject a transient retry prompt; the error itself persists in context (#223)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-retry-it1.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
 
@@ -230,7 +230,7 @@ test("failed compress toolResults never inject a transient retry prompt; the err
 
 test("neutral and no-op outcomes inject nothing; only the counter state changes", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-retry-noop.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
 
@@ -300,7 +300,7 @@ test("noteCompressOutcomes: no-op panels advance the counter toward the cap", ()
 
 test("emergency nudge stops re-injecting once the turn's cap is burned (issue #6 loop breaker)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 180_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 180_000 })(api as any);
   const stateFile = "/tmp/pai-acp-retry-emerg.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
 
diff --git a/tests/compress-tool.test.ts b/tests/compress-tool.test.ts
index 63886b9..88d60f9 100644
--- a/tests/compress-tool.test.ts
+++ b/tests/compress-tool.test.ts
@@ -61,7 +61,7 @@ async function runContextRound(handlers: Map<string, any[]>, ctx: any) {
 
 test("compress beforeTokens at density=1 is uncalibrated estimateTokens", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-compress-density-a.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const entries = [userMsg("e1", "hello world"), userMsg("e2", ZH)];
@@ -81,7 +81,7 @@ test("compress beforeTokens at density=1 is uncalibrated estimateTokens", async
 
 test("compress beforeTokens scales with calibrated density (Phase 2)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-compress-density-b.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const entries = [userMsg("e1", "hello world"), userMsg("e2", ZH)];
@@ -118,7 +118,7 @@ test("compress beforeTokens scales with calibrated density (Phase 2)", async ()
 // cumulative summary mass of all blocks, exactly in long sessions.
 test("compress afterTokens is measured on the same sent-view scale as beforeTokens (multi-block)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-compress-scales.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const entries = [userMsg("e1", "hello world"), userMsg("e2", ZH), userMsg("e3", ZH2), userMsg("e4", ZH2)];
diff --git a/tests/decompress-cmd.test.ts b/tests/decompress-cmd.test.ts
index de442a5..adfb21c 100644
--- a/tests/decompress-cmd.test.ts
+++ b/tests/decompress-cmd.test.ts
@@ -51,7 +51,7 @@ function fakeCtx(entries: any[], stateFile: string, notifies: string[]) {
 // (append semantics: the block stays folded, content shown via notify).
 test("/acp-decompress returns a block's content and stays repeatable (append mode)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const stateFile = "/tmp/pai-acp-decompress-it.session.json";
   await cleanState(stateFile);
@@ -99,7 +99,7 @@ test("/acp-decompress returns a block's content and stays repeatable (append mod
 
 test("/acp-decompress rejects invalid input with a usage message", async () => {
   const { api } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const notifies: string[] = [];
   const ctx = fakeCtx([], "/tmp/pai-acp-decompress-invalid.session.json", notifies);
@@ -118,7 +118,7 @@ test("/acp-decompress rejects invalid input with a usage message", async () => {
 
 test("/acp-decompress reports not-found for a valid id with no matching block", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const notifies: string[] = [];
   const ctx = fakeCtx([userMsg("e1", "only message")], "/tmp/pai-acp-decompress-nf.session.json", notifies);
diff --git a/tests/decompress-tool.test.ts b/tests/decompress-tool.test.ts
index 793b73c..04b83e2 100644
--- a/tests/decompress-tool.test.ts
+++ b/tests/decompress-tool.test.ts
@@ -58,7 +58,7 @@ function fakeCtxFullTree(allEntries: any[], activeEntries: any[], stateFile: str
 // handles + ctx so each test can drive the decompress tool.
 async function setupWithCompressedBlock() {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const stateFile = "/tmp/pai-acp-decompress-tool-it.session.json";
   await cleanState(stateFile);
@@ -173,7 +173,7 @@ test("decompress keeps the block active after a file-mode call", async () => {
 
 test("decompress restores a block's original text via getEntry fallback after tree navigation (undo)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-decompress-fallback-undo.session.json";
   await cleanState(stateFile);
   const longText = "This is a detailed message that needs to be compressed. ".repeat(130);
@@ -210,7 +210,7 @@ test("decompress restores a block's original text via getEntry fallback after tr
 
 test("decompress keeps the degraded message when the ref is gone from both branch and full tree", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-decompress-fallback-gone.session.json";
   await cleanState(stateFile);
   const longText = "This is a detailed message that needs to be compressed. ".repeat(130);
@@ -239,7 +239,7 @@ test("decompress keeps the degraded message when the ref is gone from both branc
 
 test("decompress restores multi tool-call assistant messages (refs carry # suffix) after undo", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-decompress-fallback-tools.session.json";
   await cleanState(stateFile);
   const filler = (n: string) => `filler ${n} `.repeat(400);
@@ -285,7 +285,7 @@ test("decompress restores multi tool-call assistant messages (refs carry # suffi
 
 test("decompress survives repeated compress → navigate → decompress cycles (state not lost)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/pai-acp-decompress-fallback-cycles.session.json";
   await cleanState(stateFile);
   const longText = "This is a detailed message that needs to be compressed. ".repeat(130);
diff --git a/tests/density-usage-fixes.test.ts b/tests/density-usage-fixes.test.ts
index 7ebd460..f0cff21 100644
--- a/tests/density-usage-fixes.test.ts
+++ b/tests/density-usage-fixes.test.ts
@@ -63,7 +63,7 @@ async function runCalibratedScenario(realScale: number) {
   const stateFile = `/tmp/pai-acp-calibrated-${realScale}.session.json`;
   await rm(`${stateFile}.acp.json`, { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   let usage = 0;
   const ctx: any = {
diff --git a/tests/image-tokens.test.ts b/tests/image-tokens.test.ts
index 259ce1c..b1d4734 100644
--- a/tests/image-tokens.test.ts
+++ b/tests/image-tokens.test.ts
@@ -115,7 +115,7 @@ test("sent-view token count includes image tokens (vision model)", async () => {
   await rm(logFile, { force: true });
   process.env.ACP_LOG_FILE = logFile;
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 10_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 10_000 })(api as any);
   const entries = Array.from({ length: 8 }, (_, i) => imgEntry(`e${i}`));
   const ctx = ctxWithModel(entries, 10_000, ["text", "image"]);
   await handlers.get("context")![0]!({ type: "context", messages: entries.map((e) => e.message) }, ctx);
@@ -129,7 +129,7 @@ test("sent-view token count ignores images for non-vision models", async () => {
   await rm(logFile, { force: true });
   process.env.ACP_LOG_FILE = logFile;
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 10_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 10_000 })(api as any);
   const entries = Array.from({ length: 8 }, (_, i) => imgEntry(`e${i}`));
   const ctx = ctxWithModel(entries, 10_000, ["text"]);
   await handlers.get("context")![0]!({ type: "context", messages: entries.map((e) => e.message) }, ctx);
@@ -144,7 +144,7 @@ test("sent-view token count ignores images for non-vision models", async () => {
 test("emergency nudge fires when images push the sent view past the window", async () => {
   await rm(`${STATE_FILE}.acp.json`, { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 10_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 10_000 })(api as any);
   const filler = "lorem ".repeat(300);
   const entries = [
     ...Array.from({ length: 30 }, (_, i) => ({
@@ -167,7 +167,7 @@ test("emergency nudge fires when images push the sent view past the window", asy
 test("identical text-only session stays quiet (same window)", async () => {
   await rm(`${STATE_FILE}.acp.json`, { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 10_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 10_000 })(api as any);
   const entries = Array.from({ length: 8 }, (_, i) => textEntry(`e${i}`, "x"));
   const ctx = ctxWithModel(entries, 10_000, ["text", "image"]);
   const r = await handlers.get("context")![0]!({ type: "context", messages: entries.map((e) => e.message) }, ctx);
@@ -178,7 +178,7 @@ test("identical text-only session stays quiet (same window)", async () => {
 test("pi host: image-only user message survives the transform with its ref tag", async () => {
   await rm(`${STATE_FILE}.acp.json`, { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const entries = [imgEntry("e1")];
   const ctx = ctxWithModel(entries, 200_000, ["text", "image"]);
   const r = await handlers.get("context")![0]!({ type: "context", messages: entries.map((e) => e.message) }, ctx);
diff --git a/tests/integration.test.ts b/tests/integration.test.ts
index 989c6f1..020ca04 100644
--- a/tests/integration.test.ts
+++ b/tests/integration.test.ts
@@ -59,12 +59,12 @@ function userMsg(id: string, text: string) {
   return { type: "message", id, parentId: null, timestamp: "", message: { role: "user", content: text, timestamp: Date.now() } };
 }
 
-test("factory registers the compress tool and 5 flat commands", () => {
+test("factory registers the compress tool and 6 flat commands", () => {
   const { api, handlers } = captureApi();
-  createAcpExtension()(api as any);
+  createAcpExtension({ rollover: false })(api as any);
 
   assert.ok(api.tools.some((t) => t.name === "compress"), "compress tool registered");
-  assert.deepEqual([...api.commands.keys()].sort(), ["acp", "acp-decompress", "acp-search", "acp-status", "acp-subagents"]);
+  assert.deepEqual([...api.commands.keys()].sort(), ["acp", "acp-decompress", "acp-rollover", "acp-search", "acp-status", "acp-subagents"]);
   assert.ok(handlers.has("context"), "context event wired");
   assert.ok(handlers.has("session_before_compact"), "compaction-disable wired");
   assert.ok(handlers.has("before_agent_start"), "system-prompt wired");
@@ -72,14 +72,14 @@ test("factory registers the compress tool and 5 flat commands", () => {
 
 test("session_before_compact cancels Pi's auto-compaction", () => {
   const { api, handlers } = captureApi();
-  createAcpExtension()(api as any);
+  createAcpExtension({ rollover: false })(api as any);
   const result = handlers.get("session_before_compact")![0]!({}, {});
   assert.deepEqual(result, { cancel: true });
 });
 
 test("before_agent_start appends the ACP system prompt", () => {
   const { api, handlers } = captureApi();
-  createAcpExtension()(api as any);
+  createAcpExtension({ rollover: false })(api as any);
   const result = handlers.get("before_agent_start")![0]!({ systemPrompt: "BASE" }, {});
   assert.ok(result.systemPrompt.startsWith("BASE"));
   assert.ok(result.systemPrompt.includes("compress"));
@@ -88,7 +88,7 @@ test("before_agent_start appends the ACP system prompt", () => {
 
 test("context handler tags every message with a ref even when length matches event.messages", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const entries = [userMsg("e1", "first"), userMsg("e2", "second"), userMsg("e3", "third")];
   const ctx = fakeCtx(entries, "/tmp/nonexistent-pai-acp-it.session.json");
@@ -106,7 +106,7 @@ test("context handler tags every message with a ref even when length matches eve
 
 test("context handler works under omp (oh-my-pi) where sessionManager exposes getBranch() not buildContextEntries()", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const entries = [userMsg("e1", "first"), userMsg("e2", "second")];
   const ctx = {
@@ -129,7 +129,7 @@ test("context handler works under omp (oh-my-pi) where sessionManager exposes ge
 
 test("omp context handler keeps the current (not-yet-persisted) user message: branch lags event.messages by one", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const stateFile = "/tmp/nonexistent-pai-acp-omp-lag.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   // Simulate omp's real timing: the branch only holds the PREVIOUS turn's
@@ -166,7 +166,7 @@ test("omp context handler keeps the current (not-yet-persisted) user message: br
 
 test("omp live message keeps the same entry id once persisted (stable refs across turns)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   // Turn 1: branch empty (brand-new session), event carries the first message.
   const turn1Messages = [{ role: "user", content: [{ type: "text", text: "hello" }], timestamp: Date.now() }];
@@ -204,7 +204,7 @@ test("omp live message keeps the same entry id once persisted (stable refs acros
 
 test("omp migrates tagged live refs to stable entry ids", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-identity.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const texts = ["This tagged message must retain its stable persisted identity. ".repeat(130), "filler two ".repeat(400)];
@@ -221,7 +221,7 @@ test("omp migrates tagged live refs to stable entry ids", async () => {
 
 test("omp matches a persisted context suffix before assigning live refs", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-suffix.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const texts = ["This suffix message must retain its persisted identity. ".repeat(130), "filler two ".repeat(400)];
@@ -236,7 +236,7 @@ test("omp matches a persisted context suffix before assigning live refs", async
 
 test("omp rejects a non-contiguous persisted subsequence", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-gap.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const persisted = [userMsg("e1", "A"), userMsg("gap", "X"), userMsg("e2", "B")];
@@ -253,7 +253,7 @@ test("omp rejects a non-contiguous persisted subsequence", async () => {
 
 test("omp rejects ambiguous equal-length persisted runs", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-ambiguous-run.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const persisted = [userMsg("e1", "same"), userMsg("gap", "different"), userMsg("e2", "same")];
@@ -271,7 +271,7 @@ test("omp rejects ambiguous equal-length persisted runs", async () => {
 
 test("omp migrates a live ref after the provider context evicts its prefix", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-shifted-live-ref.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   let persisted: ReturnType<typeof userMsg>[] = [];
@@ -289,7 +289,7 @@ type PersistedEntry = { type: "message"; id: string; parentId: null; timestamp:
 
 test("omp does not bind a different toolCallId with identical visible text to the persisted identity", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-toolcallid.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const toolResult = (toolCallId: string) => ({
@@ -313,7 +313,7 @@ test("omp does not bind a different toolCallId with identical visible text to th
 
 test("omp does not bind differing image content with identical visible text to the persisted identity", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-image.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const imgMsg = (data: string) => ({
@@ -337,7 +337,7 @@ test("omp does not bind differing image content with identical visible text to t
 
 test("omp matches emergency-truncated tool results before compression", async (t) => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const dir = await mkdtemp(join(tmpdir(), "pai-acp-omp-truncation-"));
   t.after(() => rm(dir, { recursive: true, force: true }));
   const stateFile = join(dir, "session.json");
@@ -362,7 +362,7 @@ test("omp matches emergency-truncated tool results before compression", async (t
 
 test("omp does not collapse distinct multimodal user messages with identical text (images survive)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
 
   const persistedImage = { type: "image", mimeType: "image/png", data: "persisted-image-payload" };
   const liveImage = { type: "image", mimeType: "image/png", data: "live-image-payload" };
@@ -397,7 +397,7 @@ test("omp does not collapse distinct multimodal user messages with identical tex
 
 test("omp does not collapse distinct multimodal tool results with identical text (images survive)", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
 
   const persistedImage = { type: "image", mimeType: "image/png", data: "persisted-image-payload" };
   const liveImage = { type: "image", mimeType: "image/png", data: "live-image-payload" };
@@ -432,7 +432,7 @@ test("omp does not collapse distinct multimodal tool results with identical text
 
 test("acp_status refs remain usable by the next compress call", async () => {
   const { api } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-status-compress.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const originalText = "This range is reported by acp_status and must remain addressable by compress. ".repeat(130);
@@ -448,7 +448,7 @@ test("acp_status refs remain usable by the next compress call", async () => {
 
 test("omp rebuilds refs after stale live state before status compression", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-stale-live.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const longText = "This stale live state must be rebuilt against the current persisted branch. ".repeat(130);
@@ -470,7 +470,7 @@ test("omp rebuilds refs after stale live state before status compression", async
 
 test("system prompt sources compression rules from acp-kernel (no hardcoded drift, no markers)", () => {
   const { api, handlers } = captureApi();
-  createAcpExtension()(api as any);
+  createAcpExtension({ rollover: false })(api as any);
   const result = handlers.get("before_agent_start")![0]!({ systemPrompt: "" }, {});
   const sp = result.systemPrompt;
   // kernel constants inlined (regression guard against reverting to a hardcoded copy)
@@ -494,7 +494,7 @@ test("system prompt sources compression rules from acp-kernel (no hardcoded drif
 
 test("context handler persists state so a second call is idempotent on the same entries", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
 
   const entries = [userMsg("e1", "alpha"), userMsg("e2", "beta")];
   const ctx = fakeCtx(entries, "/tmp/nonexistent-pai-acp-it2.session.json");
@@ -510,7 +510,7 @@ test("context handler persists state so a second call is idempotent on the same
 });
 test("omp migrates assistant tool-call refs after prefix eviction", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-assistant-origin.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const assistant = (id: string) => ({ role: "assistant", content: [{ type: "toolCall", id, name: "read", arguments: { path: "x" } }], timestamp: Date.now() });
@@ -529,7 +529,7 @@ test("omp migrates assistant tool-call refs after prefix eviction", async () =>
 
 test("omp migrates parallel assistant tool-call child refs after prefix eviction", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-parallel-origin.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const assistant = () => ({ role: "assistant", content: [
@@ -560,13 +560,13 @@ test("omp reloads assistant origins before migrating after prefix eviction", asy
   let persisted: ReturnType<typeof userMsg>[] = [];
   const makeCtx = () => ({ ...fakeCtx(persisted, stateFile), sessionManager: { getBranch: () => persisted, getSessionId: () => "test-session", getSessionFile: () => stateFile } });
   const first = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(first.api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(first.api);
   await first.handlers.get("context")![0]!({ type: "context", messages: [assistant("call-a")] }, makeCtx());
   const initial = JSON.parse(await readFile(`${stateFile}.acp.json`, "utf8"));
   const ref = initial.messageRefs.byRaw["live-0"];
   assert.ok(ref);
   const second = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(second.api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(second.api);
   persisted = [{ type: "message", id: "e-assistant", parentId: null, timestamp: "", message: assistant("call-a") }];
   await second.handlers.get("context")![0]!({ type: "context", messages: [assistant("call-a")] }, makeCtx());
   const saved = JSON.parse(await readFile(`${stateFile}.acp.json`, "utf8"));
@@ -582,7 +582,7 @@ test("omp preserves stable destination when migrating a colliding live ref", asy
   let persisted: ReturnType<typeof userMsg>[] = [];
   const makeCtx = () => ({ ...fakeCtx(persisted, stateFile), sessionManager: { getBranch: () => persisted, getSessionId: () => "test-session", getSessionFile: () => stateFile } });
   const first = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(first.api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(first.api);
   await first.handlers.get("context")![0]!({ type: "context", messages: [assistant("call-a")] }, makeCtx());
   const initial = JSON.parse(await readFile(`${stateFile}.acp.json`, "utf8"));
   const liveRef = initial.messageRefs.byRaw["live-0"];
@@ -593,7 +593,7 @@ test("omp preserves stable destination when migrating a colliding live ref", asy
   const { writeFile } = await import("node:fs/promises");
   await writeFile(`${stateFile}.acp.json`, JSON.stringify(initial), "utf8");
   const second = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(second.api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(second.api);
   persisted = [{ type: "message", id: "e-assistant", parentId: null, timestamp: "", message: assistant("call-a") }];
   await second.handlers.get("context")![0]!({ type: "context", messages: [assistant("call-a")] }, makeCtx());
   const saved = JSON.parse(await readFile(`${stateFile}.acp.json`, "utf8"));
@@ -605,7 +605,7 @@ test("omp preserves stable destination when migrating a colliding live ref", asy
 
 test("empty live context preserves refs created for an unpersisted message", async () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api);
   const stateFile = "/tmp/nonexistent-pai-acp-empty-live.session.json";
   await rm(`${stateFile}.acp.json`, { force: true });
   const ctx = fakeCtx([], stateFile);
@@ -616,7 +616,7 @@ test("empty live context preserves refs created for an unpersisted message", asy
 });
 test("omp keeps compression blocks active when provider context has an extra prefix", async (t) => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
   const dir = await mkdtemp(join(tmpdir(), "pai-acp-omp-provider-prefix-"));
   t.after(() => rm(dir, { recursive: true, force: true }));
   const stateFile = join(dir, "session.json");
@@ -667,7 +667,7 @@ test("omp keeps compression blocks active when provider context has an extra pre
 
 test("omp keeps compression active when persisted and provider tails diverge", async (t) => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as unknown as ExtensionAPI);
   const dir = await mkdtemp(join(tmpdir(), "pai-acp-omp-branch-divergence-"));
   t.after(() => rm(dir, { recursive: true, force: true }));
   const stateFile = join(dir, "session.json");
@@ -721,7 +721,7 @@ test("omp keeps compression active when persisted and provider tails diverge", a
 
 test("delegate:false omits the ACP_DELEGATE NOTIFICATIONS section from the system prompt", () => {
   const { api, handlers } = captureApi();
-  createAcpExtension({ delegate: false })(api as any);
+  createAcpExtension({ rollover: false, delegate: false })(api as any);
   const result = handlers.get("before_agent_start")![0]!({ systemPrompt: "" }, {});
   assert.ok(!result.systemPrompt.includes("ACP_DELEGATE NOTIFICATIONS"), "delegate section omitted when delegate:false");
   // Core ACP prompt is still present — only the delegate section is dropped.
@@ -745,7 +745,7 @@ test("modelContextLimit changes in .pi/acp.json are picked up on the next contex
 
   try {
     const { api, handlers } = captureApi();
-    createAcpExtension()(api as any);
+    createAcpExtension({ rollover: false })(api as any);
 
     function ctxWithCwd() {
       return { ...fakeCtx([], join(tmp, "state.json")), cwd: tmp };
@@ -810,7 +810,7 @@ test("headless (hasUI=false) context handler awaits the update check so an exiti
   try {
     const npm = pendingNpm();
     const { api, handlers } = captureApi();
-    createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+    createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
     const stateFile = "/tmp/nonexistent-pai-acp-headless-update.session.json";
     await rm(`${stateFile}.acp.json`, { force: true });
     const ctx = fakeCtx([userMsg("e1", "hi")], stateFile);
@@ -848,7 +848,7 @@ test("TUI (hasUI=true) context handler resolves without waiting for the update c
   try {
     const npm = pendingNpm();
     const { api, handlers } = captureApi();
-    createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+    createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
     const stateFile = "/tmp/nonexistent-pai-acp-tui-update.session.json";
     await rm(`${stateFile}.acp.json`, { force: true });
     const ctx = { ...fakeCtx([userMsg("e1", "hi")], stateFile), hasUI: true };
diff --git a/tests/prefix-stab.test.ts b/tests/prefix-stab.test.ts
index 2ef0c0a..01e6ba3 100644
--- a/tests/prefix-stab.test.ts
+++ b/tests/prefix-stab.test.ts
@@ -41,7 +41,7 @@ function fakeCtx(entries: any[], stateFile: string) {
 
 test("outbound provider view stays byte-stable across context rounds; compression rewrites it exactly once", async () => {
     const { api, handlers } = captureApi();
-    createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+    createAcpExtension({ modelContextLimit: 200_000, rollover: false })(api as any);
     const dir = await mkdtemp(join(tmpdir(), "pi-prefix-"));
     const stateFile = join(dir, "s.session.json");
 
diff --git a/tests/rollover.test.ts b/tests/rollover.test.ts
new file mode 100644
index 0000000..3b0c913
--- /dev/null
+++ b/tests/rollover.test.ts
@@ -0,0 +1,387 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { createAcpExtension } from "../src/index.js";
+
+// Batch rollover (#241): in-phase pending compress/absorb calls must keep the
+// outbound provider view byte-stable (append-only), and the rollover itself
+// must rewrite the view exactly once, after which it re-anchors.
+
+function captureApi() {
+  const handlers = new Map<string, Function[]>();
+  const tools: any[] = [];
+  const commands = new Map<string, any>();
+  const api = {
+    on: (event: string, handler: Function) => {
+      const list = handlers.get(event) ?? [];
+      list.push(handler);
+      handlers.set(event, list);
+    },
+    registerTool: (t: any) => { tools.push(t); },
+    registerCommand: (name: string, options: any) => { commands.set(name, options); },
+  };
+  return { api, handlers, tools, commands };
+}
+
+function entry(id: string, role: "user" | "assistant", text: string) {
+  return { type: "message", id, parentId: null, timestamp: "", message: { role, content: text, timestamp: 0 } };
+}
+
+function toolCallEntry(id: string, callId: string, name: string, args: unknown) {
+  return {
+    type: "message",
+    id,
+    parentId: null,
+    timestamp: "",
+    message: { role: "assistant", content: [{ type: "toolCall", id: callId, name, arguments: args }], timestamp: 0 },
+  };
+}
+
+function resultEntry(id: string, callId: string, name: string, text: string) {
+  return {
+    type: "message",
+    id,
+    parentId: null,
+    timestamp: "",
+    message: { role: "toolResult", toolCallId: callId, toolName: name, content: [{ type: "text", text }], isError: false, timestamp: 0 },
+  };
+}
+
+function fakeCtx(entries: any[], stateFile: string, window: number) {
+  return {
+    mode: "rpc",
+    hasUI: false,
+    ui: { notify: () => {}, setStatus: () => {}, setTitle: () => {}, setFooter: () => {}, setHeader: () => {}, note: () => {}, suggestForInput: () => {} },
+    model: { contextWindow: window, id: "test-model" },
+    sessionManager: {
+      getBranch: () => entries,
+      getEntries: () => entries,
+      getSessionId: () => "rollover-test-session",
+      getSessionFile: () => stateFile,
+    },
+  };
+}
+
+const FILLER = "the quick brown fox jumps over the lazy dog. ".repeat(30);
+
+function firstDiv(prev: any[], cur: any[]): number {
+  const p = prev.map((m) => JSON.stringify(m));
+  const c = cur.map((m) => JSON.stringify(m));
+  for (let k = 0; k < Math.min(p.length, c.length); k++) {
+    if (p[k] !== c[k]) return k;
+  }
+  return -1;
+}
+
+function divergences(views: any[][]): number[] {
+  const out: number[] = [];
+  for (let k = 1; k < views.length; k++) {
+    if (firstDiv(views[k - 1]!, views[k]!) !== -1) out.push(k);
+  }
+  return out;
+}
+
+function visibleIn(view: any[], needle: string): boolean {
+  return view.some((m) => JSON.stringify(m).includes(needle));
+}
+
+function refOfView(view: any[], needle: string): string {
+  for (const m of view) {
+    const json = JSON.stringify(m);
+    if (!json.includes(needle)) continue;
+    const match = json.match(/>(m\d{5})<\/acp>/);
+    if (match) return match[1]!;
+  }
+  throw new Error(`no tagged message containing ${needle}`);
+}
+
+async function withStoreDir(fn: (dir: string) => Promise<void>): Promise<void> {
+  const dir = await mkdtemp(join(tmpdir(), "acp-rollover-test-"));
+  try {
+    await fn(dir);
+  } finally {
+    await rm(dir, { recursive: true, force: true });
+  }
+}
+
+test("in-phase pending compress calls keep the view byte-stable (restore fix)", async () => {
+  await withStoreDir(async (dir) => {
+    const stateFile = join(dir, "session.jsonl");
+    const { api, handlers, tools } = captureApi();
+    createAcpExtension({ modelContextLimit: 200_000, rollover: true, autoUpdate: false })(api as any);
+    const entries: any[] = [];
+    const ctxOf = () => fakeCtx(entries, stateFile, 200_000);
+    const compress = tools.find((t) => t.name === "compress")!;
+    const ctx = ctxOf();
+    const runRound = async () => {
+      const res = await handlers.get("context")![0]!({
+        type: "context",
+        messages: entries.map((e) => e.message),
+      }, ctx);
+      return res.messages as any[];
+    };
+
+    const views: any[][] = [];
+    for (let t = 1; t <= 16; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+
+    const ranges = ["m00002..m00005", "m00008..m00011", "m00014..m00017"];
+    for (let i = 0; i < ranges.length; i++) {
+      const [startId, endId] = ranges[i]!.split("..");
+      const callId = `tc${i + 1}`;
+      const out = await compress.execute(callId, {
+         content: [{ startId: startId!, endId: endId!, summary: `Pending range ${i + 1} summary: filler turns ${Number(startId!.slice(1)) / 2 - 1}..${Number(endId!.slice(1)) / 2} — repeated fox sentences only, no decisions recorded.` }],
+      }, undefined, undefined, ctx) as { content: { text: string }[] };
+      const text = out.content[0]!.text;
+      assert.ok(!text.includes("FAILED"), `compress ${i + 1} failed: ${text}`);
+      assert.ok(text.includes("recorded for next rollover"), `expected pending panel, got: ${text}`);
+      entries.push(toolCallEntry(`c${i + 1}`, callId, "compress", { content: [{ startId: startId!, endId: endId!, summary: `Pending range ${i + 1}` }] }));
+      entries.push(resultEntry(`r${i + 1}`, callId, "compress", text));
+      views.push(await runRound());
+    }
+
+    // 16 base rounds + 3 post-compress rounds = 19 views. Every pair must be
+    // append-only: the 3rd pending call would hide the 1st mid-history
+    // (KEEP_LAST_ORPHANED=2) without the restore fix.
+    assert.equal(views.length, 19);
+    const divs = divergences(views);
+    assert.deepEqual(divs, [], `view diverged in-phase at round(s) ${divs.join(", ")}`);
+  });
+});
+
+test("rollover fires at threshold — one rewrite, then re-anchors", async () => {
+  await withStoreDir(async (dir) => {
+    const stateFile = join(dir, "session.jsonl");
+    const { api, handlers, tools } = captureApi();
+    createAcpExtension({ modelContextLimit: 12_000, rollover: true, autoUpdate: false })(api as any);
+    const entries: any[] = [];
+    const ctxOf = () => fakeCtx(entries, stateFile, 12_000);
+    const compress = tools.find((t) => t.name === "compress")!;
+    const ctx = ctxOf();
+    const runRound = async () => {
+      const res = await handlers.get("context")![0]!({
+        type: "context",
+        messages: entries.map((e) => e.message),
+      }, ctx);
+      return res.messages as any[];
+    };
+
+    // 11 turns (~7.3K tokens, 61% of 12K): past the 5000-token protected zone
+    // for the m00002..m00005 range, below the 70% rollover threshold.
+    const views: any[][] = [];
+    for (let t = 1; t <= 11; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+
+    const out = await compress.execute("tc1", {
+      content: [{ startId: "m00002", endId: "m00005", summary: "Early filler turns 1-3: repeated fox sentences, no decisions." }],
+    }, undefined, undefined, ctx) as { content: { text: string }[] };
+    const text = out.content[0]!.text;
+    assert.ok(text.includes("recorded for next rollover"), `expected pending panel, got: ${text}`);
+    entries.push(toolCallEntry("c1", "tc1", "compress", { content: [{ startId: "m00002", endId: "m00005", summary: "Early filler" }] }));
+    entries.push(resultEntry("r1", "tc1", "compress", text));
+    views.push(await runRound());
+
+    let rolloverAt = -1;
+    for (let t = 12; t <= 30 && rolloverAt === -1; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      const view = await runRound();
+      views.push(view);
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+      if (visibleIn(view, "▣ ACP rollover")) rolloverAt = views.length - 1;
+    }
+    assert.ok(rolloverAt > -1, "rollover never fired before 30 turns");
+    const report = views[rolloverAt]!.map((m) => JSON.stringify(m)).find((j) => j.includes("▣ ACP rollover"))!;
+    assert.match(report, /applied/);
+    assert.match(report, /1 compression/);
+
+    for (let t = 31; t <= 33; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+
+    const divs = divergences(views);
+    assert.deepEqual(divs, [rolloverAt, rolloverAt + 1], `expected the rollover round + its report-drop round, got ${divs.join(", ")}`);
+  });
+});
+
+test("absorb: pair visible until rollover, gone after the single rewrite", async () => {
+  await withStoreDir(async (dir) => {
+    const stateFile = join(dir, "session.jsonl");
+    const { api, handlers, tools } = captureApi();
+    createAcpExtension({ modelContextLimit: 12_000, rollover: { threshold: 0.6 }, autoUpdate: false })(api as any);
+    const entries: any[] = [];
+    const ctxOf = () => fakeCtx(entries, stateFile, 12_000);
+    const ctx = ctxOf();
+    await handlers.get("session_start")![0]!({}, ctx);
+    const absorb = tools.find((t) => t.name === "absorb")!;
+    assert.ok(absorb, "absorb tool not registered");
+    const runRound = async () => {
+      const res = await handlers.get("context")![0]!({
+        type: "context",
+        messages: entries.map((e) => e.message),
+      }, ctx);
+      return res.messages as any[];
+    };
+
+    const BASH_BODY = "ABSORB-ME-OUTPUT " + "line of verbose tool output ".repeat(120);
+    const views: any[][] = [];
+    for (let t = 1; t <= 3; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+    entries.push(toolCallEntry("cb", "cbash", "bash", { command: "ls -laR" }));
+    entries.push(resultEntry("rb", "cbash", "bash", BASH_BODY));
+    views.push(await runRound());
+    assert.ok(visibleIn(views[views.length - 1]!, "ABSORB-ME-OUTPUT"));
+
+    const ref = refOfView(views[views.length - 1]!, "ABSORB-ME-OUTPUT");
+    const out = await absorb.execute("tabsorb", { ref, summary: "bash ls -laR: 42 files, 3 dirs; nothing unusual." }, undefined, undefined, ctx) as { content: { text: string }[] };
+    const text = out.content[0]!.text;
+    assert.ok(text.includes("recorded for the next rollover"), `expected pending absorb text, got: ${text}`);
+    entries.push(toolCallEntry("ca", "tabsorb", "absorb", { ref, summary: "bash ls -laR: 42 files, 3 dirs; nothing unusual." }));
+    entries.push(resultEntry("ra", "tabsorb", "absorb", text));
+    views.push(await runRound());
+    assert.ok(visibleIn(views[views.length - 1]!, "ABSORB-ME-OUTPUT"), "absorbed pair must stay visible until rollover");
+
+    let rolloverAt = -1;
+    for (let t = 4; t <= 30 && rolloverAt === -1; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      const view = await runRound();
+      views.push(view);
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+      if (visibleIn(view, "▣ ACP rollover")) rolloverAt = views.length - 1;
+    }
+    assert.ok(rolloverAt > -1, "rollover never fired before 30 turns");
+    assert.ok(!visibleIn(views[rolloverAt]!, "ABSORB-ME-OUTPUT"), "absorbed pair must be gone after rollover");
+    assert.ok(visibleIn(views[rolloverAt]!, "42 files, 3 dirs"), "absorb summary must remain visible");
+
+    for (let t = 31; t <= 33; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+    const divs = divergences(views);
+    assert.deepEqual(divs, [rolloverAt, rolloverAt + 1], `expected the rollover round + its report-drop round, got ${divs.join(", ")}`);
+  });
+});
+
+test("decompress/search results append to tail (prefix byte-identical)", async () => {
+  await withStoreDir(async (dir) => {
+    const stateFile = join(dir, "session.jsonl");
+    const { api, handlers, tools } = captureApi();
+    createAcpExtension({ modelContextLimit: 12_000, rollover: true, autoUpdate: false })(api as any);
+    const entries: any[] = [];
+    const ctxOf = () => fakeCtx(entries, stateFile, 12_000);
+    const ctx = ctxOf();
+    const compress = tools.find((t) => t.name === "compress")!;
+    const search = tools.find((t) => t.name === "search_context")!;
+    const decompress = tools.find((t) => t.name === "decompress")!;
+    const runRound = async () => {
+      const res = await handlers.get("context")![0]!({
+        type: "context",
+        messages: entries.map((e) => e.message),
+      }, ctx);
+      return res.messages as any[];
+    };
+
+    const views: any[][] = [];
+    for (let t = 1; t <= 11; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+    const out = await compress.execute("tc1", {
+      content: [{ startId: "m00002", endId: "m00005", summary: "Filler History: early fox-sentence turns, no decisions.", topic: "Filler History" }],
+    }, undefined, undefined, ctx) as { content: { text: string }[] };
+    entries.push(toolCallEntry("c1", "tc1", "compress", { content: [{ startId: "m00002", endId: "m00005", summary: "Filler History" }] }));
+    entries.push(resultEntry("r1", "tc1", "compress", out.content[0]!.text));
+    views.push(await runRound());
+
+    let rolloverAt = -1;
+    for (let t = 12; t <= 30 && rolloverAt === -1; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      const view = await runRound();
+      views.push(view);
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+      if (visibleIn(view, "▣ ACP rollover")) rolloverAt = views.length - 1;
+    }
+    assert.ok(rolloverAt > -1, "rollover never fired");
+
+    const base = views.length;
+    const sOut = await search.execute("tsearch", { query: "Filler History" }, undefined, undefined, ctx) as { content: { text: string }[] };
+    entries.push(toolCallEntry("cs", "tsearch", "search_context", { query: "Filler History" }));
+    entries.push(resultEntry("rs", "tsearch", "search_context", sOut.content[0]!.text));
+    views.push(await runRound());
+
+    const outPath = join(dir, "b1-decompressed.txt");
+    const dOut = await decompress.execute("tdecomp", { blockId: "b1", toFile: outPath }, undefined, undefined, ctx) as { content: { text: string }[] };
+    assert.ok(!dOut.content[0]!.text.includes("FAILED"), `decompress failed: ${dOut.content[0]!.text}`);
+    entries.push(toolCallEntry("cd", "tdecomp", "decompress", { blockId: "b1", toFile: outPath }));
+    entries.push(resultEntry("rd", "tdecomp", "decompress", dOut.content[0]!.text));
+    views.push(await runRound());
+    const fileContent = await readFile(outPath, "utf8");
+    assert.ok(fileContent.includes("user turn 2:"), "decompressed file should contain the original messages");
+
+    for (let k = base + 1; k < views.length; k++) {
+      assert.equal(firstDiv(views[k - 1]!, views[k]!), -1, `retrieval round ${k} must be a pure tail append`);
+    }
+  });
+});
+
+test("manual /acp-rollover applies pending work immediately", async () => {
+  await withStoreDir(async (dir) => {
+    const stateFile = join(dir, "session.jsonl");
+    const { api, handlers, tools, commands } = captureApi();
+    createAcpExtension({ modelContextLimit: 200_000, rollover: true, autoUpdate: false })(api as any);
+    const entries: any[] = [];
+    const ctxOf = () => fakeCtx(entries, stateFile, 200_000);
+    const ctx = ctxOf();
+    const compress = tools.find((t) => t.name === "compress")!;
+    const runRound = async () => {
+      const res = await handlers.get("context")![0]!({
+        type: "context",
+        messages: entries.map((e) => e.message),
+      }, ctx);
+      return res.messages as any[];
+    };
+
+    const views: any[][] = [];
+    for (let t = 1; t <= 12; t++) {
+      entries.push(entry(`u${t}`, "user", `user turn ${t}: ${FILLER}`));
+      views.push(await runRound());
+      entries.push(entry(`a${t}`, "assistant", `assistant reply ${t}: ${FILLER}`));
+    }
+    const out = await compress.execute("tc1", {
+      content: [{ startId: "m00002", endId: "m00005", summary: "Manual rollover target: early filler turns with repeated fox sentences, no decisions recorded." }],
+    }, undefined, undefined, ctx) as { content: { text: string }[] };
+    assert.ok(out.content[0]!.text.includes("recorded for next rollover"));
+    entries.push(toolCallEntry("c1", "tc1", "compress", { content: [{ startId: "m00002", endId: "m00005", summary: "Manual rollover target" }] }));
+    entries.push(resultEntry("r1", "tc1", "compress", out.content[0]!.text));
+    views.push(await runRound());
+    assert.equal(firstDiv(views[views.length - 2]!, views[views.length - 1]!), -1, "no rollover below threshold");
+
+    const notify: string[] = [];
+    (ctx as any).ui.notify = (t: string) => notify.push(t);
+    const cmd = commands.get("acp-rollover")!;
+    assert.ok(cmd, "acp-rollover command not registered");
+    await cmd.handler("", ctx);
+    assert.ok(notify.length > 0, "command should notify");
+    assert.match(notify.join("\n"), /▣ ACP rollover/);
+    assert.match(notify.join("\n"), /applied/);
+
+    views.push(await runRound());
+    const last = views[views.length - 1]!;
+    assert.ok(!visibleIn(last, "assistant reply 1: "), "compressed range must be gone after manual rollover");
+    assert.ok(visibleIn(last, "user turn 12: "), "recent history must survive");
+  });
+});
diff --git a/tests/sent-view-arbitration.test.ts b/tests/sent-view-arbitration.test.ts
index ce7805e..5404767 100644
--- a/tests/sent-view-arbitration.test.ts
+++ b/tests/sent-view-arbitration.test.ts
@@ -63,7 +63,7 @@ const nudgeCount = (r: any) =>
 test("context transform ignores session-tree accounting (180K window, 366K tree)", async () => {
   await rm(`${STATE_FILE}.365606.acp.json`, { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 180_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 180_000 })(api as any);
 
   // Host reports a tree that outgrew the window (switched down from 1M).
   // The live stream the model actually sees is ~36K — 20% of the window.
@@ -80,7 +80,7 @@ test("context transform ignores session-tree accounting (180K window, 366K tree)
 test("context transform DOES go emergency when the sent view itself overflows", async () => {
   await rm(`${STATE_FILE}.1000.acp.json`, { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 180_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 180_000 })(api as any);
 
   // Host reports a small tree; irrelevant now. The stream itself is 60 ×
   // ~4.5K ≈ 270K tokens → 150% of the 180K window.
diff --git a/tests/t3-rewrite-guard.test.ts b/tests/t3-rewrite-guard.test.ts
index 8eaf9cd..209595f 100644
--- a/tests/t3-rewrite-guard.test.ts
+++ b/tests/t3-rewrite-guard.test.ts
@@ -60,7 +60,7 @@ function fakeCtx(entries: Array<ReturnType<typeof userMsg>>, stateFile: string)
 async function setup(stateFile: string) {
   await rm(stateFile + ".acp.json", { force: true });
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as never);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as never);
   const entries = Array.from({ length: 12 }, (_, i) => userMsg(i));
   const ctx = fakeCtx(entries, stateFile);
   await handlers.get("context")![0]!({ type: "context", messages: [] }, ctx);
diff --git a/tests/warnings-passthrough.test.ts b/tests/warnings-passthrough.test.ts
index 8b5c2f8..eb82c17 100644
--- a/tests/warnings-passthrough.test.ts
+++ b/tests/warnings-passthrough.test.ts
@@ -45,7 +45,7 @@ function fakeCtx(entries: any[]) {
 
 async function setup(entries: any[]) {
   const { api, handlers } = captureApi();
-  createAcpExtension({ modelContextLimit: 200_000 })(api as any);
+  createAcpExtension({ rollover: false, modelContextLimit: 200_000 })(api as any);
   const ctx = fakeCtx(entries);
   await handlers.get("context")![0]!({ type: "context", messages: [] }, ctx);
   const compressTool = api.tools.find((t: any) => t.name === "compress")!;