Genarrative/rust/crates/agent-storage-sqlite/tests/storage.rs

#[cfg(feature = "core-adapter")]
use agent_runtime_contracts::{
    DurableCheckpointInput, DurableLeaseIdentity, DurableToolCallCheckpointRuntimeCommit,
    DurableToolCallInput, DurableToolCallRuntimeCommit,
};
#[cfg(feature = "core-adapter")]
use agent_runtime_core::{
    RUNTIME_EVENT_SCHEMA_VERSION, RunSnapshot, RuntimeEvent, RuntimeEventKind, RuntimeSnapshot,
    RuntimeStore, StoreErrorKind, SystemClock, ToolCall, ToolResult, reduce,
};
use agent_storage_sqlite::{
    MAX_STALE_RUN_SCAN_LIMIT, NewApproval, NewCheckpoint, NewEvent, NewExternalSession, NewRun,
    NewSession, NewSnapshot, NewToolCall, SqliteStore, StorageError,
};
use serde_json::json;
use std::path::PathBuf;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

fn local_tempdir() -> tempfile::TempDir {
    // Respect an explicit TMPDIR, but keep direct test runs out of /tmp by
    // defaulting to the repository-wide ~/data/tmp location. TempDir owns the
    // child directory cleanup even when a test returns early or panics.
    let parent = std::env::var_os("TMPDIR")
        .filter(|value| !value.is_empty())
        .map(PathBuf::from)
        .or_else(|| {
            std::env::var_os("HOME")
                .map(PathBuf::from)
                .map(|home| home.join("data/tmp"))
        })
        .expect("TMPDIR 或 HOME 未设置，无法创建测试数据库目录");
    std::fs::create_dir_all(&parent).expect("创建测试临时目录");
    tempfile::Builder::new()
        .prefix("agent-storage-test-")
        .tempdir_in(parent)
        .expect("tempdir")
}

fn store_with_run() -> (SqliteStore, String) {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "session-1".into(),
            agent_id: Some("agent-1".into()),
            status: "active".into(),
            metadata: json!({"purpose": "test"}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "run-1".into(),
            session_id: "session-1".into(),
            status: "running".into(),
            input: json!({"message": "hello"}),
        })
        .expect("create run");
    (store, "run-1".into())
}

/// 等待 SQLite 里记录的毫秒级到期时间，避免用固定 sleep 在慢 CI 上产生竞态。
fn wait_until_epoch_ms(target: i64) {
    let deadline = Instant::now() + Duration::from_secs(3);
    loop {
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock before unix epoch")
            .as_millis() as i64;
        if now >= target {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "timed out waiting for lease expiry at {target}, current time {now}"
        );
        std::thread::sleep(Duration::from_millis(1));
    }
}

fn user_message(text: &str) -> serde_json::Value {
    json!({
        "role": "user",
        "content": [{"type": "text", "text": text}]
    })
}

fn assistant_tool_call_message(call_id: &str) -> serde_json::Value {
    json!({
        "role": "assistant",
        "content": [{
            "type": "tool-call",
            "id": call_id,
            "name": "echo",
            "arguments": {"text": "hello"}
        }]
    })
}

fn tool_result_message(call_id: &str) -> serde_json::Value {
    json!({
        "role": "tool",
        "content": [{
            "type": "tool-result",
            "toolCallId": call_id,
            "output": {"ok": true},
            "isError": false
        }]
    })
}

#[cfg(feature = "core-adapter")]
fn tool_call_runtime_fixture() -> (
    SqliteStore,
    String,
    RuntimeSnapshot,
    Vec<RuntimeEvent>,
    ToolCall,
    DurableLeaseIdentity,
) {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    let session_id = "tool-transaction-session".to_owned();
    let run_id = "tool-transaction-run".to_owned();
    let runtime_id = "tool-transaction-runtime".to_owned();
    store
        .create_session(NewSession {
            id: session_id.clone(),
            agent_id: Some("agent-1".to_owned()),
            status: "active".to_owned(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: run_id.clone(),
            session_id,
            status: "queued".to_owned(),
            input: json!({"message": "tool transaction"}),
        })
        .expect("create run");

    let run = RunSnapshot::try_new(&run_id, "agent-1", "tool transaction", 1).expect("runtime run");
    let runtime_created = RuntimeEvent::runtime_created(&runtime_id, 1, 1).expect("runtime event");
    let run_created = RuntimeEvent::run_created(&runtime_id, 2, 2, &run).expect("run event");
    let started =
        RuntimeEvent::status_changed(&runtime_id, 3, 3, &run_id, RuntimeEventKind::RunStarted)
            .expect("started event");
    let mut snapshot = RuntimeSnapshot::try_new(&runtime_id).expect("runtime snapshot");
    for event in [&runtime_created, &run_created, &started] {
        snapshot = reduce(&snapshot, event).expect("reduce initial runtime event");
    }
    let mut runtime_store = store.clone();
    runtime_store
        .commit(
            &runtime_id,
            None,
            &snapshot,
            &[runtime_created, run_created, started],
        )
        .expect("commit initial runtime");
    let (_, lease) = store
        .claim_run_with_lease(
            &run_id,
            "tool-worker",
            "tool-lease",
            Duration::from_secs(30),
        )
        .expect("claim run");
    let call =
        ToolCall::try_new("tool-call-1", "echo", json!({"text": "hello"})).expect("tool call");
    let requested =
        RuntimeEvent::tool_call_requested(&runtime_id, snapshot.revision() + 1, 4, &run_id, &call)
            .expect("tool requested event");
    let next = reduce(&snapshot, &requested).expect("reduce tool request");
    let lease = DurableLeaseIdentity {
        worker_id: lease.worker_id,
        lease_token: lease.lease_token,
    };
    (store, runtime_id, next, vec![requested], call, lease)
}

fn reconciling_checkpoint(
    phase: &str,
    messages: serde_json::Value,
    provider_request_id: Option<&str>,
    tool_call_id: Option<&str>,
) -> (SqliteStore, String, i64) {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "reconcile-session".into(),
            agent_id: Some("agent-1".into()),
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    let run_id = "reconcile-run".to_owned();
    store
        .create_run(NewRun {
            id: run_id.clone(),
            session_id: "reconcile-session".into(),
            status: "queued".into(),
            input: json!({"message": "reconcile"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(&run_id, "worker-1", "lease-1", Duration::from_secs(30))
        .expect("claim run");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.clone(),
                phase: phase.to_owned(),
                step: 0,
                next_step: 0,
                messages,
                provider_request_id: provider_request_id.map(str::to_owned),
                tool_call_id: tool_call_id.map(str::to_owned),
                attempt: lease.attempt,
            },
            "worker-1",
            "lease-1",
        )
        .expect("save in-flight checkpoint");
    store
        .release_run_lease(&run_id, "worker-1", "lease-1")
        .expect("enter reconciliation");
    (store, run_id, lease.attempt)
}

#[test]
fn migrations_and_wal_are_configured() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    // SQLite 内存库不能启用 WAL，但迁移表和业务表应已存在。
    assert_eq!(store.journal_mode().expect("journal mode"), "memory");
    let session = store
        .create_session(NewSession {
            id: "s".into(),
            agent_id: None,
            status: "new".into(),
            metadata: json!({}),
        })
        .expect("insert");
    assert_eq!(store.get_session(&session.id).expect("read"), Some(session));
}

#[test]
fn checkpoint_is_fenced_and_round_trips() {
    let (store, _) = store_with_run();
    let run_id = "checkpoint-run".to_owned();
    store
        .create_run(NewRun {
            id: run_id.clone(),
            session_id: "session-1".into(),
            status: "queued".into(),
            input: json!({"message": "checkpoint"}),
        })
        .expect("create checkpoint run");
    store
        .claim_run_with_lease(&run_id, "worker-1", "token-1", Duration::from_secs(30))
        .expect("claim");
    let saved = store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.clone(),
                phase: "provider_in_flight".into(),
                step: 2,
                next_step: 3,
                messages: json!([{"role":"user","content":"hello"}]),
                provider_request_id: Some("req-1".into()),
                tool_call_id: Some("tool-1".into()),
                attempt: 1,
            },
            "worker-1",
            "token-1",
        )
        .expect("save checkpoint");
    assert_eq!(saved.phase, "provider_in_flight");
    assert_eq!(
        store
            .get_checkpoint(&run_id, "worker-1", "token-1")
            .expect("load"),
        Some(saved)
    );
    assert!(matches!(
        store.get_checkpoint(&run_id, "worker-2", "wrong-token"),
        Err(StorageError::LeaseConflict { .. })
    ));
    store
        .clear_checkpoint_with_lease(&run_id, "worker-1", "token-1")
        .expect("clear");
    assert_eq!(
        store
            .get_checkpoint(&run_id, "worker-1", "token-1")
            .expect("load cleared"),
        None
    );
}

#[test]
fn file_backed_restart_recovers_expired_provider_checkpoint() {
    let tempdir = local_tempdir();
    let database = tempdir.path().join("restart-recovery.db");
    let run_id = "restart-provider-run".to_owned();
    let lease_expires_at;

    // This scope models the worker process that created the in-flight call.
    // Dropping the store closes the SQLite connection before the recovery
    // process opens the same file again.
    {
        let store = SqliteStore::open(&database).expect("open file-backed sqlite");
        store
            .create_session(NewSession {
                id: "restart-session".into(),
                agent_id: Some("agent-1".into()),
                status: "active".into(),
                metadata: json!({"purpose": "restart-recovery"}),
            })
            .expect("create restart session");
        store
            .create_run(NewRun {
                id: run_id.clone(),
                session_id: "restart-session".into(),
                status: "queued".into(),
                input: json!({"message": "restart"}),
            })
            .expect("create restart run");

        let (_, lease) = store
            .claim_run_with_lease(
                &run_id,
                "restart-worker",
                "restart-lease",
                Duration::from_millis(20),
            )
            .expect("claim restart run");
        store
            .save_checkpoint_with_lease(
                NewCheckpoint {
                    run_id: run_id.clone(),
                    phase: "provider_in_flight".into(),
                    step: 2,
                    next_step: 3,
                    messages: json!([user_message("restart")]),
                    provider_request_id: Some("provider-restart-request".into()),
                    tool_call_id: None,
                    attempt: lease.attempt,
                },
                "restart-worker",
                "restart-lease",
            )
            .expect("save provider checkpoint");
        lease_expires_at = lease.lease_expires_at;
    }

    wait_until_epoch_ms(lease_expires_at);

    // A fresh connection must be able to identify the stale lease and recover
    // without deleting the checkpoint needed for explicit external reconciliation.
    {
        let store = SqliteStore::open(&database).expect("reopen file-backed sqlite");
        let recovered = store
            .recover_expired_run(&run_id)
            .expect("recover stale run after reopen");
        assert_eq!(recovered.status, "reconciling");
        assert_eq!(
            store
                .get_run(&run_id)
                .expect("read recovered run")
                .expect("recovered run exists")
                .status,
            "reconciling"
        );
        // 低层 `recover_expired_run` 只收束 run；需要同步 session/runtime 投影时，
        // 由 RuntimeService 的跨表 recovery facade 调用对应原子入口。
        assert_eq!(store.get_run_lease(&run_id).expect("read lease"), None);

        let checkpoint = store
            .read_checkpoint(&run_id)
            .expect("read preserved checkpoint")
            .expect("checkpoint remains after recovery");
        assert_eq!(checkpoint.phase, "provider_in_flight");
        assert_eq!(
            checkpoint.provider_request_id.as_deref(),
            Some("provider-restart-request")
        );
        assert_eq!(checkpoint.tool_call_id, None);
    }
}

