第 5 周：Agent 跑到一半被 kill -9，从哪接着跑？

$ python3 bughunt_worker.py /tmp/bh --checkpoint log --effects idempotent --delay 0.05 &
$ sleep 0.75; kill -9 $!
exit=137
$ wc -l < /tmp/bh/events.jsonl; tail -2 /tmp/bh/events.jsonl
       7
{"finding": false, "page": 5, "seq": 6, "t": "page_done", "tokens": 1185}
{"finding": false, "page": 6, "seq": 7, "t": "page_done", "tokens": 1222}
$ python3 bughunt_worker.py /tmp/bh --checkpoint log --effects idempotent
{"status": "done", "next_page": 13, "findings": [3, 7, 10], "aborted": [], "tokens": 14886, "resume": {"source": "log", "replayed": 7, "bad_lines": 0}}
$ sqlite3 /tmp/bh/tracker.db 'select page,key from reports'
3|bughunt-task-001:p03
7|bughunt-task-001:p07
10|bughunt-task-001:p10

// Only the newest compaction can bound replay. If it is incomplete, an older compaction
// cannot replace the history or window state that the newer one may have changed.
let (index, compacted) = rollout_items
    .iter()
    .enumerate()
    .rev()                                   // 倒着找
    .find_map(|(index, item)| match item {
        RolloutItem::Compacted(compacted) => Some((index, compacted)),
        _ => None,
    })?;                                     // 找到的第一个就是最新的 compacted
// ...
if compacted.replacement_history.is_none()
    || compacted.window_number.is_none()
    // ...
{
    return None;                             // 最新的不完整：不用它截短输入，扫描全部条目
}

db.execute("CREATE TABLE IF NOT EXISTS reports (id INTEGER PRIMARY KEY, page INT, key TEXT UNIQUE)")
db.execute("INSERT OR IGNORE INTO reports (page, key) VALUES (?, ?)", (page, key))  # key 为 NULL 时不去重

// The model emitted a tool call; log it, persist the item immediately, and queue the tool execution.
Ok(Some(call)) => {
    // ...
    record_completed_response_item(ctx.sess.as_ref(), ctx.step_context.as_ref(), &item)
        .await;                                  // 先等调用条目记录完
    let cancellation_token = ctx.cancellation_token.child_token();
    let tool_future: InFlightFuture<'static> = Box::pin(
        ctx.tool_runtime
            .clone()
            .handle_tool_call(call, cancellation_token),   // 再排队执行工具
    );
    // ...
}

ResponseItem::FunctionCall { id, call_id, .. }
    if !function_output_ids.contains(call_id.as_str()) =>   // 这个调用找不到输出
{
    info!("Function call output is missing for call id: {call_id}");
    missing_outputs_to_insert.push((
        idx,                                                 // 记下位置，之后插在调用后面
        ResponseItemEnvelope::new(ResponseItem::FunctionCallOutput {
            id: synthetic_output_id("fco", id.as_deref()),
            call_id: Some(call_id.clone()),                  // 和调用配对
            // ...
            output: FunctionCallOutputPayload::from_text("aborted".to_string()),  // 内容就是 "aborted"
            // ...
        }),
    ));
}

let mut value: Value = match serde_json::from_str(&line) {
    Ok(value) => value,
    Err(e) => {
        warn!("failed to parse line as JSON: {line:?}, error: {e}");
        parse_errors = parse_errors.saturating_add(1);   // 计数，不中断
        continue;                                         // 跳过这一行
    }
};

fn ensure_rollout_is_newline_terminated(file: &mut File) -> std::io::Result<()> {
    if file.metadata()?.len() == 0 {          // 空文件，不用管
        return Ok(());
    }
    file.seek(SeekFrom::End(-1))?;            // 跳到最后一个字节
    let mut final_byte = [0];
    file.read_exact(&mut final_byte)?;        // 读出来
    if final_byte[0] != b'\n' {               // 不是换行：上次写到一半
        file.write_all(b"\n")?;               // 补一个，把半行隔离成单独的一行
        file.flush()?;
    }
    Ok(())
}

