我建议不要“照搬 Claude Code 的四个函数”，而是吸收它的核心思想：cheap-first、stage pipeline、结构化结果、昂贵 LLM summary 最后执行、overflow 时走更激进 reactive path。在 lovia 里最优雅的实现应该围绕 TranscriptEntry，而不是 provider-shaped Message。

核心设计 把 SummarizingContextPolicy 拆成一个可组合 pipeline：

ContextPolicy
  └── CompactingContextPolicy
        ├── stages: list[ContextStage]
        ├── summarizer: Summarizer
        ├── archive: CompactionArchive | None
        └── apply() / apply_reactive()

每个 stage 都是便宜、可测试、无 LLM 调用的 transcript rewrite：

class ContextStage(Protocol):
    name: str

    async def apply(
        self,
        entries: list[TranscriptEntry],
        *,
        ctx: PolicyContext,
    ) -> "StageResult": ...

结构化返回：

@dataclass
class StageResult:
    entries: list[TranscriptEntry]
    changed: bool = False
    tokens_freed_estimate: int | None = None
    note: str | None = None


@dataclass
class ContextPolicyResult:
    entries: list[TranscriptEntry]
    changed: bool
    summary: str | None = None
    reason: Literal[
        "tool_result_budget",
        "middle_snip",
        "tool_result_retention",
        "auto_summary",
        "reactive_summary",
        "manual_summary",
    ] | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

为了兼容现有用户自定义 policy，可以短期让 runner 接受两种返回：

result = await policy.apply(...)
if isinstance(result, list):
    entries_after = result
    summary = extract_compaction_summary(result)  # legacy only
else:
    entries_after = result.entries
    summary = result.summary

长期再把 list[TranscriptEntry] 返回标为 deprecated。

推荐 pipeline 顺序

1. ToolResultBudgetStage 对应 Claude Code 的 L3，但在 lovia 里应该先执行。它处理“单个或最近一批 tool result 太大”的问题。

2. MiddleSnipStage 对应 L1。超过 entry 数量或 token 数时，保留开头少量系统/初始上下文和最近窗口，中间替换为 placeholder。

3. ToolResultRetentionStage 对应 L2。把较旧 ToolResultEntry.output 替换为短 placeholder。

4. AutoSummaryStage 对应 L4。前三层仍超过阈值时，才调用 summarizer。

5. ReactiveSummaryStage provider 抛 ContextOverflowError 时触发，更激进，保留更短 tail，最多 retry 一次或配置次数。

顺序非常重要：budget 必须在 retention 前。否则旧的大输出先被 placeholder 替换，就没机会保存完整内容。

关键接口命名

我会把现在的参数改成更直观的名字，同时保留旧参数兼容一版：

CompactingContextPolicy(
    context_window_tokens: int | None = None,
    trigger_ratio: float = 0.8,
    max_entries: int | None = 80,
    keep_initial_entries: int = 3,
    keep_recent_entries: int = 40,
    reactive_keep_recent_entries: int = 8,
    keep_recent_tool_results: int = 3,
    tool_result_budget_chars: int = 200_000,
    large_tool_result_chars: int = 20_000,
    tool_result_preview_chars: int = 2_000,
    summarizer: Summarizer | None = None,
    archive: CompactionArchive | None = None,
    max_summary_failures: int = 3,
)

旧名字映射：

max_tokens -> context_window_tokens
compact_at_ratio -> trigger_ratio
keep_recent_messages -> keep_recent_entries
reactive_keep_recent_messages -> reactive_keep_recent_entries

生产可用的 archive / store 设计

不要让 core 直接依赖 .task_outputs/ 或 sandbox。core 应该定义协议：

class CompactionArchive(Protocol):
    async def save_transcript(
        self,
        entries: list[TranscriptEntry],
        *,
        ctx: PolicyContext,
        reason: str,
    ) -> "ArchiveRef": ...

    async def save_tool_result(
        self,
        output: str,
        *,
        call_id: str,
        ctx: PolicyContext,
    ) -> "ArchiveRef": ...