#[test]
fn safe_resume_requeues_only_safe_reconciled_run_and_preserves_checkpoint() {
    let (store, _) = store_with_run();
    let run_id = "safe-resume-run".to_owned();
    store
        .create_run(NewRun {
            id: run_id.clone(),
            session_id: "session-1".into(),
            status: "queued".into(),
            input: json!({"message": "resume"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(&run_id, "worker-old", "token-old", Duration::from_secs(1))
        .expect("claim");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.clone(),
                phase: "safe".into(),
                step: 4,
                next_step: 5,
                messages: json!([{"role":"user","content":"resume"}]),
                provider_request_id: None,
                tool_call_id: None,
                attempt: 1,
            },
            "worker-old",
            "token-old",
        )
        .expect("save safe checkpoint");
    wait_until_epoch_ms(lease.lease_expires_at);
    let reconciled = store.recover_expired_run(&run_id).expect("reconcile");
    let reconciled_updated_at = reconciled.updated_at;
    // The first transition still records a fresh durable update; only an
    // already queued idempotent retry is allowed to preserve its timestamp.
    wait_until_epoch_ms(reconciled_updated_at.saturating_add(1));

    let queued = store.requeue_reconciled_run(&run_id).expect("safe resume");
    assert_eq!(queued.status, "queued");
    assert!(queued.updated_at > reconciled_updated_at);
    // worker 尚未启动时重复触发 resume-safe 仍应保持同一 queued run，
    // 让 spawn 失败或 CLI 重试不会把安全游标卡死在不可恢复状态。
    // 等待时钟前进，确保回归覆盖“不更新时间”而非仅仅碰巧落在同一毫秒。
    wait_until_epoch_ms(queued.updated_at.saturating_add(1));
    let queued_again = store
        .resume_safe_run(&run_id)
        .expect("idempotent safe resume");
    assert_eq!(queued_again, queued);
    assert_eq!(
        store
            .read_checkpoint(&run_id)
            .expect("read checkpoint")
            .unwrap()
            .phase,
        "safe"
    );

    // Requeue does not start the engine; a separate lease claim is still required.
    let (running, _) = store
        .claim_run_with_lease(&run_id, "worker-new", "token-new", Duration::from_secs(30))
        .expect("new worker claim");
    assert_eq!(running.status, "running");
    assert!(matches!(
        store.save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.clone(),
                phase: "safe".into(),
                step: 5,
                next_step: 6,
                messages: json!([]),
                provider_request_id: None,
                tool_call_id: None,
                attempt: 2,
            },
            "worker-old",
            "token-old",
        ),
        Err(StorageError::LeaseLost { .. }) | Err(StorageError::LeaseConflict { .. })
    ));
}

#[test]
fn safe_resume_rejects_non_safe_phase_and_cancelled_run() {
    let (store, _) = store_with_run();
    let run_id = "unsafe-resume-run".to_owned();
    store
        .create_run(NewRun {
            id: run_id.clone(),
            session_id: "session-1".into(),
            status: "queued".into(),
            input: json!({"message": "unsafe"}),
        })
        .expect("create run");
    let (_, lease) = store
        // 留出保存 checkpoint 的时间，再等待真实到期，避免测试本身先丢 lease。
        .claim_run_with_lease(&run_id, "worker", "token", Duration::from_millis(100))
        .expect("claim");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.clone(),
                phase: "provider_in_flight".into(),
                step: 1,
                next_step: 2,
                messages: json!([]),
                provider_request_id: Some("request".into()),
                tool_call_id: None,
                attempt: 1,
            },
            "worker",
            "token",
        )
        .expect("save in-flight checkpoint");
    wait_until_epoch_ms(lease.lease_expires_at);
    store.recover_expired_run(&run_id).expect("reconcile");
    assert!(matches!(
        store.requeue_reconciled_run(&run_id),
        Err(StorageError::InvalidInput(_))
    ));

    let (store, _) = store_with_run();
    let run_id = "cancelled-resume-run".to_owned();
    store
        .create_run(NewRun {
            id: run_id.clone(),
            session_id: "session-1".into(),
            status: "queued".into(),
            input: json!({"message": "cancelled"}),
        })
        .expect("create run");
    let (_, lease) = store
        // 先留出保存 safe checkpoint 的时间，再等待真实到期，避免测试本身
        // 因调度抖动在写入前丢 lease。
        .claim_run_with_lease(&run_id, "worker", "token", Duration::from_millis(100))
        .expect("claim");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.clone(),
                phase: "safe".into(),
                step: 1,
                next_step: 2,
                messages: json!([]),
                provider_request_id: None,
                tool_call_id: None,
                attempt: 1,
            },
            "worker",
            "token",
        )
        .expect("save safe checkpoint");
    store.request_cancel(&run_id).expect("cancel request");
    wait_until_epoch_ms(lease.lease_expires_at);
    let reconciled = store
        .recover_expired_run(&run_id)
        .expect("reconcile cancelled");
    assert!(reconciled.cancel_requested);
    assert!(matches!(
        store.resume_safe_run(&run_id),
        Err(StorageError::InvalidInput(_))
    ));
}

#[test]
fn terminal_transition_clears_checkpoint_atomically() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "checkpoint-session".into(),
            agent_id: None,
            status: "queued".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "checkpoint-terminal".into(),
            session_id: "checkpoint-session".into(),
            status: "queued".into(),
            input: json!({"message": "finish"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(
            "checkpoint-terminal",
            "worker-1",
            "token-1",
            Duration::from_secs(30),
        )
        .expect("claim");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: "checkpoint-terminal".into(),
                phase: "safe".into(),
                step: 0,
                next_step: 1,
                messages: json!([]),
                provider_request_id: None,
                tool_call_id: None,
                attempt: lease.attempt,
            },
            "worker-1",
            "token-1",
        )
        .expect("save");
    let completed = store
        .complete_run_with_lease(
            "checkpoint-terminal",
            "worker-1",
            "token-1",
            Some(json!({"ok": true})),
        )
        .expect("complete");
    assert_eq!(completed.status, "completed");
    assert!(
        store
            .read_checkpoint("checkpoint-terminal")
            .expect("read")
            .is_none()
    );
}

#[test]
fn unclaimed_cancel_transition_clears_checkpoint_atomically() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "unclaimed-cancel-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");

    // safe 游标被重新排队后，控制端可以在没有 worker lease 的情况下直接收口。
    store
        .create_run(NewRun {
            id: "queued-safe-cancel".into(),
            session_id: "unclaimed-cancel-session".into(),
            status: "queued".into(),
            input: json!({"message": "queued safe cancel"}),
        })
        .expect("create queued run");
    store
        .claim_run_with_lease(
            "queued-safe-cancel",
            "worker-queued",
            "token-queued",
            Duration::from_secs(30),
        )
        .expect("claim queued run");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: "queued-safe-cancel".into(),
                phase: "safe".into(),
                step: 0,
                next_step: 1,
                messages: json!([]),
                provider_request_id: None,
                tool_call_id: None,
                attempt: 1,
            },
            "worker-queued",
            "token-queued",
        )
        .expect("save safe checkpoint");
    store
        .release_run_lease("queued-safe-cancel", "worker-queued", "token-queued")
        .expect("release to reconciliation");
    store
        .requeue_reconciled_run("queued-safe-cancel")
        .expect("requeue safe run");
    store
        .request_cancel("queued-safe-cancel")
        .expect("request queued cancel");
    let cancelled = store
        .mark_cancelled("queued-safe-cancel", None)
        .expect("cancel queued run");
    assert_eq!(cancelled.status, "cancelled");
    assert!(
        store
            .read_checkpoint("queued-safe-cancel")
            .expect("read queued checkpoint")
            .is_none()
    );

    // reconciling 中的未知外部调用也可以被控制端终止；终态不能遗留游标。
    store
        .create_run(NewRun {
            id: "reconciling-cancel".into(),
            session_id: "unclaimed-cancel-session".into(),
            status: "queued".into(),
            input: json!({"message": "reconciling cancel"}),
        })
        .expect("create reconciling run");
    store
        .claim_run_with_lease(
            "reconciling-cancel",
            "worker-reconciling",
            "token-reconciling",
            Duration::from_secs(30),
        )
        .expect("claim reconciling run");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: "reconciling-cancel".into(),
                phase: "provider_in_flight".into(),
                step: 0,
                next_step: 0,
                messages: json!([]),
                provider_request_id: Some("provider-request".into()),
                tool_call_id: None,
                attempt: 1,
            },
            "worker-reconciling",
            "token-reconciling",
        )
        .expect("save provider checkpoint");
    store
        .release_run_lease(
            "reconciling-cancel",
            "worker-reconciling",
            "token-reconciling",
        )
        .expect("release provider run");
    store
        .request_cancel("reconciling-cancel")
        .expect("request reconciling cancel");
    let cancelled = store
        .mark_cancelled("reconciling-cancel", Some(json!({"reason": "user"})))
        .expect("cancel reconciling run");
    assert_eq!(cancelled.status, "cancelled");
    assert!(
        store
            .read_checkpoint("reconciling-cancel")
            .expect("read reconciling checkpoint")
            .is_none()
    );
}

#[test]
fn all_unleased_terminal_entrypoints_clear_checkpoint_atomically() {
    for (run_id, terminal) in [
        ("unleased-complete", "complete"),
        ("unleased-fail", "fail"),
        ("unleased-update", "update"),
    ] {
        let store = SqliteStore::open_in_memory().expect("open sqlite");
        store
            .create_session(NewSession {
                id: format!("{run_id}-session"),
                agent_id: None,
                status: "active".into(),
                metadata: json!({}),
            })
            .expect("create session");
        store
            .create_run(NewRun {
                id: run_id.into(),
                session_id: format!("{run_id}-session"),
                status: "queued".into(),
                input: json!({"message": "terminal cleanup"}),
            })
            .expect("create run");
        store
            .claim_run_with_lease(run_id, "worker", "token", Duration::from_secs(30))
            .expect("claim run");
        store
            .save_checkpoint_with_lease(
                NewCheckpoint {
                    run_id: run_id.into(),
                    phase: "safe".into(),
                    step: 0,
                    next_step: 1,
                    messages: json!([]),
                    provider_request_id: None,
                    tool_call_id: None,
                    attempt: 1,
                },
                "worker",
                "token",
            )
            .expect("save checkpoint");
        store
            .release_run_lease(run_id, "worker", "token")
            .expect("release lease");

        // 构造历史无 lease running 记录，验证三个公开兼容入口都维持相同
        // 终态不变量。非终态 update 不应提前删掉恢复游标。
        store
            .update_run(run_id, "running", None)
            .expect("restore legacy running state");
        assert!(store.read_checkpoint(run_id).unwrap().is_some());
        let record = match terminal {
            "complete" => store.complete_run(run_id, None).expect("complete run"),
            "fail" => store.fail_run(run_id, None).expect("fail run"),
            "update" => store
                .update_run(run_id, "cancelled", None)
                .expect("update terminal run"),
            _ => unreachable!(),
        };
        assert!(matches!(
            record.status.as_str(),
            "completed" | "failed" | "cancelled"
        ));
        assert!(store.read_checkpoint(run_id).unwrap().is_none());
    }
}

#[test]
fn external_session_lifecycle_can_update_id_and_status_in_place() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "external-lifecycle-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "external-lifecycle-run".into(),
            session_id: "external-lifecycle-session".into(),
            status: "running".into(),
            input: json!({"message": "lifecycle"}),
        })
        .expect("create run");
    store
        .upsert_external_session(NewExternalSession {
            id: "external-lifecycle-record".into(),
            session_id: "external-lifecycle-session".into(),
            run_id: Some("external-lifecycle-run".into()),
            backend: "fixture".into(),
            external_id: "request-1".into(),
            status: "running".into(),
            metadata: json!({"lifecycle": "running"}),
        })
        .expect("record dispatch");

    let updated = store
        .update_external_session(
            "external-lifecycle-record",
            "turn-1",
            "completed",
            json!({"lifecycle": "completed", "externalIdKnown": true}),
        )
        .expect("update lifecycle");
    assert_eq!(updated.id, "external-lifecycle-record");
    assert_eq!(updated.external_id, "turn-1");
    assert_eq!(updated.status, "completed");
    assert_eq!(updated.metadata["externalIdKnown"], true);
    assert_eq!(
        store
            .get_external_session("external-lifecycle-record")
            .expect("read lifecycle")
            .unwrap()
            .external_id,
        "turn-1"
    );
}

#[test]
fn checkpoint_phase_is_bounded() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    let (_, run_id) = store_with_run();
    let error = store.save_run_checkpoint(
        NewCheckpoint {
            run_id,
            phase: "unknown".into(),
            step: 0,
            next_step: 0,
            messages: json!([]),
            provider_request_id: None,
            tool_call_id: None,
            attempt: 0,
        },
        "worker",
        "token",
    );
    assert!(matches!(error, Err(StorageError::InvalidInput(message)) if message.contains("phase")));
}

