fix(session,_recovery,types): 修 Codex 对抗审查报的五条
逐条核实全部成立。最重的是第一条：停止判定的结果只存在于结束记录里，而步记录与结束记录是 两次写。崩在两者之间那次判定就丢了，恢复照常回到预算准入——一次「恰好用满预算完成」被改写 成「预算耗尽」（两条轨迹长度一模一样），一次已经达成目标的运行接着往下跑，一次该以连续解析 失败收尾的运行再花一次模型调用。修法是续跑时把那次判定重演一遍：判定要的东西全在日志里 （动作结果在步记录那条原子写里、工具名在步记录上、模型回复在模型调用结果里）。最终回答那 一档更麻烦，答案文本只存在于结束记录里，靠重新解释那条已存下来的回复找回来——那时副作用还 没发生，重新解释是安全的。五种收尾各一条测试。 其余四条： - 模型调用失败步在续跑时补写，prompt_chars 填了 0。重新装配出来的和被打断时是同一份， 照它算。填 0 是把轨迹里那一列改写成假值。 - 取消正好落在写运行开始记录那一下时不留取消标记，留下一份只有开始记录的日志：run 因标识 已存在而拒绝、resume 当成可以从第 0 步续跑。用 shield 让那条记录一定落地，取消结束记录 才有地方挂（反过来先写结束记录会拼出结构上说不通的日志）。 - ModelCallResult 能同时带 reply 和 failure，而恢复只看 reply is None，于是把一次失败的 调用当成成功、接着解释那段回复执行动作。加构造期不变量。 - call_id 空串一路能进持久化记录。docstring 里那句「绝不能是空串」原本没有任何东西守着。 顺带把主循环里重复的一次上下文装配去掉，并把只在续跑第一次迭代成立的那个分支挪出循环。
This commit is contained in:
@@ -306,6 +306,103 @@ class _Driver:
|
||||
)
|
||||
return result
|
||||
|
||||
async def _drain_within_grace(self, writing: "asyncio.Future[None]") -> None:
|
||||
"""等一个已经发出去的写入在宽限期内落完，等不到就放弃。
|
||||
|
||||
取消进来时那次写入可能还在半路。不等它就写结束记录的话，两条记录可能以相反的顺序落地，
|
||||
而一份「先有结束记录、后有开始记录」的日志在恢复那边是结构上说不通的状态。
|
||||
"""
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
asyncio.shield(writing), timeout=self._request.cancel_grace_seconds
|
||||
)
|
||||
except TimeoutError:
|
||||
writing.cancel()
|
||||
logger.error(
|
||||
"取消宽限期内没写完在途记录，运行 %s 的日志可能不完整", self._request.run_id
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("取消时在途记录写失败，运行 %s 的日志可能不完整", self._request.run_id)
|
||||
|
||||
async def start(self, snapshot: Mapping[str, str]) -> None:
|
||||
"""写运行开始记录。取消正好落在这一下时，屏蔽它让它写完再补结束记录。
|
||||
|
||||
不屏蔽的话会留下一份**只有开始记录**的日志：`run` 因为标识已存在而拒绝，`resume` 把它
|
||||
当成可以从第 0 步续跑——而它其实是被人主动叫停的。屏蔽让这条记录一定落地，取消结束
|
||||
记录才有地方挂：先写结束记录再写开始记录会拼出一份结构上说不通的日志。
|
||||
"""
|
||||
writing: asyncio.Future[None] = asyncio.ensure_future(
|
||||
self._definition.store.write_run_started(
|
||||
RunStarted(run_id=self._request.run_id, parameter_snapshot=snapshot)
|
||||
)
|
||||
)
|
||||
try:
|
||||
await asyncio.shield(writing)
|
||||
except asyncio.CancelledError:
|
||||
await self._drain_within_grace(writing)
|
||||
await self._finish_cancelled()
|
||||
raise
|
||||
|
||||
def _assemble(self) -> tuple[Message, ...]:
|
||||
return assemble(
|
||||
context=self._request.context,
|
||||
injections=self._request.injections,
|
||||
steps=self._steps,
|
||||
observation_template=self._request.observation_template,
|
||||
)
|
||||
|
||||
def seed(self, plan: ResumePlan) -> None:
|
||||
"""把从日志读回来的计数与步序列装进来。"""
|
||||
self._counters = RunCounters(
|
||||
steps_appended=plan.steps_appended,
|
||||
actions_executed=plan.actions_executed,
|
||||
consecutive_parse_failures=plan.consecutive_parse_failures,
|
||||
)
|
||||
self._steps = list(plan.steps)
|
||||
|
||||
async def settle_interrupted_tail(self, plan: ResumePlan) -> RunResult | None:
|
||||
"""把最后一步之后的那次停止判定重演一遍。返回 `None` 表示那一步之后确实该接着跑。
|
||||
|
||||
**这一步不是可选的优化，是正确性。** 停止判定的结果只存在于结束记录里，而步记录与结束
|
||||
记录是两次写；崩在两者之间，那次判定就丢了。恢复照常回到预算准入的话：一次「恰好用满
|
||||
预算完成」会被改写成「预算耗尽」（两者的轨迹长度一模一样，事后分不出来），一次已经
|
||||
达成目标的运行会接着往下跑，一次该以连续解析失败收尾的运行会再花一次模型调用。
|
||||
|
||||
判定所需的东西全都在日志里：动作结果在步记录那条原子写里，工具名在步记录上，模型回复
|
||||
在模型调用结果里。这里只是把它们重新读一遍。
|
||||
"""
|
||||
entry = plan.last_step_completed
|
||||
if entry is None:
|
||||
return None
|
||||
step = entry.step
|
||||
|
||||
if not step.parse_ok and step.parse_error is None:
|
||||
# 模型调用失败那一步。它一写完就该以模型调用失败收尾。
|
||||
return await self._finish(StopReason.LLM_ERROR)
|
||||
|
||||
if not step.parse_ok:
|
||||
stop = parse_failure_admission(self._counters, self._request.budget)
|
||||
return None if stop is None else await self._finish(stop)
|
||||
|
||||
if entry.action_outcome is not None:
|
||||
spec = None if step.tool_name is None else self._request.tools.spec_for(step.tool_name)
|
||||
stop = completion_verdict(
|
||||
entry.action_outcome, bool(spec is not None and spec.completes_run)
|
||||
)
|
||||
return None if stop is None else await self._finish(stop)
|
||||
|
||||
# 有效决策却没有动作结果，只剩最终回答那一档。答案文本只存在于结束记录里，所以重新
|
||||
# 解释那条已经存下来的回复把它找回来——那时副作用还没发生，重新解释是安全的。
|
||||
if plan.last_reply is None:
|
||||
raise CorruptLogError("最后一步是最终回答，却找不到它那次模型调用的回复")
|
||||
decision = self._definition.decision_parser.parse(plan.last_reply).decision
|
||||
if not isinstance(decision, FinalAnswer):
|
||||
raise CorruptLogError(
|
||||
"最后一步记成了最终回答，重新解释同一条回复却得到别的分支——"
|
||||
"解释器在两次运行之间被换过，或者它不是确定性的"
|
||||
)
|
||||
return await self._finish(StopReason.AGENT_FINISHED, final_answer=decision.text)
|
||||
|
||||
async def _finish_cancelled(self) -> None:
|
||||
"""取消进来时尽力写下结束标记，宽限期用完就放弃。
|
||||
|
||||
@@ -322,22 +419,13 @@ class _Driver:
|
||||
final_answer=None,
|
||||
steps=tuple(self._steps),
|
||||
)
|
||||
writing = asyncio.ensure_future(
|
||||
self._definition.store.write_run_finished(
|
||||
RunFinished(run_id=self._request.run_id, result=result)
|
||||
await self._drain_within_grace(
|
||||
asyncio.ensure_future(
|
||||
self._definition.store.write_run_finished(
|
||||
RunFinished(run_id=self._request.run_id, result=result)
|
||||
)
|
||||
)
|
||||
)
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
asyncio.shield(writing), timeout=self._request.cancel_grace_seconds
|
||||
)
|
||||
except TimeoutError:
|
||||
writing.cancel()
|
||||
logger.error(
|
||||
"取消宽限期内没写完结束记录，运行 %s 的日志缺结束标记", self._request.run_id
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("取消时写结束记录失败，运行 %s 的日志缺结束标记", self._request.run_id)
|
||||
|
||||
# -- 主循环 -------------------------------------------------------------
|
||||
|
||||
@@ -372,26 +460,23 @@ class _Driver:
|
||||
skip_model_intent = plan is not None and plan.action is ResumeAction.REDO_MODEL_CALL
|
||||
skip_action_intent = plan is not None and plan.action is ResumeAction.REPLAY_LAST_ACTION
|
||||
if plan is not None:
|
||||
self._counters = RunCounters(
|
||||
steps_appended=plan.steps_appended,
|
||||
actions_executed=plan.actions_executed,
|
||||
consecutive_parse_failures=plan.consecutive_parse_failures,
|
||||
)
|
||||
self._steps = list(plan.steps)
|
||||
self.seed(plan)
|
||||
pending_reply = plan.pending_reply
|
||||
pending_failure = plan.pending_failure
|
||||
|
||||
# 模型调用失败那一步的步记录还没写完就断了：补上它，然后以模型调用失败收尾。只可能
|
||||
# 发生在续跑的第一次迭代上，所以判在循环外面。
|
||||
# **规模照现在重新装配出来的算**，那和被打断时装配出来的是同一份——历史、上下文、注入
|
||||
# 内容都没变，而续跑守卫已经比对过它们了。填 0 会把那一列改写成一个假值。
|
||||
if pending_failure is not None:
|
||||
started = time.monotonic()
|
||||
chars = prompt_chars(self._assemble())
|
||||
await self._write_step(self._failed_call_step(len(self._steps), chars, started), None)
|
||||
return await self._finish(StopReason.LLM_ERROR)
|
||||
|
||||
while True:
|
||||
call_index = len(self._steps)
|
||||
|
||||
# 模型调用失败那一步的步记录还没写完就断了：补上它，然后以模型调用失败收尾。
|
||||
if pending_failure is not None:
|
||||
step = self._failed_call_step(
|
||||
call_index, prompt_chars_used=0, started=time.monotonic()
|
||||
)
|
||||
await self._write_step(step, None)
|
||||
return await self._finish(StopReason.LLM_ERROR)
|
||||
|
||||
# A 预算准入。放在开头而不是上一次迭代的结尾：一次「恰好用满预算完成」的运行走的
|
||||
# 是完成判定，放结尾它会先撞上预算上限，而两者的轨迹长度一模一样。
|
||||
stop = budget_admission(self._counters, self._request.budget)
|
||||
@@ -399,12 +484,7 @@ class _Driver:
|
||||
return await self._finish(stop)
|
||||
|
||||
started = time.monotonic()
|
||||
messages = assemble(
|
||||
context=self._request.context,
|
||||
injections=self._request.injections,
|
||||
steps=self._steps,
|
||||
observation_template=self._request.observation_template,
|
||||
)
|
||||
messages = self._assemble()
|
||||
chars = prompt_chars(messages)
|
||||
|
||||
# B 规模判定。命中时不产生步记录、也不写任何意图——模型还没被调用、没花钱、没有
|
||||
@@ -696,10 +776,9 @@ async def run(definition: AgentDefinition, request: RunRequest) -> RunResult:
|
||||
f"运行标识 {request.run_id!r} 已经有日志了。要接着跑用 resume；"
|
||||
"这里不覆盖，因为那会毁掉一次已经花完钱的运行的留痕"
|
||||
)
|
||||
await definition.store.write_run_started(
|
||||
RunStarted(run_id=request.run_id, parameter_snapshot=_merged_snapshot(definition, request))
|
||||
)
|
||||
return await _Driver(definition, request).drive()
|
||||
driver = _Driver(definition, request)
|
||||
await driver.start(_merged_snapshot(definition, request))
|
||||
return await driver.drive()
|
||||
|
||||
|
||||
async def resume(definition: AgentDefinition, request: RunRequest) -> RunResult:
|
||||
@@ -736,10 +815,19 @@ async def resume(definition: AgentDefinition, request: RunRequest) -> RunResult:
|
||||
if plan.finished_result is None:
|
||||
raise CorruptLogError("有结束标记却读不出结果，这条记录坏了")
|
||||
return plan.finished_result
|
||||
driver = _Driver(definition, request)
|
||||
if plan.action is ResumeAction.STOP_UNKNOWN:
|
||||
return await _Driver(definition, request).finish_resume_unknown(plan)
|
||||
return await driver.finish_resume_unknown(plan)
|
||||
|
||||
return await _Driver(definition, request).drive(plan)
|
||||
if plan.action is ResumeAction.CONTINUE_AT_NEXT_STEP:
|
||||
# 上一步是完整的，但它之后那次停止判定的结果只存在于结束记录里，而那条记录没写下来。
|
||||
# 先把它重演一遍——不重演的话，一次已经达成目标的运行会被改写成预算耗尽或者接着往下跑。
|
||||
driver.seed(plan)
|
||||
settled = await driver.settle_interrupted_tail(plan)
|
||||
if settled is not None:
|
||||
return settled
|
||||
|
||||
return await driver.drive(plan)
|
||||
|
||||
|
||||
__all__ = [
|
||||
|
||||
Reference in New Issue
Block a user