#[test]
fn append_repair_terminates_nonempty_rollout_tail() -> std::io::Result<()> {
    let home = TempDir::new().expect("temp dir");
    let rollout_path = home.path().join("rollout.jsonl");
    fs::write(&rollout_path, b"{\"type\":\"event_msg\"}")?;   // 最后一行没有换行
    drop(open_log_file(&rollout_path)?);                           // 打开一次：补换行
    drop(open_log_file(&rollout_path)?);                           // 再打开一次：不应再补

    assert_eq!(fs::read(&rollout_path)?, b"{\"type\":\"event_msg\"}\n");  // 恰好一个换行
    Ok(())
}

$ python3 -c "
import os, signal
f = open('/tmp/laq_buf.txt', 'w'); f.write('page_done 7\n')
g = open('/tmp/laq_buf2.txt', 'w'); g.write('page_done 7\n'); g.flush()
os.kill(os.getpid(), signal.SIGKILL)"
$ wc -c /tmp/laq_buf.txt /tmp/laq_buf2.txt
       0 /tmp/laq_buf.txt
      12 /tmp/laq_buf2.txt

合成数据：12 个页面，注入 Bug 的页面 [3, 7, 10]，hybrid 每 4 页一份快照

== 1. 穷举崩溃点：每个点 kill -9 一次，再续跑到结束 ==
检查点 | 副作用 | 点数 | 正确 | 恢复失败 | 重复报告 | 漏报 | 状态不一致 | 重建不一致 | 平均重做模型调用 | 平均重放事件 | 最多重放事件
none | naive | 51 | 10 | 0 | 41 | 0 | 0 | 0 | 6.51 | 0 | 0
snapshot | naive | 51 | 45 | 0 | 6 | 0 | 0 | 0 | 0.76 | 0 | 0
snapshot | at_most_once | 54 | 45 | 0 | 0 | 3 | 6 | 0 | 0.61 | 0 | 0
snapshot | idempotent | 54 | 54 | 0 | 0 | 0 | 0 | 0 | 0.78 | 0 | 0
snapshot_inplace | idempotent | 54 | 42 | 12 | 0 | 0 | 0 | 0 | 0.71 | 0 | 0
log | idempotent | 54 | 54 | 0 | 0 | 0 | 0 | 0 | 0.78 | 7.26 | 15
log_norepair | idempotent | 54 | 45 | 0 | 0 | 0 | 0 | 9 | 0.78 | 7.26 | 15
hybrid | idempotent | 54 | 54 | 0 | 0 | 0 | 0 | 0 | 0.78 | 1.98 | 4

snapshot/naive 出错的崩溃点：p03:after_submit→重复报告，p03:mid_checkpoint→重复报告，p07:after_submit→重复报告，p07:mid_checkpoint→重复报告，p10:after_submit→重复报告，p10:mid_checkpoint→重复报告
snapshot/at_most_once 出错的崩溃点：p03:after_intent→漏报，p03:after_submit→状态与外部不一致，p03:mid_checkpoint→状态与外部不一致，p07:after_intent→漏报，p07:after_submit→状态与外部不一致，p07:mid_checkpoint→状态与外部不一致，p10:after_intent→漏报，p10:after_submit→状态与外部不一致，p10:mid_checkpoint→状态与外部不一致
snapshot_inplace/idempotent 出错的崩溃点：p01:mid_checkpoint→恢复失败，p02:mid_checkpoint→恢复失败，p03:mid_checkpoint→恢复失败，p04:mid_checkpoint→恢复失败，p05:mid_checkpoint→恢复失败，p06:mid_checkpoint→恢复失败，p07:mid_checkpoint→恢复失败，p08:mid_checkpoint→恢复失败，p09:mid_checkpoint→恢复失败，p10:mid_checkpoint→恢复失败，p11:mid_checkpoint→恢复失败，p12:mid_checkpoint→恢复失败
log_norepair/idempotent 出错的崩溃点：p01:mid_checkpoint→重建不一致，p02:mid_checkpoint→重建不一致，p04:mid_checkpoint→重建不一致，p05:mid_checkpoint→重建不一致，p06:mid_checkpoint→重建不一致，p08:mid_checkpoint→重建不一致，p09:mid_checkpoint→重建不一致，p11:mid_checkpoint→重建不一致，p12:mid_checkpoint→重建不一致