#[test]
fn compacting_checkpoint_is_a_durable_non_safe_boundary() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "compaction-session".into(),
            agent_id: Some("agent-1".into()),
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "compaction-run".into(),
            session_id: "compaction-session".into(),
            status: "queued".into(),
            input: json!({"message": "compact"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(
            "compaction-run",
            "compaction-worker",
            "compaction-token",
            Duration::from_secs(5),
        )
        .expect("claim run");
    let checkpoint = store
        .save_run_checkpoint(
            NewCheckpoint {
                run_id: "compaction-run".into(),
                phase: "compacting".into(),
                step: 0,
                next_step: 0,
                messages: json!([user_message("history")]),
                provider_request_id: Some("compaction-request-0".into()),
                tool_call_id: None,
                attempt: lease.attempt,
            },
            "compaction-worker",
            "compaction-token",
        )
        .expect("save compacting checkpoint");
    assert_eq!(checkpoint.phase, "compacting");
    assert_eq!(
        store
            .read_checkpoint("compaction-run")
            .unwrap()
            .unwrap()
            .phase,
        "compacting"
    );
    // A compaction cursor is intentionally not accepted by the safe-resume
    // path; callers must reconcile or explicitly terminate it first.
    let error = store
        .resume_safe_run("compaction-run")
        .expect_err("compacting must not be treated as safe");
    assert!(matches!(error, StorageError::InvalidInput(message) if message.contains("安全恢复")));
}

#[test]
fn reconciliation_result_marks_provider_and_tool_checkpoints_safe() {
    let before = user_message("before");
    let provider_messages = json!([
        before.clone(),
        {"role":"assistant","content":[{"type":"text","text":"provider result"}]}
    ]);
    let (store, run_id, attempt) = reconciling_checkpoint(
        "provider_in_flight",
        json!([before]),
        Some("provider-request-1"),
        None,
    );
    let safe = store
        .record_reconciliation_result(
            &run_id,
            "provider_in_flight",
            "provider-request-1",
            0,
            attempt,
            provider_messages,
        )
        .expect("record provider result");
    assert_eq!(safe.phase, "safe");
    assert_eq!(safe.step, 0);
    assert_eq!(safe.next_step, 1);
    assert_eq!(
        safe.provider_request_id.as_deref(),
        Some("provider-request-1")
    );
    assert_eq!(safe.tool_call_id, None);
    assert_eq!(
        store.get_run(&run_id).expect("read run").unwrap().status,
        "reconciling"
    );

    let call_id = "tool-call-1";
    let tool_before = json!([user_message("before"), assistant_tool_call_message(call_id)]);
    let tool_messages = json!([
        user_message("before"),
        assistant_tool_call_message(call_id),
        tool_result_message(call_id)
    ]);
    let (store, run_id, attempt) = reconciling_checkpoint(
        "tool_in_flight",
        tool_before,
        Some("provider-request-2"),
        Some(call_id),
    );
    let safe = store
        .record_reconciliation_result(
            &run_id,
            "tool_in_flight",
            call_id,
            0,
            attempt,
            tool_messages,
        )
        .expect("record tool result");
    assert_eq!(safe.phase, "safe");
    assert_eq!(safe.next_step, 1);
    assert_eq!(
        safe.provider_request_id.as_deref(),
        Some("provider-request-2")
    );
    // safe checkpoint 表示 tool 已经完成对账；恢复器要求这里不再暴露 in-flight call。
    assert_eq!(safe.tool_call_id, None);
}

#[test]
fn reconciliation_result_rejects_identity_cursor_phase_and_status_mismatch() {
    let before = user_message("before");
    let after = json!([
        before.clone(),
        {"role":"assistant","content":[{"type":"text","text":"result"}]}
    ]);
    let (store, run_id, attempt) = reconciling_checkpoint(
        "provider_in_flight",
        json!([before]),
        Some("request-1"),
        None,
    );

    for (phase, external_id, step, supplied_attempt) in [
        ("safe", "request-1", 0, attempt),
        ("provider_in_flight", "wrong-request", 0, attempt),
        ("provider_in_flight", "request-1", 1, attempt),
        ("provider_in_flight", "request-1", 0, attempt + 1),
    ] {
        let error = store.record_reconciliation_result(
            &run_id,
            phase,
            external_id,
            step,
            supplied_attempt,
            after.clone(),
        );
        assert!(matches!(error, Err(StorageError::InvalidInput(_))));
        assert_eq!(
            store
                .read_checkpoint(&run_id)
                .expect("read checkpoint")
                .expect("checkpoint")
                .phase,
            "provider_in_flight"
        );
    }

    // 只有 reconciling run 才允许写回外部已确认结果。
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "running-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "running-run".into(),
            session_id: "running-session".into(),
            status: "running".into(),
            input: json!({"message": "running"}),
        })
        .expect("create run");
    let error = store.record_reconciliation_result(
        "running-run",
        "provider_in_flight",
        "request-1",
        0,
        1,
        after,
    );
    assert!(
        matches!(error, Err(StorageError::InvalidInput(message)) if message.contains("待对账"))
    );
}

#[test]
fn reconciliation_result_rejects_invalid_or_incomplete_messages() {
    let before = user_message("before");
    let (store, run_id, attempt) = reconciling_checkpoint(
        "provider_in_flight",
        json!([before.clone()]),
        Some("request-1"),
        None,
    );
    let invalid_messages = [
        json!({"role":"assistant"}),
        json!([]),
        json!([{"role":"user","content":"not-an-array"}]),
        json!([{"role":"user","content":[{"type":"text","text":"changed"}]}]),
        json!([before.clone()]),
    ];
    for messages in invalid_messages {
        let error = store.record_reconciliation_result(
            &run_id,
            "provider_in_flight",
            "request-1",
            0,
            attempt,
            messages,
        );
        assert!(matches!(error, Err(StorageError::InvalidInput(_))));
        assert_eq!(
            store
                .read_checkpoint(&run_id)
                .expect("read checkpoint")
                .expect("checkpoint")
                .phase,
            "provider_in_flight"
        );
    }

    let call_id = "tool-call-2";
    let (store, run_id, attempt) = reconciling_checkpoint(
        "tool_in_flight",
        json!([assistant_tool_call_message(call_id)]),
        None,
        Some(call_id),
    );
    let missing_tool_result = json!([
        assistant_tool_call_message(call_id),
        {"role":"assistant","content":[{"type":"text","text":"not a tool result"}]}
    ]);
    let error = store.record_reconciliation_result(
        &run_id,
        "tool_in_flight",
        call_id,
        0,
        attempt,
        missing_tool_result,
    );
    assert!(
        matches!(error, Err(StorageError::InvalidInput(message)) if message.contains("tool result"))
    );
}

#[test]
fn file_database_uses_wal() {
    let directory = local_tempdir();
    let path = directory.path().join("agent.db");
    let store = SqliteStore::open(&path).expect("open file db");
    assert_eq!(store.journal_mode().expect("journal mode"), "wal");
}

#[test]
fn append_event_enforces_expected_revision() {
    let (store, run_id) = store_with_run();
    let event = store
        .append_event(
            NewEvent {
                id: "event-1".into(),
                run_id: run_id.clone(),
                event_type: "message.completed".into(),
                payload: json!({"text": "done"}),
            },
            0,
        )
        .expect("append event");
    assert_eq!(event.revision, 1);
    assert_eq!(
        store.get_run(&run_id).expect("read run").unwrap().revision,
        1
    );
    let conflict = store.append_event(
        NewEvent {
            id: "event-2".into(),
            run_id: run_id.clone(),
            event_type: "message.completed".into(),
            payload: json!({"text": "stale"}),
        },
        0,
    );
    assert!(matches!(
        conflict,
        Err(StorageError::RevisionConflict { .. })
    ));
    assert_eq!(store.list_events(&run_id, 0).expect("events").len(), 1);
}

#[test]
fn snapshot_attachments_and_jsonl_export_work() {
    let (store, run_id) = store_with_run();
    store
        .append_event(
            NewEvent {
                id: "event-1".into(),
                run_id: run_id.clone(),
                event_type: "turn.started".into(),
                payload: json!({"step": 1}),
            },
            0,
        )
        .expect("event");
    let snapshot = store
        .save_snapshot(NewSnapshot {
            session_id: "session-1".into(),
            run_id: run_id.clone(),
            revision: 1,
            state: json!({"status": "running"}),
        })
        .expect("snapshot");
    assert_eq!(
        store.latest_snapshot(&run_id).expect("snapshot").unwrap(),
        snapshot
    );
    store
        .upsert_approval(NewApproval {
            id: "approval-1".into(),
            session_id: "session-1".into(),
            run_id: run_id.clone(),
            tool_call_id: Some("tool-1".into()),
            status: "pending".into(),
            request: json!({
                "requestId": "approval-1",
                "runId": run_id,
                "call": {
                    "id": "tool-1",
                    "name": "file.read",
                    "arguments": {"path": "README.md"}
                },
                "argumentsHash": "hash-tool-1",
                "approvalToken": "token-1",
                "expiresAtMs": i64::MAX
            }),
            arguments_hash: "hash-tool-1".into(),
            approval_token: "token-1".into(),
            expires_at_ms: i64::MAX,
        })
        .expect("approval");
    store
        .insert_tool_call(NewToolCall {
            id: "tool-1".into(),
            session_id: "session-1".into(),
            run_id: run_id.clone(),
            tool_name: "file.read".into(),
            arguments: json!({"path": "README.md"}),
            status: "completed".into(),
        })
        .expect("tool call");
    let duplicate = store
        .insert_tool_call(NewToolCall {
            id: "tool-1".into(),
            session_id: "session-1".into(),
            run_id: run_id.clone(),
            tool_name: "file.read".into(),
            arguments: json!({"path": "README.md"}),
            status: "requested".into(),
        })
        .expect("identical tool call insert is idempotent");
    assert_eq!(duplicate.status, "completed");
    assert_eq!(store.list_tool_calls_for_run(&run_id).unwrap().len(), 1);
    store
        .upsert_external_session(NewExternalSession {
            id: "external-1".into(),
            session_id: "session-1".into(),
            run_id: Some(run_id.clone()),
            backend: "codex-app-server".into(),
            external_id: "turn-1".into(),
            status: "active".into(),
            metadata: json!({"auth_ref": "OPENAI_API_KEY"}),
        })
        .expect("external session");

    let mut output = Vec::new();
    let lines = store.export_jsonl(&run_id, &mut output).expect("export");
    assert_eq!(lines, 6);
    assert_eq!(output.iter().filter(|byte| **byte == b'\n').count(), lines);
    let exported = String::from_utf8(output).expect("utf8");
    assert!(exported.contains("\"kind\":\"run\""));
    assert!(exported.contains("message"));
    // 只保存了引用名称，不会把引用解析成密钥值写入数据库。
    assert!(exported.contains("auth_ref"));

    // JSONL 是诊断/迁移边界，不应把可直接用于恢复审批的 bearer token
    // 带出；审批自身的稳定身份仍要保留，方便离线审计定位。
    assert!(!exported.contains("token-1"));
    let approval_line = exported
        .lines()
        .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("jsonl line"))
        .find(|line| line.get("kind").and_then(serde_json::Value::as_str) == Some("approval"))
        .expect("approval export line");
    let approval_record = approval_line.get("record").expect("approval record");
    assert_eq!(
        approval_record
            .get("id")
            .and_then(serde_json::Value::as_str),
        Some("approval-1")
    );
    assert_eq!(
        approval_record
            .get("tool_call_id")
            .and_then(serde_json::Value::as_str),
        Some("tool-1")
    );
    assert_eq!(
        approval_record
            .get("status")
            .and_then(serde_json::Value::as_str),
        Some("pending")
    );
    assert!(approval_record.get("approval_token").is_none());
    assert!(
        approval_record
            .get("request")
            .and_then(|request| request.get("approvalToken"))
            .is_none()
    );

    // Export must not mutate the durable approval used by an explicit resume.
    let persisted = store
        .get_approval("approval-1")
        .expect("approval after export")
        .expect("persisted approval");
    assert_eq!(persisted.approval_token, "token-1");
    assert_eq!(
        persisted
            .request
            .get("approvalToken")
            .and_then(serde_json::Value::as_str),
        Some("token-1")
    );
}

#[test]
fn sensitive_json_is_rejected_before_write() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    let result = store.create_session(NewSession {
        id: "sensitive".into(),
        agent_id: None,
        status: "new".into(),
        metadata: json!({"provider": {"api_key": "do-not-save"}}),
    });
    assert!(matches!(
        result,
        Err(StorageError::SensitiveDataRejected { .. })
    ));
    assert!(store.get_session("sensitive").expect("read").is_none());
}

#[test]
fn snapshot_cannot_be_ahead_of_run() {
    let (store, run_id) = store_with_run();
    let result = store.save_snapshot(NewSnapshot {
        session_id: "session-1".into(),
        run_id,
        revision: 1,
        state: json!({}),
    });
    assert!(matches!(result, Err(StorageError::SnapshotAhead { .. })));
}

#[test]
fn run_and_session_status_updates_are_persisted() {
    let (store, run_id) = store_with_run();
    let run = store
        .update_run(&run_id, "completed", Some(json!({"answer": "ok"})))
        .expect("update run");
    assert_eq!(run.status, "completed");
    assert_eq!(run.output, Some(json!({"answer": "ok"})));
    let session = store
        .update_session("session-1", "idle", None)
        .expect("update session");
    assert_eq!(session.status, "idle");
}