默认可以是 None，不做落盘。提供一个可选实现：

FileCompactionArchive(root=".lovia")

输出类似：

.lovia/transcripts/{session_id}/{run_id}-{timestamp}.jsonl
.lovia/tool-results/{session_id}/{call_id}.txt

ToolResultBudgetStage 替换内容时写成：

<persisted-tool-result>
Full output archived at: .lovia/tool-results/s1/call_123.txt
Preview:
...
</persisted-tool-result>

这样 model 知道内容没丢；如果用户给了文件读取工具，它还能重新读取。没有 archive 时，则退化为 preview-only，不承诺可恢复。

事件模型

建议删除 ArchiveEvent，统一使用 events.ContextCompacted，但补字段：

@dataclass
class ContextCompacted(ContextEvent):
    session_id: str | None
    entries_before: list[TranscriptEntry]
    entries_after: list[TranscriptEntry]
    summary: str | None = None
    reactive: bool = False
    reason: str | None = None
    archive_ref: ArchiveRef | None = None
    tokens_before_estimate: int | None = None
    tokens_after_estimate: int | None = None

archive callback 如果还想支持，可以改为：

ArchiveCallback = Callable[[ContextCompacted], Awaitable[None]]

这样不会维护两份一模一样的数据结构。

summary entry 不要靠字符串解析

现在 runner 通过 marker 反解析 summary，这个设计应该废掉。summary 应该来自 ContextPolicyResult.summary。

summary entry 仍然可以存在，但只是给模型看的 transcript 内容：

def make_summary_entry(summary: str, *, reactive: bool) -> InputEntry:
    label = "Reactive context summary" if reactive else "Context summary"
    return InputEntry(
        role="user",
        content=f"[{label}]\n\n{summary}",
    )

extract_compaction_summary() 只作为 legacy helper 保留。

token 判断

这里不要用 usage.total_tokens 判断 context window。应该明确区分：

• context window 判断：用当前 prompt tokens

• 成本/预算判断：用 total tokens

推荐：

@dataclass
class PolicyContext:
    provider: Any
    model: str | None
    last_input_tokens: int | None = None
    session_id: str | None = None

然后：

def current_prompt_tokens(entries, ctx):
    estimate = estimate_tokens(ctx.provider, entries)
    if ctx.last_input_tokens is None:
        return estimate
    return max(estimate, ctx.last_input_tokens)

把现在那段长注释变成这个 helper 名字，代码就清楚了。

三种方案对比

方案 A：小修现有 SummarizingContextPolicy
优点：改动小，兼容性最好。
缺点：仍然是单类膨胀，stage 不可插拔，后续加 budget/archive 会越来越乱。
我不推荐作为最终方案，只适合热修。

方案 B：引入 pipeline，但保留现有 policy 名称
优点：用户 API 变化小，内部变清晰，生产能力足够。
缺点：SummarizingContextPolicy 这个名字会变得不完全准确，因为它还做 snip/budget/retention。
这是我最推荐的迁移方案。

方案 C：新增 CompactingContextPolicy，旧 SummarizingContextPolicy 变兼容 wrapper
优点：命名最准确，架构最干净，扩展最好。
缺点：新增公共 API，文档和 README 要同步更新。
如果准备做一次比较正式的 context policy 重构，我最推荐这个。

最终推荐

做方案 C：

• 新增 CompactingContextPolicy

• 新增 stage 协议和默认 stages

• SummarizingContextPolicy 保留，但内部委托给 CompactingContextPolicy

• ArchiveEvent deprecated，统一到 ContextCompacted

• runner 支持 ContextPolicyResult

• summary 不再通过 marker 解析

• 文件归档通过 CompactionArchive 协议注入，core 不直接依赖 sandbox/web

这样既接近 Claude Code 的生产思路，又符合 lovia 的架构哲学：核心小、协议清楚、扩展靠组合，不靠一个越来越重的类硬撑。