== 2. 外部 kill -9：每页两步各 sleep 10ms，随机时刻 SIGKILL，40 次 / 组，seed=2026 ==
snapshot/naive: {'正确': 39, '重复报告': 1}（其中 1 次进程已先跑完、没杀到）
snapshot/idempotent: {'正确': 40}（其中 1 次进程已先跑完、没杀到）
log/idempotent: {'正确': 40}（其中 1 次进程已先跑完、没杀到）

== 3. 每次写检查点的耗时（本机，200 次取中位数）==
快照（写临时文件 + fsync + rename）：0.228 ms
日志追加一行（只 write，不 fsync）：0.0020 ms
日志追加一行 + fsync：0.033 ms

== 2. 外部 kill -9：每页两步各 sleep 10ms，随机时刻 SIGKILL，40 次 / 组，seed=2026 ==
snapshot/naive: {'正确': 40}（其中 0 次进程已先跑完、没杀到）
snapshot/idempotent: {'正确': 40}（其中 0 次进程已先跑完、没杀到）
log/idempotent: {'正确': 40}（其中 0 次进程已先跑完、没杀到）

== 3. 每次写检查点的耗时（本机，200 次取中位数）==
快照（写临时文件 + fsync + rename）：0.300 ms
日志追加一行（只 write，不 fsync）：0.0025 ms
日志追加一行 + fsync：0.037 ms

$ python3 -m unittest -v test_checkpoint_resume
test_hybrid_replay_is_bounded_by_snapshot_interval (test_checkpoint_resume.CrashSweepTest) ... ok
test_log_idempotent_survives_every_crash_point (test_checkpoint_resume.CrashSweepTest) ... ok
test_snapshot_naive_duplicates_only_in_submit_window (test_checkpoint_resume.CrashSweepTest) ... ok
test_apply_event_returns_new_state (test_checkpoint_resume.ReplayUnitTest) ... ok
test_inplace_snapshot_torn_write_is_detected (test_checkpoint_resume.ReplayUnitTest) ... ok
test_torn_last_line_is_skipped_and_newline_repaired (test_checkpoint_resume.ReplayUnitTest) ... ok

----------------------------------------------------------------------
Ran 6 tests in 17.184s

OK

def apply_event(state, ev):
    """把一条事件应用到状态上，返回新状态（不改原对象）。"""
    s = {**state, "findings": list(state["findings"]), "aborted": list(state["aborted"]), "seq": ev["seq"]}
    if ev["t"] == "intent":
        s["pending"] = {"page": ev["page"], "key": ev["key"]}
    elif ev["t"] == "page_done":
        s["next_page"] = ev["page"] + 1
        s["tokens"] += ev["tokens"]
        s["pending"] = None
        if ev["finding"]:
            s["findings"].append(ev["page"])
    elif ev["t"] == "aborted":
        s["next_page"] = ev["page"] + 1
        s["pending"] = None
        s["aborted"].append(ev["page"])
    return s


def atomic_write(path, data, torn):
    tmp = path.with_suffix(".tmp")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
    try:
        write_all_or_torn(fd, data, torn)
        os.fsync(fd)
    finally:
        os.close(fd)
    os.replace(tmp, path)  # POSIX 保证 rename 原子：要么旧文件，要么新文件

# run() 里的主循环
    pending = state["pending"]
    if pending and effects == "at_most_once":
        record({"t": "aborted", "page": pending["page"]})  # 不知道提交过没有，宁可不做

    for page in range(state["next_page"], N_PAGES + 1):
        time.sleep(delay)
        ledger(root, f"model {page}")
        tokens = model_tokens(page)
        crash.point(page, "after_model")
        time.sleep(delay)
        ledger(root, f"visit {page}")
        crash.point(page, "after_visit")
        found = page in BUG_PAGES
        if found:
            key = None
            if effects != "naive":
                key = f"{TASK_ID}:p{page:02d}"
                record({"t": "intent", "page": page, "key": key})
                crash.point(page, "after_intent")
            submit_report(root, page, key)
            crash.point(page, "after_submit")
        record({"t": "page_done", "page": page, "tokens": tokens, "finding": found}, torn=crash.torn(page))
        crash.point(page, "after_checkpoint")