#[test]
fn cancel_request_is_cas_idempotent_and_terminal_safe() {
    let (store, run_id) = store_with_run();
    let initial = store.get_run(&run_id).expect("read run").unwrap();
    assert!(!initial.cancel_requested);
    assert!(!store.is_cancel_requested(&run_id).expect("cancel flag"));

    let requested = store.request_cancel(&run_id).expect("request cancel");
    assert_eq!(requested.status, "cancel_requested");
    assert!(requested.cancel_requested);
    assert!(store.is_cancel_requested(&run_id).expect("cancel flag"));
    // 第二次请求不更新时间戳，也不产生新的状态变化。
    assert_eq!(
        store.request_cancel(&run_id).expect("repeat cancel"),
        requested
    );

    let cancelled = store
        .mark_cancelled(&run_id, Some(json!({"reason": "user"})))
        .expect("mark cancelled");
    assert_eq!(cancelled.status, "cancelled");
    assert!(cancelled.cancel_requested);
    // 已经 cancelled 的收束和取消请求都保持幂等。
    assert_eq!(
        store.mark_cancelled(&run_id, None).expect("repeat mark"),
        cancelled
    );
    assert_eq!(
        store.request_cancel(&run_id).expect("repeat request"),
        cancelled
    );

    let active_store = SqliteStore::open_in_memory().expect("open active store");
    active_store
        .create_session(NewSession {
            id: "active-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create active session");
    active_store
        .create_run(NewRun {
            id: "active-run".into(),
            session_id: "active-session".into(),
            status: "running".into(),
            input: json!({"message": "complete"}),
        })
        .expect("create active run");
    let completed = active_store
        .complete_run("active-run", Some(json!({"answer": "ok"})))
        .expect("complete active run");
    assert_eq!(completed.status, "completed");
    // 完成收束同样是幂等的；重复完成不会覆盖已有结果。
    assert_eq!(
        active_store
            .complete_run("active-run", Some(json!({"answer": "changed"})))
            .expect("repeat complete"),
        completed
    );
    assert!(matches!(
        active_store.request_cancel("active-run"),
        Err(StorageError::TerminalRun { .. })
    ));

    let failed_store = SqliteStore::open_in_memory().expect("open failed store");
    failed_store
        .create_session(NewSession {
            id: "failed-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create failed session");
    failed_store
        .create_run(NewRun {
            id: "failed-run".into(),
            session_id: "failed-session".into(),
            status: "running".into(),
            input: json!({"message": "fail"}),
        })
        .expect("create failed run");
    let failed = failed_store
        .fail_run("failed-run", Some(json!({"error": "boom"})))
        .expect("fail run");
    assert_eq!(failed.status, "failed");
    assert_eq!(
        failed_store
            .fail_run("failed-run", Some(json!({"error": "changed"})))
            .expect("repeat fail"),
        failed
    );

    let fail_cancel_race_store = SqliteStore::open_in_memory().expect("open fail race store");
    fail_cancel_race_store
        .create_session(NewSession {
            id: "fail-cancel-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create fail cancel session");
    fail_cancel_race_store
        .create_run(NewRun {
            id: "fail-cancel-run".into(),
            session_id: "fail-cancel-session".into(),
            status: "running".into(),
            input: json!({"message": "race"}),
        })
        .expect("create fail cancel run");
    fail_cancel_race_store
        .request_cancel("fail-cancel-run")
        .expect("request fail cancel");
    assert!(
        fail_cancel_race_store
            .fail_run("fail-cancel-run", Some(json!({"error": "late"})))
            .is_err()
    );
    assert_eq!(
        fail_cancel_race_store
            .get_run("fail-cancel-run")
            .expect("read fail cancel run")
            .unwrap()
            .status,
        "cancel_requested"
    );

    let cancelled_store = SqliteStore::open_in_memory().expect("open cancelled store");
    cancelled_store
        .create_session(NewSession {
            id: "cancel-race-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create cancel race session");
    cancelled_store
        .create_run(NewRun {
            id: "cancel-race-run".into(),
            session_id: "cancel-race-session".into(),
            status: "running".into(),
            input: json!({"message": "race"}),
        })
        .expect("create cancel race run");
    cancelled_store
        .request_cancel("cancel-race-run")
        .expect("request race cancel");
    assert!(
        cancelled_store
            .complete_run("cancel-race-run", Some(json!({"answer": "late"})))
            .is_err()
    );
    assert_eq!(
        cancelled_store
            .get_run("cancel-race-run")
            .expect("read cancel race run")
            .unwrap()
            .status,
        "cancel_requested"
    );

    let terminal_store = SqliteStore::open_in_memory().expect("open sqlite");
    terminal_store
        .create_session(NewSession {
            id: "terminal-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create terminal session");
    terminal_store
        .create_run(NewRun {
            id: "terminal-run".into(),
            session_id: "terminal-session".into(),
            status: "completed".into(),
            input: json!({"message": "done"}),
        })
        .expect("create terminal run");
    assert!(matches!(
        terminal_store.request_cancel("terminal-run"),
        Err(StorageError::TerminalRun { .. })
    ));
    assert!(matches!(
        terminal_store.mark_cancelled("terminal-run", None),
        Err(StorageError::TerminalRun { .. })
    ));
    assert!(
        !terminal_store
            .get_run("terminal-run")
            .expect("read terminal run")
            .unwrap()
            .cancel_requested
    );
}

#[test]
fn claim_run_allows_one_queued_worker() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "queue-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create queue session");
    store
        .create_run(NewRun {
            id: "queue-run".into(),
            session_id: "queue-session".into(),
            status: "queued".into(),
            input: json!({"message": "queued"}),
        })
        .expect("create queue run");

    let claimed = store.claim_run("queue-run").expect("claim run");
    assert_eq!(claimed.status, "running");
    assert!(!claimed.cancel_requested);
    assert!(store.claim_run("queue-run").is_err());

    store.request_cancel("queue-run").expect("request cancel");
    assert!(store.claim_run("queue-run").is_err());
}

#[test]
fn worker_lease_fences_writes_and_tracks_attempt() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "lease-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "lease-run".into(),
            session_id: "lease-session".into(),
            status: "queued".into(),
            input: json!({"message": "lease"}),
        })
        .expect("create run");

    let (claimed, lease) = store
        .claim_run_with_lease("lease-run", "worker-a", "lease-a", Duration::from_secs(30))
        .expect("claim with lease");
    assert_eq!(claimed.status, "running");
    assert_eq!(lease.run_id, "lease-run");
    assert_eq!(lease.worker_id, "worker-a");
    assert_eq!(lease.lease_token, "lease-a");
    assert_eq!(lease.attempt, 1);
    assert_eq!(
        store.get_run_lease("lease-run").expect("read lease"),
        Some(lease.clone())
    );

    assert!(matches!(
        store.claim_run_with_lease("lease-run", "worker-b", "lease-b", Duration::from_secs(30)),
        Err(StorageError::LeaseConflict { .. })
    ));
    assert!(matches!(
        store.heartbeat_run("lease-run", "worker-b", "lease-a", Duration::from_secs(30)),
        Err(StorageError::LeaseConflict { .. }) | Err(StorageError::LeaseLost { .. })
    ));
    // 无 lease 的旧收束 API 不能绕过 fencing。
    assert!(matches!(
        store.complete_run("lease-run", Some(json!({"answer": "late"}))),
        Err(StorageError::LeaseConflict { .. })
    ));

    let renewed = store
        .heartbeat_run("lease-run", "worker-a", "lease-a", Duration::from_secs(30))
        .expect("heartbeat");
    assert!(renewed.lease_expires_at >= lease.lease_expires_at);
    let completed = store
        .complete_run_with_lease(
            "lease-run",
            "worker-a",
            "lease-a",
            Some(json!({"answer": "ok"})),
        )
        .expect("complete with lease");
    assert_eq!(completed.status, "completed");
    assert_eq!(store.get_run_lease("lease-run").expect("read lease"), None);
    // 终态幂等，不会被迟到结果覆盖。
    assert_eq!(
        store
            .complete_run_with_lease(
                "lease-run",
                "worker-a",
                "lease-a",
                Some(json!({"answer": "changed"}))
            )
            .expect("repeat complete"),
        completed
    );
}

#[test]
fn two_worker_handles_are_fenced_by_the_same_sqlite_lease() {
    let directory = local_tempdir();
    let path = directory.path().join("two-workers.db");
    let worker_a = SqliteStore::open(&path).expect("open worker A store");
    worker_a
        .create_session(NewSession {
            id: "two-worker-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    worker_a
        .create_run(NewRun {
            id: "two-worker-run".into(),
            session_id: "two-worker-session".into(),
            status: "queued".into(),
            input: json!({"message": "fence"}),
        })
        .expect("create run");

    // 两个独立连接代表两个进程；SQLite CAS 仍只允许一个 owner。
    let worker_b = SqliteStore::open(&path).expect("open worker B store");
    let (_, lease_a) = worker_a
        .claim_run_with_lease(
            "two-worker-run",
            "worker-a",
            "token-a",
            Duration::from_secs(30),
        )
        .expect("worker A claim");
    assert_eq!(
        worker_b
            .get_run_lease("two-worker-run")
            .expect("read shared lease"),
        Some(lease_a.clone())
    );
    assert!(matches!(
        worker_b.claim_run_with_lease(
            "two-worker-run",
            "worker-b",
            "token-b",
            Duration::from_secs(30)
        ),
        Err(StorageError::LeaseConflict { .. })
    ));

    // 未持有 token 的 worker 不能通过任一收束入口写终态。
    assert!(matches!(
        worker_b.complete_run_with_lease(
            "two-worker-run",
            "worker-b",
            "token-b",
            Some(json!({"answer": "stale"}))
        ),
        Err(StorageError::LeaseConflict { .. }) | Err(StorageError::LeaseLost { .. })
    ));
    assert!(matches!(
        worker_b.fail_run_with_lease(
            "two-worker-run",
            "worker-b",
            "token-b",
            Some(json!({"error": "stale"}))
        ),
        Err(StorageError::LeaseConflict { .. }) | Err(StorageError::LeaseLost { .. })
    ));
    assert!(matches!(
        worker_b.mark_cancelled_with_lease(
            "two-worker-run",
            "worker-b",
            "token-b",
            Some(json!({"reason": "stale"}))
        ),
        Err(StorageError::LeaseConflict { .. }) | Err(StorageError::LeaseLost { .. })
    ));
    assert_eq!(
        worker_b
            .get_run("two-worker-run")
            .expect("read fenced run")
            .unwrap()
            .status,
        "running"
    );

    // 当前 owner 仍可续租并完成，证明拒绝旧 worker 没有破坏有效 lease。
    worker_a
        .heartbeat_run(
            "two-worker-run",
            "worker-a",
            &lease_a.lease_token,
            Duration::from_secs(30),
        )
        .expect("worker A heartbeat");
    let completed = worker_a
        .complete_run_with_lease(
            "two-worker-run",
            "worker-a",
            &lease_a.lease_token,
            Some(json!({"answer": "owner"})),
        )
        .expect("worker A complete");
    assert_eq!(completed.status, "completed");
    assert_eq!(completed.output, Some(json!({"answer": "owner"})));
}

#[test]
fn heartbeat_renews_lease_and_blocks_expiry_recovery() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "heartbeat-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "heartbeat-run".into(),
            session_id: "heartbeat-session".into(),
            status: "queued".into(),
            input: json!({"message": "heartbeat"}),
        })
        .expect("create run");

    let (_, initial) = store
        .claim_run_with_lease(
            "heartbeat-run",
            "worker-heartbeat",
            "heartbeat-token",
            Duration::from_secs(2),
        )
        .expect("claim");
    let renewed = store
        .heartbeat_run(
            "heartbeat-run",
            "worker-heartbeat",
            "heartbeat-token",
            Duration::from_secs(30),
        )
        .expect("renew lease");
    assert!(renewed.heartbeat_at >= initial.heartbeat_at);
    assert!(renewed.lease_expires_at > initial.lease_expires_at);
    assert!(matches!(
        store.recover_expired_run("heartbeat-run"),
        Err(StorageError::LeaseConflict { .. })
    ));
    assert!(matches!(
        store.heartbeat_run(
            "heartbeat-run",
            "other-worker",
            "heartbeat-token",
            Duration::from_secs(30)
        ),
        Err(StorageError::LeaseConflict { .. }) | Err(StorageError::LeaseLost { .. })
    ));

    let completed = store
        .complete_run_with_lease(
            "heartbeat-run",
            "worker-heartbeat",
            "heartbeat-token",
            Some(json!({"answer": "alive"})),
        )
        .expect("complete after heartbeat");
    assert_eq!(completed.status, "completed");
}

