历史要变短，先决定哪里可以改
用户要先查旧日志，再按项目规范检查 main.py 与 expected.txt，运行前问他，获准后核对实际输出。思想篇已经区分了快照、活跃历史和摘要。现在把整理放在一个明确时刻：一组工具结果全部写入以后，下一次主模型请求发出以前。
正式代码在公开仓库 examples/s08-context-compact。为了让小例子也能触发容量压力，仓库准备了四份可重复的教学旧日志；它们不代表当前程序已经运行。旧内容逐渐退出活跃视图，当前文件检查、许可与真实进程仍沿前面的实现执行。
三个主要函数按顺序负责：split_history找完整交接边界，trim_results移出旧长正文，prepare_context在归档和预算约束下生成候选。下面保留完整函数正文；设断点后点击“调试”，图跟随同一个执行游标，切换输入会重置断点。helper来自同一公开提交，准备和调用入口可展开查看。
split_history：一个切口不能拆散一次交接
原循环在首次请求前只有一条文本用户问题；执行工具以后，则追加完整 assistant 工具请求和 user 工具结果。split_history只处理这个时刻的历史，不尝试兼容所有聊天、多模态或最终答复形式。
它先确认剩余消息能两两组成交接，再让 checked_exchange 逐组检查角色与ID。每个 tool_use 必须有一个同ID的结果；同一组多个调用也必须全部对应，不能少一项、重复一项或拿别的结果来代替。
真实执行轨迹回放；纯内存消息处理，输入与同源helper可展开查看。
Pys08/core/runtime.py · split_history
未运行
def split_history(messages, keep_recent=1):
"""在工具请求已经收齐结果时整理；最近完整交接作为不可拆的尾部。"""
if type(keep_recent) is not int or not 1 <= keep_recent <= 4:
raise ValueError('至少保留1组，最多4组最近交接。')
if type(messages) is not list or not messages or type(messages[0]) is not dict or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('历史必须从文本用户问题开始。')
if (len(messages) - 1) % 2:
raise ValueError('工具结果尚未收齐，不能压缩。')
groups = []
for index in range(1, len(messages), 2):
checked_exchange(messages[index], messages[index + 1])
groups.append(deepcopy(messages[index:index + 2]))
cut = max(0, len(groups) - keep_recent)
return {'question': deepcopy(messages[0]), 'older': groups[:cut], 'recent': groups[cut:]}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
from pathlib import Path
from uuid import uuid4
import tempfile
scenario='normal'
question="先检查，再获准运行。"
def pair(number,text,error=False):
return [{"role":"assistant","content":[{"type":"tool_use","id":"call_"+str(number),"name":"read_file","input":{"path":"notes.txt"}}]}, {"role":"user","content":[{"type":"tool_result","tool_use_id":"call_"+str(number),"content":text,"is_error":error}]}]
messages=[{"role":"user","content":question}]+pair(1,"old"*400)+pair(2,"近期证据。")
if scenario=="pending":messages.pop()
if scenario=="mismatch":messages[-1]["content"][0]["tool_use_id"]="wrong"
if scenario=="multi":
messages[1]["content"].append({"type":"tool_use","id":"second","name":"read_file","input":{"path":"other.txt"}})
messages[2]["content"].append({"type":"tool_result","tool_use_id":"second","content":"另一条结果","is_error":False})
if scenario=="error":messages[2]["content"][0]["is_error"]=True
if scenario=="recent":messages[-1]["content"][0]["content"]="R"*1600
if scenario=="fit":messages=[{"role":"user","content":question}]+pair(1,"短结果。")
if scenario=="trim":messages=[{"role":"user","content":question}]+pair(1,"A"*1200)+pair(2,"B"*1200)+pair(3,"近期证据。")
if scenario in ("summary","bad_summary","archive_error","exhausted"):
messages=[{"role":"user","content":question}]
for i in range(4):messages+=pair(i,"旧"*180)
messages+=pair(4,"近期证据。")
if scenario=="recent_large":messages=[{"role":"user","content":question}]+pair(1,"R"*2400)
before=deepcopy(messages)
state={"archives":[],"summaries":0,"max_summaries":0 if scenario=="exhausted" else 2}
_temp=tempfile.TemporaryDirectory(prefix="lesson8-")
def checked_exchange(request, response):
if type(request) is not dict or type(response) is not dict:
raise ValueError('消息必须是对象。')
if request.get('role') != 'assistant' or response.get('role') != 'user':
raise ValueError('历史必须由完整请求与结果交替组成。')
blocks, results = request.get('content'), response.get('content')
if type(blocks) is not list or type(results) is not list:
raise ValueError('本课只整理工具交接历史。')
if any(type(b) is not dict or b.get('type') not in ('text','tool_use') for b in blocks):
raise ValueError('请求块不受支持。')
ids = [b.get('id') for b in blocks if b['type'] == 'tool_use']
if not ids or any(type(i) is not str or not i for i in ids) or len(set(ids)) != len(ids):
raise ValueError('工具调用ID必须非空且唯一。')
if any(type(b) is not dict or b.get('type') != 'tool_result' or type(b.get('content')) is not str or type(b.get('is_error')) is not bool for b in results):
raise ValueError('结果格式不受支持。')
found = [b.get('tool_use_id') for b in results]
if any(type(i) is not str for i in found) or len(found) != len(ids) or set(found) != set(ids):
raise ValueError('每个工具请求必须有且只有一个同ID结果。')
def trim_results(messages, receipt, keep_recent=1, max_chars=600):
"""调用者已保存快照，才可移出旧正文；角色、ID和错误标记不变。"""
if type(receipt) is not str or not receipt or type(max_chars) is not int or max_chars < 1:
raise ValueError('需要归档收据和正的正文上限。')
parts = split_history(messages, keep_recent)
candidate = deepcopy(messages)
for index in range(2, 1 + 2 * len(parts['older']), 2):
for slot, block in enumerate(candidate[index]['content']):
if len(block['content']) > max_chars:
block['content'] = f'[旧正文已移出；宿主快照 {receipt}，消息{index}/块{slot}，ID={block["tool_use_id"]}。此标记不是执行结果。]'
return candidate
def history_bytes(messages):
return len(json.dumps(messages, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
def archive_writer(directory):
directory = Path(directory)
def save(messages):
payload = ''.join(json.dumps(m, ensure_ascii=False) + '\n' for m in messages)
if len(payload.encode('utf-8')) > 1_000_000:
raise ValueError('快照超过本例1MB上限。')
directory.mkdir(parents=True, exist_ok=True)
target = directory / ('context-' + uuid4().hex[:12] + '.jsonl')
with target.open('x', encoding='utf-8') as stream:
stream.write(payload)
return target.name
return save
archive=archive_writer(Path(_temp.name))
if scenario=="archive_error":
def archive(_):raise OSError("教学归档失败")
def summarize(prefix):
return "" if scenario=="bad_summary" else "旧正文已归档；仍须检查当前文件并取得运行许可。"
# 调用入口
try:
observed=split_history(messages,keep_recent=1)
print("已生成候选；原历史保持不变。")
except (ValueError,OSError) as error:
observed={"error":str(error)}
print(str(error))
assert messages==before
- 核对每组工具请求与结果的角色和ID。
- 只在完整组之间切开旧组与近期组。
- 返回分区副本，不改变原历史。
选择“缺少结果”，会在真正分组前失败；“ID不符”则到检查这一组时失败。“一组多调用”保留整组，并没有为了方便切片把它再拆成两段。
cut计算的是完整组的位置。默认保留最近1组，允许1到4组，旧部分和近期部分都来自深副本。函数没有删除正文，也没有调用模型；它先把“哪里能改、哪里暂时保留”变成一个可检查的结构。
本课不把 answered 后追加的最终 assistant 答复再送进这个分组器。后续若支持用户续问或多模态内容，应先扩展历史语法和边界验证，不能沿用当前“两条一组”的假设处理所有会话。
trim_results：只动正文，不改动作的身份
移出正文以前，调用者必须已经保存压缩前快照，并拿到 receipt。函数自身只生成消息副本。它遍历旧组的 tool_result，超过600字符的正文换成收据与消息/块位置；近期组不进入这个循环。
这里的600是单个旧正文的字符阈值，和后面的整份消息字节预算不是一回事。保留短反馈，是因为“用户拒绝”这类结果本身已经很短，没有必要为了压缩又制造更长的占位符。
真实执行轨迹回放；先在Worker临时虚拟文件保存快照，再处理内存副本。
Pys08/core/runtime.py · trim_results
未运行
def trim_results(messages, receipt, keep_recent=1, max_chars=600):
"""调用者已保存快照，才可移出旧正文；角色、ID和错误标记不变。"""
if type(receipt) is not str or not receipt or type(max_chars) is not int or max_chars < 1:
raise ValueError('需要归档收据和正的正文上限。')
parts = split_history(messages, keep_recent)
candidate = deepcopy(messages)
for index in range(2, 1 + 2 * len(parts['older']), 2):
for slot, block in enumerate(candidate[index]['content']):
if len(block['content']) > max_chars:
block['content'] = f'[旧正文已移出；宿主快照 {receipt}，消息{index}/块{slot}，ID={block["tool_use_id"]}。此标记不是执行结果。]'
return candidate输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
from pathlib import Path
from uuid import uuid4
import tempfile
scenario='old'
question="先检查，再获准运行。"
def pair(number,text,error=False):
return [{"role":"assistant","content":[{"type":"tool_use","id":"call_"+str(number),"name":"read_file","input":{"path":"notes.txt"}}]}, {"role":"user","content":[{"type":"tool_result","tool_use_id":"call_"+str(number),"content":text,"is_error":error}]}]
messages=[{"role":"user","content":question}]+pair(1,"old"*400)+pair(2,"近期证据。")
if scenario=="pending":messages.pop()
if scenario=="mismatch":messages[-1]["content"][0]["tool_use_id"]="wrong"
if scenario=="multi":
messages[1]["content"].append({"type":"tool_use","id":"second","name":"read_file","input":{"path":"other.txt"}})
messages[2]["content"].append({"type":"tool_result","tool_use_id":"second","content":"另一条结果","is_error":False})
if scenario=="error":messages[2]["content"][0]["is_error"]=True
if scenario=="recent":messages[-1]["content"][0]["content"]="R"*1600
if scenario=="fit":messages=[{"role":"user","content":question}]+pair(1,"短结果。")
if scenario=="trim":messages=[{"role":"user","content":question}]+pair(1,"A"*1200)+pair(2,"B"*1200)+pair(3,"近期证据。")
if scenario in ("summary","bad_summary","archive_error","exhausted"):
messages=[{"role":"user","content":question}]
for i in range(4):messages+=pair(i,"旧"*180)
messages+=pair(4,"近期证据。")
if scenario=="recent_large":messages=[{"role":"user","content":question}]+pair(1,"R"*2400)
before=deepcopy(messages)
state={"archives":[],"summaries":0,"max_summaries":0 if scenario=="exhausted" else 2}
_temp=tempfile.TemporaryDirectory(prefix="lesson8-")
def checked_exchange(request, response):
if type(request) is not dict or type(response) is not dict:
raise ValueError('消息必须是对象。')
if request.get('role') != 'assistant' or response.get('role') != 'user':
raise ValueError('历史必须由完整请求与结果交替组成。')
blocks, results = request.get('content'), response.get('content')
if type(blocks) is not list or type(results) is not list:
raise ValueError('本课只整理工具交接历史。')
if any(type(b) is not dict or b.get('type') not in ('text','tool_use') for b in blocks):
raise ValueError('请求块不受支持。')
ids = [b.get('id') for b in blocks if b['type'] == 'tool_use']
if not ids or any(type(i) is not str or not i for i in ids) or len(set(ids)) != len(ids):
raise ValueError('工具调用ID必须非空且唯一。')
if any(type(b) is not dict or b.get('type') != 'tool_result' or type(b.get('content')) is not str or type(b.get('is_error')) is not bool for b in results):
raise ValueError('结果格式不受支持。')
found = [b.get('tool_use_id') for b in results]
if any(type(i) is not str for i in found) or len(found) != len(ids) or set(found) != set(ids):
raise ValueError('每个工具请求必须有且只有一个同ID结果。')
def split_history(messages, keep_recent=1):
"""在工具请求已经收齐结果时整理；最近完整交接作为不可拆的尾部。"""
if type(keep_recent) is not int or not 1 <= keep_recent <= 4:
raise ValueError('至少保留1组，最多4组最近交接。')
if type(messages) is not list or not messages or type(messages[0]) is not dict or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('历史必须从文本用户问题开始。')
if (len(messages) - 1) % 2:
raise ValueError('工具结果尚未收齐，不能压缩。')
groups = []
for index in range(1, len(messages), 2):
checked_exchange(messages[index], messages[index + 1])
groups.append(deepcopy(messages[index:index + 2]))
cut = max(0, len(groups) - keep_recent)
return {'question': deepcopy(messages[0]), 'older': groups[:cut], 'recent': groups[cut:]}
def history_bytes(messages):
return len(json.dumps(messages, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
def archive_writer(directory):
directory = Path(directory)
def save(messages):
payload = ''.join(json.dumps(m, ensure_ascii=False) + '\n' for m in messages)
if len(payload.encode('utf-8')) > 1_000_000:
raise ValueError('快照超过本例1MB上限。')
directory.mkdir(parents=True, exist_ok=True)
target = directory / ('context-' + uuid4().hex[:12] + '.jsonl')
with target.open('x', encoding='utf-8') as stream:
stream.write(payload)
return target.name
return save
archive=archive_writer(Path(_temp.name))
if scenario=="archive_error":
def archive(_):raise OSError("教学归档失败")
def summarize(prefix):
return "" if scenario=="bad_summary" else "旧正文已归档；仍须检查当前文件并取得运行许可。"
receipt=archive(messages)
# 调用入口
try:
observed=trim_results(messages,receipt,keep_recent=1)
print("已生成候选；原历史保持不变。")
except (ValueError,OSError) as error:
observed={"error":str(error)}
print(str(error))
assert messages==before
- 先由调用者保存快照并取得收据。
- 只替换旧组的大正文，角色/ID/错误标记保留。
- 近期整组不参与替换，返回候选副本。
在 block['content'] = ... 停住时，当前块仍是旧正文；下一步才变成归档引用。选择“保留错误标记”，可以看到替换只触及 content，原 tool_use_id 与 is_error=True 仍然在。assistant 请求没有被改写。
“近期大结果”则故意让最新正文很长。这个函数仍不碰它，因为近期整组是边界约定。后面的总预算复查如果放不下，会明确停止，而不是偷偷把这份当前证据也剪掉。
浏览器输入会先把原始消息写入 Worker 的临时虚拟文件，满足归档契约；正式 CLI 写工作目录下的 .context。引用让宿主能找到快照和位置，本例没有注册模型可调用的归档检索工具。需要旧细节时应查快照或补读源文件，不能把占位符理解成完整结果，更不能自动重跑有副作用的命令。
prepare_context：每条成功返回都有一个条件
这个函数接收当前消息、宿主持有的原始问题、整理状态、归档接口和摘要接口。budget只计 messages JSON的UTF-8字节；默认3200用来演示压力，不代表真实模型窗口。keep_recent决定保留多少完整交接。
先看“无需整理”：格式检查通过且未超预算，只返回深副本，没有归档和摘要。超预算以后，必须先调用 archive，得到有效收据，才进入 trim_results。归档写入失败时，后面没有任何移出正文的动作。
真实执行轨迹回放；归档写入Worker临时虚拟文件，摘要接口为公开固定文本替身。
Pys08/core/runtime.py · prepare_context
未运行
def prepare_context(messages, question, state, archive, summarize, budget=3200, keep_recent=1):
"""先生成候选，不改输入；归档/摘要/复查任一步失败都不给循环新历史。"""
if type(budget) is not int or budget < 400 or type(question) is not str or not question:
raise ValueError('需要至少400字节预算和原始问题。')
parts = split_history(messages, keep_recent)
if history_bytes(messages) <= budget:
return deepcopy(messages)
receipt = archive(deepcopy(messages))
if type(receipt) is not str or not receipt:
raise ValueError('归档没有返回有效收据。')
state['archives'].append(receipt)
candidate = trim_results(messages, receipt, keep_recent)
if history_bytes(candidate) <= budget:
return candidate
parts = split_history(candidate, keep_recent)
if not parts['older']:
raise ValueError('最近交接本身超限；缩小工具输出，不能拆开它。')
if state['summaries'] >= state['max_summaries']:
raise ValueError('摘要预算用完，交回宿主。')
state['summaries'] += 1
prefix = [parts['question']] + [message for pair in parts['older'] for message in pair]
summary = summarize(deepcopy(prefix))
if type(summary) is not str or not summary.strip() or len(summary) > 1000:
raise ValueError('摘要必须是1到1000字符的非空文本。')
content = question + '\n\n[历史摘要，仅供参考]\n' + summary.strip()
content += '\n[压缩前快照：' + receipt + '；旧规范需重新加载；旧允许不授权本次运行。]'
candidate = [{'role': 'user', 'content': content}] + [message for pair in parts['recent'] for message in pair]
split_history(candidate, keep_recent)
if history_bytes(candidate) > budget or history_bytes(candidate) >= history_bytes(messages):
raise ValueError('候选历史仍超限或没有变小，不提交。')
return candidate输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
from pathlib import Path
from uuid import uuid4
import tempfile
scenario='fit'
question="先检查，再获准运行。"
def pair(number,text,error=False):
return [{"role":"assistant","content":[{"type":"tool_use","id":"call_"+str(number),"name":"read_file","input":{"path":"notes.txt"}}]}, {"role":"user","content":[{"type":"tool_result","tool_use_id":"call_"+str(number),"content":text,"is_error":error}]}]
messages=[{"role":"user","content":question}]+pair(1,"old"*400)+pair(2,"近期证据。")
if scenario=="pending":messages.pop()
if scenario=="mismatch":messages[-1]["content"][0]["tool_use_id"]="wrong"
if scenario=="multi":
messages[1]["content"].append({"type":"tool_use","id":"second","name":"read_file","input":{"path":"other.txt"}})
messages[2]["content"].append({"type":"tool_result","tool_use_id":"second","content":"另一条结果","is_error":False})
if scenario=="error":messages[2]["content"][0]["is_error"]=True
if scenario=="recent":messages[-1]["content"][0]["content"]="R"*1600
if scenario=="fit":messages=[{"role":"user","content":question}]+pair(1,"短结果。")
if scenario=="trim":messages=[{"role":"user","content":question}]+pair(1,"A"*1200)+pair(2,"B"*1200)+pair(3,"近期证据。")
if scenario in ("summary","bad_summary","archive_error","exhausted"):
messages=[{"role":"user","content":question}]
for i in range(4):messages+=pair(i,"旧"*180)
messages+=pair(4,"近期证据。")
if scenario=="recent_large":messages=[{"role":"user","content":question}]+pair(1,"R"*2400)
before=deepcopy(messages)
state={"archives":[],"summaries":0,"max_summaries":0 if scenario=="exhausted" else 2}
_temp=tempfile.TemporaryDirectory(prefix="lesson8-")
def checked_exchange(request, response):
if type(request) is not dict or type(response) is not dict:
raise ValueError('消息必须是对象。')
if request.get('role') != 'assistant' or response.get('role') != 'user':
raise ValueError('历史必须由完整请求与结果交替组成。')
blocks, results = request.get('content'), response.get('content')
if type(blocks) is not list or type(results) is not list:
raise ValueError('本课只整理工具交接历史。')
if any(type(b) is not dict or b.get('type') not in ('text','tool_use') for b in blocks):
raise ValueError('请求块不受支持。')
ids = [b.get('id') for b in blocks if b['type'] == 'tool_use']
if not ids or any(type(i) is not str or not i for i in ids) or len(set(ids)) != len(ids):
raise ValueError('工具调用ID必须非空且唯一。')
if any(type(b) is not dict or b.get('type') != 'tool_result' or type(b.get('content')) is not str or type(b.get('is_error')) is not bool for b in results):
raise ValueError('结果格式不受支持。')
found = [b.get('tool_use_id') for b in results]
if any(type(i) is not str for i in found) or len(found) != len(ids) or set(found) != set(ids):
raise ValueError('每个工具请求必须有且只有一个同ID结果。')
def split_history(messages, keep_recent=1):
"""在工具请求已经收齐结果时整理；最近完整交接作为不可拆的尾部。"""
if type(keep_recent) is not int or not 1 <= keep_recent <= 4:
raise ValueError('至少保留1组，最多4组最近交接。')
if type(messages) is not list or not messages or type(messages[0]) is not dict or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('历史必须从文本用户问题开始。')
if (len(messages) - 1) % 2:
raise ValueError('工具结果尚未收齐，不能压缩。')
groups = []
for index in range(1, len(messages), 2):
checked_exchange(messages[index], messages[index + 1])
groups.append(deepcopy(messages[index:index + 2]))
cut = max(0, len(groups) - keep_recent)
return {'question': deepcopy(messages[0]), 'older': groups[:cut], 'recent': groups[cut:]}
def trim_results(messages, receipt, keep_recent=1, max_chars=600):
"""调用者已保存快照，才可移出旧正文；角色、ID和错误标记不变。"""
if type(receipt) is not str or not receipt or type(max_chars) is not int or max_chars < 1:
raise ValueError('需要归档收据和正的正文上限。')
parts = split_history(messages, keep_recent)
candidate = deepcopy(messages)
for index in range(2, 1 + 2 * len(parts['older']), 2):
for slot, block in enumerate(candidate[index]['content']):
if len(block['content']) > max_chars:
block['content'] = f'[旧正文已移出；宿主快照 {receipt}，消息{index}/块{slot}，ID={block["tool_use_id"]}。此标记不是执行结果。]'
return candidate
def history_bytes(messages):
return len(json.dumps(messages, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
def archive_writer(directory):
directory = Path(directory)
def save(messages):
payload = ''.join(json.dumps(m, ensure_ascii=False) + '\n' for m in messages)
if len(payload.encode('utf-8')) > 1_000_000:
raise ValueError('快照超过本例1MB上限。')
directory.mkdir(parents=True, exist_ok=True)
target = directory / ('context-' + uuid4().hex[:12] + '.jsonl')
with target.open('x', encoding='utf-8') as stream:
stream.write(payload)
return target.name
return save
archive=archive_writer(Path(_temp.name))
if scenario=="archive_error":
def archive(_):raise OSError("教学归档失败")
def summarize(prefix):
return "" if scenario=="bad_summary" else "旧正文已归档；仍须检查当前文件并取得运行许可。"
# 调用入口
try:
observed=prepare_context(messages,question,state,archive,summarize,budget=1600)
print("已生成候选；原历史保持不变。")
except (ValueError,OSError) as error:
observed={"error":str(error)}
print(str(error))
assert messages==before
- 先检查预算；超限时先归档。
- 本地移出正文，不够才按单独预算摘要旧前缀。
- 复查候选；失败保留原输入，成功返回后才由循环提交。
选择“移出即可”，走完旧正文替换就返回，没有消耗摘要额度。“需要摘要”准备了多组较短结果，单条不到正文阈值，累计却超过总预算，所以本地移出仍不够，才走到 summarize。
在 state['summaries'] += 1 前停一下：这是一次尝试的占用点。下一行接口失败、返回空文本或者给出过长摘要，都已经消耗这次额度。状态里保留次数和快照收据，不代表候选历史已经提交。
摘要接口只收到旧前缀的副本。它看到的旧大正文可能已经是引用，因此不能据此编造原始日志的细节；近期组留在接口外面，后面直接接回候选。教学绑定用固定文本代替摘要模型，归档则在浏览器临时虚拟文件上真实执行；两个边界在展开区都可查看。
摘要必须是非空文本且不超过1000字符。普通 SDK 摘要调用还会检查 stop_reason=end_turn；被 max_tokens 截断的文本不能作为完整摘要。类型和长度只能验证结构，不能证明内容准确，关键判断仍需当前证据。
重建时带回什么，失败时留下什么？
新的首条 user 内容从宿主持有的原问题开始，再附摘要和快照说明。这里不让模型把原问题重新概括一遍，也不把旧摘要无限叠加回去。近期组直接从副本接回来，其请求、结果和ID保持原样。
宿主另外提醒：摘要只供参考，旧允许不授权本次运行，旧规范需要重新加载。第五课的当前计划仍从独立 state 附到每次请求副本，system和工具定义也由原组装器提供。这些来源不会因为摘要写了“全部完成”就自动改变。
最后再次检查候选的交接格式、是否满足预算、是否确实比输入更小。只有走到 return candidate，调用者才拿到一份可提交的新视图。归档、摘要或复查失败，都不会原地覆盖传进来的 messages。
选择“近期本身超限”，只有一个大交接，已经没有旧前缀可摘要。函数会停止并要求缩小输出；选择“摘要预算用完”，也不会继续调用接口。失败后可能已经留下快照或消耗摘要次数，但原输入仍然保持不变。
给原循环加一个提交点
之前的 run_loop 在模型请求异常时返回 model_error。上下文整理失败应在发请求之前交回宿主，因此第一课唯一循环新增可选 prepare=None：默认保持旧行为；提供回调时，先在深副本上准备候选，成功后才切片替换活跃列表。
下面是新增位置的静态摘录，完整实现仍在公开 s01/core/runtime.py，没有为第八课复制另一份循环：
python
if prepare is not None:
try:
candidate = prepare(deepcopy(messages))
if type(candidate) is not list:
raise ValueError("prepare必须返回候选消息列表")
except Exception as error:
return finish("context_error", "上下文准备失败；原历史保留。", number - 1)
messages[:] = candidate
response = ask(messages)列表返回类型由循环检查，协议与预算由本课 prepare_context 检查。异常时本轮 ask 尚未执行，所以 calls等于此前已发送的父请求数；它不包含子任务或摘要请求，后两者分别有自己的预算。
装配顺序也要保持：prepare先整理活跃历史，ask再通过第五课 prepare_messages给本次请求副本附当前计划。模型收到的最终请求还包含system、工具定义与计划，这些没有算入3200字节。因此这个预算是可验证的教学计量，不是对API token上限的保证。
跑一遍，再故意让它停下
Python3.10+ 标准库即可运行离线路径：
bash
python3 examples/s08-context-compact/user_code/practice.py --decision allow
python3 examples/s08-context-compact/user_code/practice.py --decision deny
python3 examples/s08-context-compact/user_code/practice.py --summary-limit 0
python3 examples/s08-context-compact/user_code/practice.py --budget 1600
python3 examples/s08-context-compact/user_code/verify_practice.py默认主模型、子模型和摘要模型都是固定响应替身，文件、快照及获准后的进程真实执行。父9次、子3次判断之外，额外发生1次摘要；旧正文先多次移出，之后才需要摘要。输出会列出整理前后字节数与快照收据，允许时最终取得42并核对，拒绝时程序没有执行标记。
固定主模型用近期结果的调用ID推进教学脚本，避免历史被摘要后又从第一步开始；它不证明真实模型能无损记住所有进度。若核对所需的完整预期已不在活跃历史，脚本会报告缺少依据，不能凭摘要声称一致。
将摘要额度设为0，第三次父请求执行后，下一次整理就停为context_error；将预算设为1600，第一份近期日志已无法容纳，第二次父请求不会发出。这两条路径都没有启动程序，也没有自动反复摘要。
公开34项断言验证完整组、原ID/错误标记、JSONL快照、失败不提交、旧许可与当前计划，以及完整输出多一行时不能误判一致。固定 Anthropic SDK 1.13.0 的模拟 HTTP 对照允许/拒绝各13次请求，并拒收截断摘要；没有调用在线模型。显式 --live 才接真实主/子/摘要模型，摘要也会消耗实际请求与费用。
本例保存的是每次整理前的输入快照，多次整理形成记录链，并没有实现每轮完整审计或跨进程恢复。课程参考还讨论大结果落盘和API超长后的应急处理；这里先让请求前整理与失败保留可验证，单条近期结果太大就停下，没有无限重试或自动检索归档的承诺。
资料核对于2026-10-19：课程 Context Compact、Claude Code 官方机制说明与Claude Platform 压缩文档。本例使用普通Messages请求生成摘要，读取上限、字节阈值、历史语法和提交接口均是本册独立设计。