#[test]
fn expired_token_cannot_write_terminal_state_before_reconciliation() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "expired-token-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "expired-token-run".into(),
            session_id: "expired-token-session".into(),
            status: "queued".into(),
            input: json!({"message": "expired"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(
            "expired-token-run",
            "worker-old",
            "token-old",
            Duration::from_millis(1),
        )
        .expect("claim short lease");
    wait_until_epoch_ms(lease.lease_expires_at);

    // 到期 token 的三种终态写回都必须失败，且 run 仍保持 running，等待对账。
    assert!(matches!(
        store.complete_run_with_lease(
            "expired-token-run",
            "worker-old",
            "token-old",
            Some(json!({"answer": "late"}))
        ),
        Err(StorageError::LeaseLost { .. }) | Err(StorageError::LeaseConflict { .. })
    ));
    assert!(matches!(
        store.fail_run_with_lease(
            "expired-token-run",
            "worker-old",
            "token-old",
            Some(json!({"error": "late"}))
        ),
        Err(StorageError::LeaseLost { .. }) | Err(StorageError::LeaseConflict { .. })
    ));
    assert!(matches!(
        store.mark_cancelled_with_lease(
            "expired-token-run",
            "worker-old",
            "token-old",
            Some(json!({"reason": "late"}))
        ),
        Err(StorageError::LeaseLost { .. }) | Err(StorageError::LeaseConflict { .. })
    ));
    assert_eq!(
        store
            .get_run("expired-token-run")
            .expect("read expired run")
            .unwrap()
            .status,
        "running"
    );

    // 对账动作只清理 fencing 信息并进入 reconciling，不替旧 worker 重放调用。
    let reconciled = store
        .recover_expired_run("expired-token-run")
        .expect("reconcile expired run");
    assert_eq!(reconciled.status, "reconciling");
    assert_eq!(
        store.get_run_lease("expired-token-run").expect("lease"),
        None
    );
    assert!(matches!(
        store.complete_run_with_lease(
            "expired-token-run",
            "worker-old",
            "token-old",
            Some(json!({"answer": "after-reconcile"}))
        ),
        Err(StorageError::LeaseLost { .. }) | Err(StorageError::LeaseConflict { .. })
    ));
}

#[test]
fn expired_lease_only_enters_reconciling_and_never_replays() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "recover-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "recover-run".into(),
            session_id: "recover-session".into(),
            status: "queued".into(),
            input: json!({"message": "recover"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(
            "recover-run",
            "worker-a",
            "lease-a",
            Duration::from_millis(1),
        )
        .expect("claim short lease");
    wait_until_epoch_ms(lease.lease_expires_at);

    let recovered = store
        .recover_expired_run("recover-run")
        .expect("recover expired run");
    assert_eq!(recovered.status, "reconciling");
    assert_eq!(
        store.get_run_lease("recover-run").expect("read lease"),
        None
    );
    assert!(matches!(
        store.heartbeat_run(
            "recover-run",
            "worker-a",
            "lease-a",
            Duration::from_secs(30)
        ),
        Err(StorageError::LeaseLost { .. }) | Err(StorageError::LeaseConflict { .. })
    ));
    // 重复恢复是幂等读，不会创建第二个 attempt 或重新领取。
    assert_eq!(
        store
            .recover_expired_run("recover-run")
            .expect("repeat recovery"),
        recovered
    );
    assert!(matches!(
        store.claim_run_with_lease(
            "recover-run",
            "worker-b",
            "lease-b",
            Duration::from_secs(30)
        ),
        Err(StorageError::LeaseConflict { .. })
    ));
}

#[test]
fn cancel_requested_lease_can_heartbeat_and_finalize() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "cancel-lease-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "cancel-lease-run".into(),
            session_id: "cancel-lease-session".into(),
            status: "queued".into(),
            input: json!({"message": "cancel"}),
        })
        .expect("create run");
    store
        .claim_run_with_lease(
            "cancel-lease-run",
            "worker-a",
            "lease-a",
            Duration::from_secs(30),
        )
        .expect("claim");
    store
        .request_cancel("cancel-lease-run")
        .expect("request cancel");
    store
        .heartbeat_run(
            "cancel-lease-run",
            "worker-a",
            "lease-a",
            Duration::from_secs(30),
        )
        .expect("heartbeat while cancelling");
    let cancelled = store
        .mark_cancelled_with_lease(
            "cancel-lease-run",
            "worker-a",
            "lease-a",
            Some(json!({"reason": "user"})),
        )
        .expect("cancel with lease");
    assert_eq!(cancelled.status, "cancelled");
    assert_eq!(store.get_run_lease("cancel-lease-run").expect("read"), None);
}

#[test]
fn expired_cancel_requested_lease_enters_reconciling() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "cancel-recover-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    store
        .create_run(NewRun {
            id: "cancel-recover-run".into(),
            session_id: "cancel-recover-session".into(),
            status: "queued".into(),
            input: json!({"message": "cancel recovery"}),
        })
        .expect("create run");
    let (_, lease) = store
        .claim_run_with_lease(
            "cancel-recover-run",
            "worker-a",
            "lease-a",
            Duration::from_millis(1),
        )
        .expect("claim");
    store
        .request_cancel("cancel-recover-run")
        .expect("request cancel");
    wait_until_epoch_ms(lease.lease_expires_at);
    let recovered = store
        .recover_expired_run("cancel-recover-run")
        .expect("recover cancelled lease");
    assert_eq!(recovered.status, "reconciling");
    assert!(recovered.cancel_requested);
}

#[test]
fn existing_v1_database_receives_run_control_migration() {
    let directory = local_tempdir();
    let path = directory.path().join("v1-agent.db");
    {
        let connection = rusqlite::Connection::open(&path).expect("open old database");
        connection
            .execute_batch(include_str!("../migrations/0001_initial.sql"))
            .expect("apply v1 schema");
        connection
            .execute_batch(
                "CREATE TABLE schema_migrations (
                     version INTEGER PRIMARY KEY,
                     applied_at INTEGER NOT NULL
                 );
                 INSERT INTO schema_migrations(version, applied_at) VALUES (1, 1);",
            )
            .expect("record v1 migration");
    }

    let store = SqliteStore::open(&path).expect("upgrade old database");
    store
        .create_session(NewSession {
            id: "v1-session".into(),
            agent_id: None,
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    let run = store
        .create_run(NewRun {
            id: "v1-run".into(),
            session_id: "v1-session".into(),
            status: "running".into(),
            input: json!({"message": "old"}),
        })
        .expect("create run");
    assert!(!run.cancel_requested);
    assert_eq!(
        store
            .request_cancel("v1-run")
            .expect("cancel old run")
            .status,
        "cancel_requested"
    );
}

#[test]
fn existing_v2_database_receives_worker_lease_migration() {
    let directory = local_tempdir();
    let path = directory.path().join("v2-agent.db");
    {
        let connection = rusqlite::Connection::open(&path).expect("open v2 database");
        connection
            .execute_batch(include_str!("../migrations/0001_initial.sql"))
            .expect("apply v1 schema");
        connection
            .execute_batch(include_str!("../migrations/0002_run_control.sql"))
            .expect("apply v2 schema");
        connection
            .execute_batch(
                "CREATE TABLE schema_migrations (
                     version INTEGER PRIMARY KEY,
                     applied_at INTEGER NOT NULL
                 );
                 INSERT INTO schema_migrations(version, applied_at) VALUES (1, 1), (2, 2);
                 INSERT INTO sessions(id, agent_id, status, metadata_json, created_at, updated_at)
                     VALUES ('legacy-session', NULL, 'active', '{}', 1, 1);
                 INSERT INTO runs(id, session_id, status, revision, input_json, created_at, updated_at)
                     VALUES ('legacy-run', 'legacy-session', 'queued', 0, '{\"message\":\"legacy\"}', 1, 1);",
            )
            .expect("seed v2 records");
    }

    let store = SqliteStore::open(&path).expect("upgrade v2 database");
    let legacy = store
        .get_run("legacy-run")
        .expect("read legacy run")
        .expect("legacy run exists");
    assert_eq!(legacy.status, "queued");
    assert!(!legacy.cancel_requested);
    assert_eq!(store.get_run_lease("legacy-run").expect("lease"), None);

    // v3 字段应使用默认值，不丢失 v2 的业务记录；升级后可正常领取并递增 attempt。
    let (_, lease) = store
        .claim_run_with_lease(
            "legacy-run",
            "worker-v3",
            "token-v3",
            Duration::from_secs(30),
        )
        .expect("claim migrated run");
    assert_eq!(lease.attempt, 1);

    let connection = rusqlite::Connection::open(&path).expect("inspect upgraded database");
    let version: i64 = connection
        .query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
            row.get(0)
        })
        .expect("migration version");
    assert!(
        version >= 5,
        "runtime event schema migration was not applied: {version}"
    );
    for column in [
        "worker_id",
        "lease_token",
        "lease_expires_at",
        "heartbeat_at",
        "attempt",
    ] {
        let present: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM pragma_table_info('runs') WHERE name = ?1",
                [column],
                |row| row.get(0),
            )
            .expect("lease column lookup");
        assert_eq!(present, 1, "missing migrated column {column}");
    }
    let checkpoint_table: i64 = connection
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master
             WHERE type = 'table' AND name = 'run_checkpoints'",
            [],
            |row| row.get(0),
        )
        .expect("checkpoint table lookup");
    assert_eq!(checkpoint_table, 1, "missing migrated checkpoint table");
    let event_schema_column: i64 = connection
        .query_row(
            "SELECT COUNT(*) FROM pragma_table_info('runtime_events')
             WHERE name = 'schema_version'",
            [],
            |row| row.get(0),
        )
        .expect("runtime event schema column lookup");
    assert_eq!(
        event_schema_column, 1,
        "missing runtime event schema column"
    );
}

#[test]
fn legacy_running_without_lease_enters_reconciling() {
    // v2/旧 API 可能留下没有 fencing 信息的 running 记录；不能把它当作
    // 可安全重放，显式 recovery 应先把状态送进 reconciliation gate。
    let (store, run_id) = store_with_run();
    let recovered = store
        .recover_expired_run(&run_id)
        .expect("recover unleased running run");
    assert_eq!(recovered.status, "reconciling");
    assert!(store.get_run_lease(&run_id).expect("read lease").is_none());
}

#[test]
fn stale_run_scan_is_bounded_deterministic_and_excludes_active_work() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    for (session_id, run_id, status) in [
        ("stale-session-a", "stale-run-a", "running"),
        ("stale-session-b", "stale-run-b", "running"),
        ("expired-session", "expired-run", "queued"),
        ("active-session", "active-run", "queued"),
        ("queued-session", "queued-run", "queued"),
    ] {
        store
            .create_session(NewSession {
                id: session_id.to_owned(),
                agent_id: Some("agent-1".to_owned()),
                status: "active".to_owned(),
                metadata: json!({}),
            })
            .expect("create session");
        store
            .create_run(NewRun {
                id: run_id.to_owned(),
                session_id: session_id.to_owned(),
                status: status.to_owned(),
                input: json!({"task": run_id}),
            })
            .expect("create run");
    }
    let (_, expired_lease) = store
        .claim_run_with_lease(
            "expired-run",
            "worker-expired",
            "token-expired",
            Duration::from_secs(1),
        )
        .expect("claim expired candidate");
    let (_, active_lease) = store
        .claim_run_with_lease(
            "active-run",
            "worker-active",
            "token-active",
            Duration::from_secs(30),
        )
        .expect("claim active run");

    // 注入一个不依赖 sleep 的边界时间：expired lease 在该时刻恰好失效，
    // active lease 仍在未来；无 lease 的 legacy running 记录也应被发现。
    let first = store
        .list_stale_run_ids(MAX_STALE_RUN_SCAN_LIMIT, expired_lease.lease_expires_at)
        .expect("list stale runs");
    let second = store
        .list_stale_run_ids(MAX_STALE_RUN_SCAN_LIMIT, expired_lease.lease_expires_at)
        .expect("repeat list stale runs");
    assert_eq!(first, second, "候选排序必须稳定");
    assert_eq!(first.len(), 3);
    assert!(first.iter().any(|id| id == "expired-run"));
    assert!(first.iter().any(|id| id == "stale-run-a"));
    assert!(first.iter().any(|id| id == "stale-run-b"));
    assert!(!first.iter().any(|id| id == "active-run"));
    assert!(!first.iter().any(|id| id == "queued-run"));
    assert!(active_lease.lease_expires_at > expired_lease.lease_expires_at);

    assert_eq!(
        store
            .list_stale_run_ids(0, expired_lease.lease_expires_at)
            .expect("zero limit"),
        Vec::<String>::new()
    );
    assert!(matches!(
        store.list_stale_run_ids(MAX_STALE_RUN_SCAN_LIMIT + 1, expired_lease.lease_expires_at),
        Err(StorageError::InvalidInput(message)) if message.contains("不能超过")
    ));
}

#[test]
fn stale_run_scan_finds_expired_reconciling_lease_but_not_active_one() {
    let tempdir = local_tempdir();
    let database = tempdir.path().join("reconciling-stale-scan.db");
    let expired_run = "reconciling-expired-run";
    let active_run = "reconciling-active-run";

    // A normal release clears the lease as it enters reconciliation.  Keep the
    // token deliberately present here to model a crash between the two durable
    // recovery steps in an older worker.
    {
        let store = SqliteStore::open(&database).expect("open sqlite");
        for (session_id, run_id) in [
            ("reconciling-expired-session", expired_run),
            ("reconciling-active-session", active_run),
        ] {
            store
                .create_session(NewSession {
                    id: session_id.to_owned(),
                    agent_id: Some("agent-1".to_owned()),
                    status: "active".to_owned(),
                    metadata: json!({}),
                })
                .expect("create session");
            store
                .create_run(NewRun {
                    id: run_id.to_owned(),
                    session_id: session_id.to_owned(),
                    status: "queued".to_owned(),
                    input: json!({"task": run_id}),
                })
                .expect("create run");
        }
        store
            .claim_run_with_lease(
                expired_run,
                "reconciling-expired-worker",
                "reconciling-expired-token",
                Duration::from_millis(5),
            )
            .expect("claim expired run");
        store
            .claim_run_with_lease(
                active_run,
                "reconciling-active-worker",
                "reconciling-active-token",
                Duration::from_secs(30),
            )
            .expect("claim active run");
    }

    // Simulate the historical split state without keeping the first SQLite
    // connection open while the migration/recovery connection writes.
    let connection = rusqlite::Connection::open(&database).expect("open inspection connection");
    connection
        .execute(
            "UPDATE runs SET status = 'reconciling' WHERE id IN (?1, ?2)",
            [expired_run, active_run],
        )
        .expect("inject reconciling statuses");
    drop(connection);

    let store = SqliteStore::open(&database).expect("reopen sqlite");
    let expired_at: i64 = store
        .get_run_lease(expired_run)
        .expect("read expired lease")
        .expect("expired lease")
        .lease_expires_at;
    wait_until_epoch_ms(expired_at);

    let candidates = store
        .list_stale_run_ids(MAX_STALE_RUN_SCAN_LIMIT, expired_at)
        .expect("list stale reconciling runs");
    assert!(candidates.iter().any(|id| id == expired_run));
    assert!(!candidates.iter().any(|id| id == active_run));
}

#[test]
fn recovery_clears_only_expired_reconciling_lease_and_is_idempotent() {
    let tempdir = local_tempdir();
    let database = tempdir.path().join("reconciling-stale-recovery.db");
    let run_id = "reconciling-residual-run";

    let lease_expires_at;
    {
        let store = SqliteStore::open(&database).expect("open sqlite");
        store
            .create_session(NewSession {
                id: "reconciling-residual-session".to_owned(),
                agent_id: Some("agent-1".to_owned()),
                status: "active".to_owned(),
                metadata: json!({}),
            })
            .expect("create session");
        store
            .create_run(NewRun {
                id: run_id.to_owned(),
                session_id: "reconciling-residual-session".to_owned(),
                status: "queued".to_owned(),
                input: json!({"task": "recover residual lease"}),
            })
            .expect("create run");
        let (_, lease) = store
            .claim_run_with_lease(
                run_id,
                "reconciling-residual-worker",
                "reconciling-residual-token",
                Duration::from_millis(5),
            )
            .expect("claim run");
        lease_expires_at = lease.lease_expires_at;
    }
    let connection = rusqlite::Connection::open(&database).expect("open inspection connection");
    connection
        .execute(
            "UPDATE runs SET status = 'reconciling' WHERE id = ?1",
            [run_id],
        )
        .expect("inject reconciling status");
    drop(connection);
    wait_until_epoch_ms(lease_expires_at);

    let store = SqliteStore::open(&database).expect("reopen sqlite");
    let recovered = store
        .recover_expired_run(run_id)
        .expect("clear expired residual lease");
    assert_eq!(recovered.status, "reconciling");
    assert!(
        store
            .get_run_lease(run_id)
            .expect("read cleared lease")
            .is_none()
    );

    // A retry after the cleanup is a no-op, while the run remains behind the
    // reconciliation gate and is never silently requeued.
    let repeated = store
        .recover_expired_run(run_id)
        .expect("repeat recovery is idempotent");
    assert_eq!(repeated.status, "reconciling");
    assert!(
        store
            .list_stale_run_ids(MAX_STALE_RUN_SCAN_LIMIT, lease_expires_at)
            .expect("scan after cleanup")
            .is_empty()
    );
}

#[test]
fn runtime_id_for_run_returns_none_without_runtime_event() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    assert_eq!(
        store.runtime_id_for_run("missing-run").expect("lookup"),
        None
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn tool_call_runtime_transaction_commits_row_and_events_together() {
    let (store, runtime_id, snapshot, events, call, lease) = tool_call_runtime_fixture();
    let record = store
        .create_tool_call_with_runtime_and_lease(DurableToolCallRuntimeCommit {
            call: DurableToolCallInput {
                id: call.id().to_owned(),
                session_id: "tool-transaction-session".to_owned(),
                run_id: "tool-transaction-run".to_owned(),
                tool_name: call.name().to_owned(),
                arguments: call.arguments().clone(),
                status: "requested".to_owned(),
            },
            runtime_id: runtime_id.clone(),
            expected_runtime_revision: Some(snapshot.revision() - 1),
            snapshot: snapshot.clone(),
            events: events.clone(),
            lease: Some(lease),
        })
        .expect("atomic tool request");

    assert_eq!(record.id, call.id());
    assert_eq!(record.status, "requested");
    assert_eq!(
        store.get_tool_call(call.id()).expect("read call").unwrap(),
        record
    );
    assert_eq!(
        store.load(&runtime_id).expect("read runtime"),
        Some(snapshot)
    );
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("events")
            .len(),
        4
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn tool_call_runtime_transaction_rejects_wrong_lease_without_writing() {
    let (store, runtime_id, snapshot, events, call, _) = tool_call_runtime_fixture();
    let error = store
        .create_tool_call_with_runtime_and_lease(DurableToolCallRuntimeCommit {
            call: DurableToolCallInput {
                id: call.id().to_owned(),
                session_id: "tool-transaction-session".to_owned(),
                run_id: "tool-transaction-run".to_owned(),
                tool_name: call.name().to_owned(),
                arguments: call.arguments().clone(),
                status: "requested".to_owned(),
            },
            runtime_id: runtime_id.clone(),
            expected_runtime_revision: Some(snapshot.revision() - 1),
            snapshot: snapshot.clone(),
            events,
            lease: Some(DurableLeaseIdentity {
                worker_id: "tool-worker".to_owned(),
                lease_token: "stale-token".to_owned(),
            }),
        })
        .expect_err("stale lease must be fenced");
    assert!(matches!(
        error,
        StorageError::LeaseLost { .. } | StorageError::LeaseConflict { .. }
    ));
    assert!(store.get_tool_call(call.id()).expect("read call").is_none());
    assert_eq!(
        store
            .load(&runtime_id)
            .expect("read runtime")
            .unwrap()
            .revision(),
        3
    );
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("events")
            .len(),
        3
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn tool_call_runtime_transaction_rolls_back_on_runtime_cas_failure() {
    let (store, runtime_id, snapshot, events, call, lease) = tool_call_runtime_fixture();
    let mut stale_event = events[0].clone();
    // Keep the event structurally valid for an expected revision of 2, but
    // make the persisted runtime (revision 3) win the CAS below.
    stale_event.revision = 3;
    let mut stale_snapshot = snapshot.clone();
    stale_snapshot.revision = 3;
    let error = store
        .create_tool_call_with_runtime_and_lease(DurableToolCallRuntimeCommit {
            call: DurableToolCallInput {
                id: call.id().to_owned(),
                session_id: "tool-transaction-session".to_owned(),
                run_id: "tool-transaction-run".to_owned(),
                tool_name: call.name().to_owned(),
                arguments: call.arguments().clone(),
                status: "requested".to_owned(),
            },
            runtime_id: runtime_id.clone(),
            // The database is at revision 3, while this valid one-event suffix
            // is deliberately fenced against stale expected revision 2.
            expected_runtime_revision: Some(2),
            snapshot: stale_snapshot,
            events: vec![stale_event],
            lease: Some(lease),
        })
        .expect_err("runtime CAS must reject stale revision");
    assert!(
        matches!(error, StorageError::RevisionConflict { .. }),
        "unexpected transaction error: {error:?}"
    );
    assert!(store.get_tool_call(call.id()).expect("read call").is_none());
    assert_eq!(store.load(&runtime_id).expect("read runtime"), {
        let events = store.list_runtime_events(&runtime_id).expect("events");
        let mut initial = RuntimeSnapshot::try_new(&runtime_id).expect("initial snapshot");
        for event in &events {
            initial = reduce(&initial, event).expect("reduce persisted event");
        }
        Some(initial)
    });
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("events")
            .len(),
        3
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn complete_tool_call_runtime_transaction_backfills_legacy_row() {
    let (store, runtime_id, requested_snapshot, requested_events, call, lease) =
        tool_call_runtime_fixture();
    let result = ToolResult::success(call.id(), json!({"echo": "hello"})).expect("result");
    let result_event = RuntimeEvent::tool_result(
        &runtime_id,
        requested_snapshot.revision() + 1,
        5,
        "tool-transaction-run",
        &result,
        false,
    )
    .expect("result event");
    let completed_snapshot = reduce(&requested_snapshot, &result_event).expect("reduce result");
    let record = store
        .complete_tool_call_with_runtime_and_lease(
            DurableToolCallRuntimeCommit {
                call: DurableToolCallInput {
                    id: call.id().to_owned(),
                    session_id: "tool-transaction-session".to_owned(),
                    run_id: "tool-transaction-run".to_owned(),
                    tool_name: call.name().to_owned(),
                    arguments: call.arguments().clone(),
                    status: "requested".to_owned(),
                },
                runtime_id: runtime_id.clone(),
                expected_runtime_revision: Some(requested_snapshot.revision() - 1),
                snapshot: completed_snapshot,
                // The legacy row is missing, so this single transaction also
                // replays the request event before the result event.
                events: [requested_events[0].clone(), result_event].to_vec(),
                lease: Some(lease),
            },
            "completed",
            json!({"echo": "hello"}),
        )
        .expect("backfill legacy tool row");
    assert_eq!(record.status, "completed");
    assert_eq!(record.result, Some(json!({"echo": "hello"})));
    assert_eq!(
        store
            .list_tool_calls_for_run("tool-transaction-run")
            .unwrap()
            .len(),
        1
    );
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("events")
            .len(),
        5
    );
    assert_eq!(requested_events.len(), 1);
}

#[test]
#[cfg(feature = "core-adapter")]
fn tool_call_checkpoint_runtime_transaction_commits_all_rows_together() {
    let (store, runtime_id, snapshot, events, call, lease) = tool_call_runtime_fixture();
    let checkpoint = DurableCheckpointInput {
        run_id: "tool-transaction-run".to_owned(),
        phase: "tool_in_flight".to_owned(),
        step: 0,
        next_step: 1,
        messages: json!([{"role": "user", "content": [{"type": "text", "text": "tool transaction"}]}]),
        provider_request_id: None,
        tool_call_id: Some(call.id().to_owned()),
        attempt: 1,
    };
    let record = store
        .create_tool_call_with_checkpoint_runtime_and_lease(
            DurableToolCallCheckpointRuntimeCommit {
                call: DurableToolCallInput {
                    id: call.id().to_owned(),
                    session_id: "tool-transaction-session".to_owned(),
                    run_id: "tool-transaction-run".to_owned(),
                    tool_name: call.name().to_owned(),
                    arguments: call.arguments().clone(),
                    status: "requested".to_owned(),
                },
                checkpoint,
                runtime_id: runtime_id.clone(),
                expected_runtime_revision: Some(snapshot.revision() - 1),
                snapshot: snapshot.clone(),
                events,
                lease,
            },
        )
        .expect("atomic tool/checkpoint request");
    assert_eq!(record.status, "requested");
    let persisted_checkpoint = store
        .read_checkpoint("tool-transaction-run")
        .expect("read checkpoint")
        .expect("checkpoint exists");
    assert_eq!(persisted_checkpoint.phase, "tool_in_flight");
    assert_eq!(
        persisted_checkpoint.tool_call_id.as_deref(),
        Some(call.id())
    );
    assert_eq!(store.load(&runtime_id).expect("runtime"), Some(snapshot));
    assert_eq!(
        store.get_tool_call(call.id()).expect("tool call"),
        Some(record)
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn tool_call_checkpoint_runtime_transaction_rolls_back_on_lease_failure() {
    let (store, runtime_id, snapshot, events, call, _) = tool_call_runtime_fixture();
    let error = store
        .create_tool_call_with_checkpoint_runtime_and_lease(
            DurableToolCallCheckpointRuntimeCommit {
                call: DurableToolCallInput {
                    id: call.id().to_owned(),
                    session_id: "tool-transaction-session".to_owned(),
                    run_id: "tool-transaction-run".to_owned(),
                    tool_name: call.name().to_owned(),
                    arguments: call.arguments().clone(),
                    status: "requested".to_owned(),
                },
                checkpoint: DurableCheckpointInput {
                    run_id: "tool-transaction-run".to_owned(),
                    phase: "tool_in_flight".to_owned(),
                    step: 0,
                    next_step: 1,
                    messages: json!([]),
                    provider_request_id: None,
                    tool_call_id: Some(call.id().to_owned()),
                    attempt: 1,
                },
                runtime_id: runtime_id.clone(),
                expected_runtime_revision: Some(snapshot.revision() - 1),
                snapshot,
                events,
                lease: DurableLeaseIdentity {
                    worker_id: "tool-worker".to_owned(),
                    lease_token: "stale-token".to_owned(),
                },
            },
        )
        .expect_err("stale lease must fence all writes");
    assert!(matches!(
        error,
        StorageError::LeaseLost { .. } | StorageError::LeaseConflict { .. }
    ));
    assert!(store.get_tool_call(call.id()).expect("tool call").is_none());
    assert!(
        store
            .read_checkpoint("tool-transaction-run")
            .expect("checkpoint")
            .is_none()
    );
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("runtime events")
            .len(),
        3
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn tool_call_checkpoint_runtime_transaction_rolls_back_on_runtime_cas_failure() {
    let (store, runtime_id, snapshot, events, call, lease) = tool_call_runtime_fixture();
    let mut stale_event = events[0].clone();
    stale_event.revision = 3;
    let mut stale_snapshot = snapshot;
    stale_snapshot.revision = 3;
    let error = store
        .create_tool_call_with_checkpoint_runtime_and_lease(
            DurableToolCallCheckpointRuntimeCommit {
                call: DurableToolCallInput {
                    id: call.id().to_owned(),
                    session_id: "tool-transaction-session".to_owned(),
                    run_id: "tool-transaction-run".to_owned(),
                    tool_name: call.name().to_owned(),
                    arguments: call.arguments().clone(),
                    status: "requested".to_owned(),
                },
                checkpoint: DurableCheckpointInput {
                    run_id: "tool-transaction-run".to_owned(),
                    phase: "tool_in_flight".to_owned(),
                    step: 0,
                    next_step: 1,
                    messages: json!([]),
                    provider_request_id: None,
                    tool_call_id: Some(call.id().to_owned()),
                    attempt: 1,
                },
                runtime_id: runtime_id.clone(),
                expected_runtime_revision: Some(2),
                snapshot: stale_snapshot,
                events: vec![stale_event],
                lease,
            },
        )
        .expect_err("stale runtime revision must roll back all writes");
    assert!(matches!(error, StorageError::RevisionConflict { .. }));
    assert!(store.get_tool_call(call.id()).expect("tool call").is_none());
    assert!(
        store
            .read_checkpoint("tool-transaction-run")
            .expect("checkpoint")
            .is_none()
    );
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("runtime events")
            .len(),
        3
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn complete_tool_call_checkpoint_runtime_transaction_commits_all_rows_together() {
    let (store, runtime_id, requested_snapshot, requested_events, call, lease) =
        tool_call_runtime_fixture();
    let result = ToolResult::success(call.id(), json!({"echo": "hello"})).expect("result");
    let result_event = RuntimeEvent::tool_result(
        &runtime_id,
        requested_snapshot.revision() + 1,
        5,
        "tool-transaction-run",
        &result,
        false,
    )
    .expect("result event");
    let completed_snapshot = reduce(&requested_snapshot, &result_event).expect("reduce result");
    let record = store
        .complete_tool_call_with_checkpoint_runtime_and_lease(
            DurableToolCallCheckpointRuntimeCommit {
                call: DurableToolCallInput {
                    id: call.id().to_owned(),
                    session_id: "tool-transaction-session".to_owned(),
                    run_id: "tool-transaction-run".to_owned(),
                    tool_name: call.name().to_owned(),
                    arguments: call.arguments().clone(),
                    status: "requested".to_owned(),
                },
                checkpoint: DurableCheckpointInput {
                    run_id: "tool-transaction-run".to_owned(),
                    phase: "safe".to_owned(),
                    step: 1,
                    next_step: 2,
                    messages: json!([]),
                    provider_request_id: None,
                    tool_call_id: None,
                    attempt: 1,
                },
                runtime_id: runtime_id.clone(),
                expected_runtime_revision: Some(requested_snapshot.revision() - 1),
                snapshot: completed_snapshot,
                events: [requested_events[0].clone(), result_event].to_vec(),
                lease,
            },
            "completed",
            json!({"echo": "hello"}),
        )
        .expect("atomic tool/checkpoint completion");
    assert_eq!(record.status, "completed");
    assert_eq!(record.result, Some(json!({"echo": "hello"})));
    let checkpoint = store
        .read_checkpoint("tool-transaction-run")
        .expect("read checkpoint")
        .expect("checkpoint exists");
    assert_eq!(checkpoint.phase, "safe");
    assert_eq!(
        store
            .list_runtime_events(&runtime_id)
            .expect("events")
            .len(),
        5
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn core_runtime_store_commit_load_and_cas_work() {
    let mut store = SqliteStore::open_in_memory().expect("open sqlite");
    let mut snapshot = RuntimeSnapshot::try_new("runtime-1").expect("snapshot");
    snapshot.revision = 1;
    let event = RuntimeEvent::runtime_created("runtime-1", 1, 10).expect("event");

    store
        .commit("runtime-1", None, &snapshot, std::slice::from_ref(&event))
        .expect("commit runtime");
    assert_eq!(
        store.load("runtime-1").expect("load"),
        Some(snapshot.clone())
    );
    assert_eq!(
        store.list_runtime_events("runtime-1").expect("events"),
        vec![event.clone()]
    );
    let mut export = Vec::new();
    assert_eq!(
        store
            .export_runtime_jsonl("runtime-1", &mut export)
            .expect("runtime export"),
        2
    );
    assert_eq!(export.iter().filter(|byte| **byte == b'\n').count(), 2);

    let conflict = store.commit(
        "runtime-1",
        Some(0),
        &snapshot,
        std::slice::from_ref(&event),
    );
    assert!(conflict.is_err());
}

#[test]
#[cfg(feature = "core-adapter")]
fn runtime_event_schema_version_round_trips_through_sqlite() {
    let mut store = SqliteStore::open_in_memory().expect("open sqlite");
    let runtime_id = "runtime-event-schema";
    let event = RuntimeEvent::runtime_created(runtime_id, 1, 10).expect("event");
    let mut snapshot = RuntimeSnapshot::try_new(runtime_id).expect("snapshot");
    snapshot = reduce(&snapshot, &event).expect("reduce event");

    store
        .commit(runtime_id, None, &snapshot, std::slice::from_ref(&event))
        .expect("commit event");

    let events = store
        .list_runtime_events(runtime_id)
        .expect("read runtime events");
    assert_eq!(events, vec![event.clone()]);
    assert_eq!(events[0].schema_version(), RUNTIME_EVENT_SCHEMA_VERSION);
    let encoded = serde_json::to_value(&events[0]).expect("serialize persisted event");
    assert_eq!(
        encoded
            .get("schemaVersion")
            .and_then(|value| value.as_str()),
        Some(RUNTIME_EVENT_SCHEMA_VERSION)
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn runtime_id_for_run_uses_runtime_event_mapping() {
    let mut store = SqliteStore::open_in_memory().expect("open sqlite");
    let mut snapshot = RuntimeSnapshot::try_new("runtime-with-run").expect("snapshot");
    let runtime_created =
        RuntimeEvent::runtime_created("runtime-with-run", 1, SystemClock.now_millis())
            .expect("runtime event");
    let run = RunSnapshot::try_new("mapped-run", "agent-1", "resume me", 10).expect("run");
    let run_created =
        RuntimeEvent::run_created("runtime-with-run", 2, 10, &run).expect("run event");
    snapshot = reduce(&snapshot, &runtime_created).expect("reduce runtime");
    snapshot = reduce(&snapshot, &run_created).expect("reduce run");
    store
        .commit(
            "runtime-with-run",
            None,
            &snapshot,
            &[runtime_created, run_created],
        )
        .expect("commit runtime");

    assert_eq!(
        store
            .runtime_id_for_run("mapped-run")
            .expect("lookup runtime"),
        Some("runtime-with-run".to_owned())
    );
    assert_eq!(
        store.runtime_id_for_run("unknown").expect("lookup missing"),
        None
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn core_runtime_store_rejects_invalid_batch_without_writing_state() {
    let mut store = SqliteStore::open_in_memory().expect("open sqlite");
    let mut snapshot = RuntimeSnapshot::try_new("runtime-2").expect("snapshot");
    snapshot.revision = 2;
    let event = RuntimeEvent::runtime_created("runtime-2", 1, 10).expect("event");
    let result = store.commit("runtime-2", None, &snapshot, std::slice::from_ref(&event));
    assert!(result.is_err());
    assert!(store.load("runtime-2").expect("load").is_none());
    assert!(
        store
            .list_runtime_events("runtime-2")
            .expect("events")
            .is_empty()
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn core_runtime_store_rejects_snapshot_not_derived_from_events() {
    let mut store = SqliteStore::open_in_memory().expect("open sqlite");
    let event = RuntimeEvent::runtime_created("runtime-mismatch", 1, 10).expect("event");
    let mut snapshot = RuntimeSnapshot::try_new("runtime-mismatch").expect("snapshot");
    snapshot.revision = 1;
    // revision/identity 都看似合法，但 snapshot 并未由事件 reducer 产生。
    snapshot.metadata = json!({"unexpected": true});
    let result = store.commit(
        "runtime-mismatch",
        None,
        &snapshot,
        std::slice::from_ref(&event),
    );
    assert!(result.is_err());
    assert!(store.load("runtime-mismatch").expect("load").is_none());
}

#[test]
#[cfg(feature = "core-adapter")]
fn runtime_store_rejects_tampered_snapshot_on_load_and_commit() {
    let tempdir = local_tempdir();
    let database = tempdir.path().join("tampered-runtime.db");
    let runtime_id = "runtime-tamper";
    let (snapshot, next_event, next_snapshot) = {
        let mut store = SqliteStore::open(&database).expect("open sqlite");
        let initial = RuntimeSnapshot::try_new(runtime_id).expect("snapshot");
        let created = RuntimeEvent::runtime_created(runtime_id, 1, 10).expect("event");
        let snapshot = reduce(&initial, &created).expect("reduce initial event");
        store
            .commit(runtime_id, None, &snapshot, std::slice::from_ref(&created))
            .expect("commit initial runtime");

        let run =
            RunSnapshot::try_new("tamper-run", "agent-1", "continue", 11).expect("run snapshot");
        let next_event = RuntimeEvent::run_created(runtime_id, 2, 11, &run).expect("run event");
        let next_snapshot = reduce(&snapshot, &next_event).expect("reduce next event");
        (snapshot, next_event, next_snapshot)
    };

    // A structurally valid JSON value can still diverge from the event log.
    // Mutate it through a second connection to model a damaged database or an
    // older adapter that wrote an unchecked value.
    let mut tampered = serde_json::to_value(&snapshot).expect("encode snapshot");
    tampered["metadata"] = json!({"tampered": true});
    let tampered_json = serde_json::to_string(&tampered).expect("encode tampered snapshot");
    let connection = rusqlite::Connection::open(&database).expect("open inspection connection");
    connection
        .execute(
            "UPDATE runtime_states SET snapshot_json = ?1 WHERE runtime_id = ?2",
            rusqlite::params![tampered_json, runtime_id],
        )
        .expect("tamper persisted snapshot");
    drop(connection);

    let mut store = SqliteStore::open(&database).expect("reopen sqlite");
    let load_error = store
        .load_runtime_snapshot(runtime_id)
        .expect_err("tampered snapshot must fail the storage boundary");
    assert!(
        matches!(load_error, StorageError::InvalidInput(message) if message.contains("重放结果"))
    );

    let trait_error = RuntimeStore::load(&store, runtime_id)
        .expect_err("RuntimeStore must classify a tampered snapshot");
    assert_eq!(trait_error.kind(), StoreErrorKind::InvalidSnapshot);

    let commit_error = RuntimeStore::commit(
        &mut store,
        runtime_id,
        Some(snapshot.revision()),
        &next_snapshot,
        std::slice::from_ref(&next_event),
    )
    .expect_err("commit must revalidate the current persisted snapshot before replay");
    assert_eq!(commit_error.kind(), StoreErrorKind::InvalidSnapshot);

    let connection = rusqlite::Connection::open(&database).expect("reopen inspection connection");
    let persisted: String = connection
        .query_row(
            "SELECT snapshot_json FROM runtime_states WHERE runtime_id = ?1",
            [runtime_id],
            |row| row.get(0),
        )
        .expect("read persisted snapshot");
    assert_eq!(persisted, tampered_json);
    let event_count: i64 = connection
        .query_row(
            "SELECT COUNT(*) FROM runtime_events WHERE runtime_id = ?1 AND revision = 2",
            [runtime_id],
            |row| row.get(0),
        )
        .expect("count uncommitted event");
    assert_eq!(event_count, 0);
}

#[test]
#[cfg(feature = "core-adapter")]
fn runtime_store_rejects_event_history_gap_on_load_and_commit() {
    let tempdir = local_tempdir();
    let database = tempdir.path().join("broken-runtime-events.db");
    let runtime_id = "runtime-event-gap";
    let (snapshot, next_event, next_snapshot) = {
        let mut store = SqliteStore::open(&database).expect("open sqlite");
        let run =
            RunSnapshot::try_new("gap-run", "agent-1", "detect a gap", 10).expect("run snapshot");
        let events = [
            RuntimeEvent::runtime_created(runtime_id, 1, 10).expect("runtime event"),
            RuntimeEvent::run_created(runtime_id, 2, 11, &run).expect("run event"),
            RuntimeEvent::status_changed(
                runtime_id,
                3,
                12,
                run.run_id(),
                agent_runtime_core::RuntimeEventKind::RunStarted,
            )
            .expect("started event"),
        ];
        let snapshot = events
            .iter()
            .try_fold(
                RuntimeSnapshot::try_new(runtime_id).expect("initial snapshot"),
                |snapshot, event| reduce(&snapshot, event),
            )
            .expect("reduce history");
        store
            .commit(runtime_id, None, &snapshot, &events)
            .expect("commit history");
        let next_event = RuntimeEvent::status_changed(
            runtime_id,
            4,
            13,
            run.run_id(),
            agent_runtime_core::RuntimeEventKind::RunCompleted,
        )
        .expect("completion event");
        let next_snapshot = reduce(&snapshot, &next_event).expect("reduce completion");
        (snapshot, next_event, next_snapshot)
    };

    // Delete the middle revision through a second connection.  The snapshot
    // row still looks internally valid, but its audit stream is no longer a
    // complete prefix and must not be used as a new commit base.
    let connection = rusqlite::Connection::open(&database).expect("open inspection connection");
    connection
        .execute(
            "DELETE FROM runtime_events WHERE runtime_id = ?1 AND revision = 2",
            [runtime_id],
        )
        .expect("delete middle event");
    drop(connection);

    let mut store = SqliteStore::open(&database).expect("reopen sqlite");
    let load_error = store
        .load_runtime_snapshot(runtime_id)
        .expect_err("event history gap must fail the storage boundary");
    assert!(matches!(
        load_error,
        StorageError::InvalidInput(message) if message.contains("event 数量")
    ));
    let list_error = store
        .list_runtime_events(runtime_id)
        .expect_err("listing must reject a broken historical prefix");
    assert!(matches!(
        list_error,
        StorageError::InvalidInput(message) if message.contains("event 数量")
    ));
    let trait_error = RuntimeStore::load(&store, runtime_id)
        .expect_err("RuntimeStore must classify a history gap");
    assert_eq!(trait_error.kind(), StoreErrorKind::InvalidSnapshot);

    let commit_error = RuntimeStore::commit(
        &mut store,
        runtime_id,
        Some(snapshot.revision()),
        &next_snapshot,
        std::slice::from_ref(&next_event),
    )
    .expect_err("commit must reject a broken historical prefix");
    assert_eq!(commit_error.kind(), StoreErrorKind::InvalidSnapshot);

    let connection = rusqlite::Connection::open(&database).expect("reopen inspection connection");
    let event_count: i64 = connection
        .query_row(
            "SELECT COUNT(*) FROM runtime_events WHERE runtime_id = ?1 AND revision = 4",
            [runtime_id],
            |row| row.get(0),
        )
        .expect("count rejected event");
    assert_eq!(event_count, 0);
}

#[test]
#[cfg(feature = "core-adapter")]
fn expired_recovery_commits_all_states_and_preserves_checkpoint() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "atomic-recovery-session".into(),
            agent_id: Some("agent-1".into()),
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    let run_id = "atomic-recovery-run";
    let runtime_id = "atomic-recovery-runtime";
    store
        .create_run(NewRun {
            id: run_id.into(),
            session_id: "atomic-recovery-session".into(),
            status: "queued".into(),
            input: json!({"message": "recover"}),
        })
        .expect("create run");

    let mut snapshot = RuntimeSnapshot::try_new(runtime_id).expect("runtime");
    let runtime_created = RuntimeEvent::runtime_created(runtime_id, 1, 1).expect("runtime event");
    let run = RunSnapshot::try_new(run_id, "agent-1", "recover", 1).expect("run snapshot");
    let run_created = RuntimeEvent::run_created(runtime_id, 2, 2, &run).expect("run event");
    snapshot = reduce(&snapshot, &runtime_created).expect("reduce runtime");
    snapshot = reduce(&snapshot, &run_created).expect("reduce run");
    let mut store_for_commit = store.clone();
    store_for_commit
        .commit(
            runtime_id,
            None,
            &snapshot,
            &[runtime_created.clone(), run_created.clone()],
        )
        .expect("commit initial runtime");

    let (_, lease) = store
        .claim_run_with_lease(
            run_id,
            "worker-atomic",
            "token-atomic",
            Duration::from_millis(100),
        )
        .expect("claim");
    let checkpoint = store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.into(),
                phase: "provider_in_flight".into(),
                step: 0,
                next_step: 0,
                messages: json!([]),
                provider_request_id: Some("request-atomic".into()),
                tool_call_id: None,
                attempt: lease.attempt,
            },
            "worker-atomic",
            "token-atomic",
        )
        .expect("save checkpoint");
    wait_until_epoch_ms(lease.lease_expires_at);

    let started = RuntimeEvent::status_changed(
        runtime_id,
        snapshot.revision() + 1,
        3,
        run_id,
        agent_runtime_core::RuntimeEventKind::RunStarted,
    )
    .expect("started event");
    let mut next = reduce(&snapshot, &started).expect("reduce started");
    let required = RuntimeEvent::status_changed(
        runtime_id,
        next.revision() + 1,
        4,
        run_id,
        agent_runtime_core::RuntimeEventKind::ReconciliationRequired,
    )
    .expect("reconciliation event");
    next = reduce(&next, &required).expect("reduce reconciliation");
    let recovered = store
        .recover_expired_run_with_runtime(
            run_id,
            runtime_id,
            Some(snapshot.revision()),
            &next,
            &[started, required],
        )
        .expect("atomic recovery");

    assert_eq!(recovered.status, "reconciling");
    assert!(store.get_run_lease(run_id).expect("lease").is_none());
    assert_eq!(
        store
            .get_session("atomic-recovery-session")
            .expect("session")
            .unwrap()
            .status,
        "reconciling"
    );
    assert_eq!(store.load(runtime_id).expect("runtime"), Some(next.clone()));
    assert_eq!(
        store.list_runtime_events(runtime_id).expect("events").len(),
        4
    );
    assert_eq!(
        store.read_checkpoint(run_id).expect("checkpoint"),
        Some(checkpoint)
    );
}

#[test]
#[cfg(feature = "core-adapter")]
fn expired_recovery_invalid_runtime_batch_rolls_back_run_and_lease() {
    let store = SqliteStore::open_in_memory().expect("open sqlite");
    store
        .create_session(NewSession {
            id: "rollback-recovery-session".into(),
            agent_id: Some("agent-1".into()),
            status: "active".into(),
            metadata: json!({}),
        })
        .expect("create session");
    let run_id = "rollback-recovery-run";
    let runtime_id = "rollback-recovery-runtime";
    store
        .create_run(NewRun {
            id: run_id.into(),
            session_id: "rollback-recovery-session".into(),
            status: "queued".into(),
            input: json!({"message": "recover"}),
        })
        .expect("create run");
    let mut snapshot = RuntimeSnapshot::try_new(runtime_id).expect("runtime");
    let runtime_created = RuntimeEvent::runtime_created(runtime_id, 1, 1).expect("runtime event");
    let run = RunSnapshot::try_new(run_id, "agent-1", "recover", 1).expect("run snapshot");
    let run_created = RuntimeEvent::run_created(runtime_id, 2, 2, &run).expect("run event");
    snapshot = reduce(&snapshot, &runtime_created).expect("reduce runtime");
    snapshot = reduce(&snapshot, &run_created).expect("reduce run");
    let mut store_for_commit = store.clone();
    store_for_commit
        .commit(runtime_id, None, &snapshot, &[runtime_created, run_created])
        .expect("commit initial runtime");
    let (_, lease) = store
        .claim_run_with_lease(
            run_id,
            "worker-rollback",
            "token-rollback",
            Duration::from_millis(100),
        )
        .expect("claim");
    store
        .save_checkpoint_with_lease(
            NewCheckpoint {
                run_id: run_id.into(),
                phase: "provider_in_flight".into(),
                step: 0,
                next_step: 0,
                messages: json!([]),
                provider_request_id: Some("request-rollback".into()),
                tool_call_id: None,
                attempt: lease.attempt,
            },
            "worker-rollback",
            "token-rollback",
        )
        .expect("save checkpoint");
    wait_until_epoch_ms(lease.lease_expires_at);
    let session_before = store
        .get_session("rollback-recovery-session")
        .expect("session")
        .unwrap();

    let invalid_event = RuntimeEvent::status_changed(
        runtime_id,
        snapshot.revision() + 2,
        3,
        run_id,
        agent_runtime_core::RuntimeEventKind::ReconciliationRequired,
    )
    .expect("invalid event shape");
    let mut invalid_next = snapshot.clone();
    invalid_next.revision = snapshot.revision() + 1;
    let error = store
        .recover_expired_run_with_runtime(
            run_id,
            runtime_id,
            Some(snapshot.revision()),
            &invalid_next,
            std::slice::from_ref(&invalid_event),
        )
        .expect_err("invalid revision must roll back");
    assert!(matches!(error, StorageError::InvalidInput(_)));
    assert_eq!(
        store.get_run(run_id).expect("run").unwrap().status,
        "running"
    );
    assert!(store.get_run_lease(run_id).expect("lease").is_some());
    assert_eq!(
        store.load(runtime_id).expect("runtime"),
        Some(snapshot.clone())
    );
    assert_eq!(
        store
            .get_session("rollback-recovery-session")
            .expect("session")
            .unwrap(),
        session_before
    );

    let started = RuntimeEvent::status_changed(
        runtime_id,
        snapshot.revision() + 1,
        4,
        run_id,
        agent_runtime_core::RuntimeEventKind::RunStarted,
    )
    .expect("started event");
    let mut next = reduce(&snapshot, &started).expect("reduce started");
    let required = RuntimeEvent::status_changed(
        runtime_id,
        next.revision() + 1,
        5,
        run_id,
        agent_runtime_core::RuntimeEventKind::ReconciliationRequired,
    )
    .expect("reconciliation event");
    next = reduce(&next, &required).expect("reduce reconciliation");
    let conflict = store
        .recover_expired_run_with_runtime(
            run_id,
            runtime_id,
            Some(snapshot.revision() + 1),
            &next,
            std::slice::from_ref(&required),
        )
        .expect_err("revision conflict must roll back");
    assert!(matches!(conflict, StorageError::RevisionConflict { .. }));
    assert_eq!(
        store.get_run(run_id).expect("run").unwrap().status,
        "running"
    );
    assert!(store.get_run_lease(run_id).expect("lease").is_some());
}