kaynine-runtime 0.1.0

use kaynine_core::budget::BudgetPolicy;
use kaynine_core::error::KaynineError;
use kaynine_core::event::RealtimeEvent;
use kaynine_core::ids::{BranchId, ModelId, SessionId};
use kaynine_core::message::{ContentBlock, FinishReason, Message};
use kaynine_core::provider::{ModelCapabilities, ProviderEvent, ReasoningLevel};
use kaynine_core::store::{
    CreateSessionRequest, RunState, SessionStore, SteerState, ToolCallState,
};
use kaynine_core::testing::{
    FakeCredentialProvider, FakeProvider, FakeTokenCounter, ScriptedPolicy, ScriptedTool,
};
use kaynine_core::tool::Tool;
use kaynine_runtime::{
    AgentRuntime, ApprovalOutcome, ApprovalResolution, CancelOutcome, CancelRequest,
    ReleaseOutcome, StartRunRequest, SteerRequest, SubscriptionItem, UpdateSessionRequest,
};
use kaynine_store::sqlite::SqliteStore;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

fn caps() -> ModelCapabilities {
    ModelCapabilities {
        context_tokens: 200_000,
        max_output_tokens: 8_192,
        supports_tools: true,
        supports_images: true,
        supports_reasoning: true,
    }
}

struct TempDb(PathBuf);

impl Drop for TempDb {
    fn drop(&mut self) {
        for suffix in ["", "-wal", "-shm"] {
            let _ = std::fs::remove_file(format!("{}{}", self.0.display(), suffix));
        }
    }
}

fn session_id() -> SessionId {
    SessionId::from("session-1")
}

fn main_branch_id() -> BranchId {
    BranchId::from("session-1_branch_main")
}

async fn setup() -> (AgentRuntime, Arc<SqliteStore>, SessionId, TempDb) {
    let path =
        std::env::temp_dir().join(format!("kaynine-runtime-test-{}.db", uuid::Uuid::new_v4()));
    let store = Arc::new(SqliteStore::open(&path).expect("open store"));
    let runtime = AgentRuntime::new(store.clone());
    let sid = session_id();
    runtime
        .create_session(CreateSessionRequest {
            command: None,
            session_id: sid.clone(),
            definition_id: "defs/agent".to_string(),
            definition_version: 1,
            default_model: ModelId::from("model-1"),
            metadata: serde_json::json!({}),
        })
        .await
        .expect("create session");
    (runtime, store, sid, TempDb(path))
}

fn text_events(text: &str) -> Vec<ProviderEvent> {
    vec![
        ProviderEvent::ResponseStarted,
        ProviderEvent::TextDelta {
            block: 0,
            text: text.into(),
        },
        ProviderEvent::ResponseCompleted {
            finish_reason: FinishReason::Stop,
        },
    ]
}

fn tool_call_events(calls: &[(&str, &str, &str)]) -> Vec<ProviderEvent> {
    let mut events = vec![ProviderEvent::ResponseStarted];
    for (i, (id, name, args)) in calls.iter().enumerate() {
        events.push(ProviderEvent::ToolCallStarted {
            block: i as u32,
            id: (*id).into(),
            name: (*name).into(),
        });
        events.push(ProviderEvent::ToolCallArgumentsDelta {
            block: i as u32,
            json: (*args).into(),
        });
    }
    events.push(ProviderEvent::ResponseCompleted {
        finish_reason: FinishReason::Stop,
    });
    events
}

#[allow(clippy::too_many_arguments)]
fn start_request(provider: Arc<FakeProvider>, tools: Vec<Arc<dyn Tool>>) -> StartRunRequest {
    StartRunRequest {
        command_id: format!("cmd-{}", uuid::Uuid::new_v4()),
        session_id: session_id(),
        branch_id: main_branch_id(),
        content: vec![ContentBlock::Text {
            text: "hello".to_string(),
        }],
        model_override: None,
        reasoning_override: None,
        capabilities: caps(),
        system_prompt: String::new(),
        provider,
        token_counter: Arc::new(FakeTokenCounter::fixed(1_000)),
        credentials: Arc::new(FakeCredentialProvider::default()),
        tools,
        budget: BudgetPolicy::default(),
        max_turns: Some(4),
        policy: std::sync::Arc::new(kaynine_core::policy::AllowAllPolicy),
        approval_timeout: None,
        prompt: None,
        compaction: None,
        compaction_selector: None,
    }
}

async fn wait_until_idle(
    runtime: &AgentRuntime,
    sid: &SessionId,
    min_revision: u64,
) -> kaynine_runtime::SessionSnapshot {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        let snapshot = runtime.get_snapshot(sid).await.expect("snapshot");
        if snapshot.active_run.is_none() && snapshot.current_revision >= min_revision {
            return snapshot;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "run did not settle; revision = {}",
            snapshot.current_revision
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

#[tokio::test]
async fn full_text_run_persists_and_broadcasts() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("hi there"));

    let accepted = runtime
        .start_run(start_request(provider, vec![]))
        .await
        .expect("start_run accepted");

    // Events: UserMessageAccepted, RunStarted, TurnStarted,
    // AssistantMessagePersisted, TurnCompleted, RunTerminal.
    let snapshot = wait_until_idle(&runtime, &sid, 6).await;
    assert_eq!(snapshot.current_revision, 6);
    let chain = &snapshot.chains[&main_branch_id()];
    assert_eq!(chain.len(), 2);
    assert_eq!(chain[0].entry_id, accepted.user_entry_id);
    assert!(matches!(&chain[0].message, Message::User { .. }));
    assert!(matches!(&chain[1].message, Message::Assistant { .. }));

    let run = store
        .get_run(&accepted.run_id)
        .await
        .unwrap()
        .expect("run row");
    assert_eq!(run.state, RunState::Completed);
    assert_eq!(run.model, ModelId::from("model-1"));
    assert_eq!(run.reasoning, ReasoningLevel::Off);
    assert!(run.started_at_unix > 0, "store stamps started_at clock");
    assert_eq!(run.branch_id, main_branch_id());

    // Head must be the assistant entry and the session revision must agree.
    let branch = &snapshot.branches[0];
    assert_eq!(branch.head_entry_id.as_ref(), Some(&chain[1].entry_id));
    let session = store.get_session(&sid).await.unwrap().unwrap();
    assert_eq!(session.current_revision, snapshot.current_revision);
}

#[tokio::test]
async fn tool_run_persists_planned_and_results() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(tool_call_events(&[("call-1", "echo", "{\"a\":1}")]));
    provider.push_events(text_events("done"));

    let accepted = runtime
        .start_run(start_request(
            provider,
            vec![Arc::new(ScriptedTool::echo("echo"))],
        ))
        .await
        .expect("start_run accepted");

    // Events: user, run_started, turn1(turn_started/asst/turn_completed),
    // tool_call_planned, tool_result_recorded, turn2(turn_started/asst/
    // turn_completed), run_terminal = 11 revisions.
    let snapshot = wait_until_idle(&runtime, &sid, 11).await;
    assert_eq!(snapshot.current_revision, 11);

    let chain = &snapshot.chains[&main_branch_id()];
    assert_eq!(
        chain.len(),
        4,
        "user, assistant(toolcall), toolresult, assistant"
    );
    assert!(matches!(&chain[2].message, Message::ToolResult { .. }));

    let calls = store.list_tool_calls(&accepted.run_id).await.unwrap();
    assert_eq!(calls.len(), 1);
    assert_eq!(calls[0].state, ToolCallState::Succeeded);
    assert_eq!(calls[0].name, "echo");
    assert_eq!(calls[0].turn, 1);

    let run = store.get_run(&accepted.run_id).await.unwrap().unwrap();
    assert_eq!(run.state, RunState::Completed);
}

#[tokio::test]
async fn second_run_rejected_while_active() {
    let (runtime, _store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_hang();

    let accepted = runtime
        .start_run(start_request(provider.clone(), vec![]))
        .await
        .expect("first run accepted");

    let second = runtime.start_run(start_request(provider, vec![])).await;
    assert_eq!(second.unwrap_err(), KaynineError::RunAlreadyActive);

    let cancel = runtime
        .cancel(CancelRequest {
            command_id: format!("cmd-{}", uuid::Uuid::new_v4()),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
        })
        .await
        .expect("cancel");
    assert!(matches!(cancel, CancelOutcome::Cancelled { .. }));

    let snapshot = wait_until_idle(&runtime, &sid, 4).await;
    assert_eq!(snapshot.current_revision, 4); // user, run_started, turn_started, run_terminal
}

#[tokio::test]
async fn cancel_persists_terminal_and_synthesis() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(tool_call_events(&[("call-1", "slow", "{}")]));

    let accepted = runtime
        .start_run(start_request(
            provider,
            vec![Arc::new(ScriptedTool::cancel_aware_sleep(
                "slow",
                Duration::from_secs(30),
            ))],
        ))
        .await
        .expect("start_run accepted");

    tokio::time::sleep(Duration::from_millis(100)).await;
    let outcome = runtime
        .cancel(CancelRequest {
            command_id: format!("cmd-{}", uuid::Uuid::new_v4()),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
        })
        .await
        .expect("cancel");
    assert!(matches!(outcome, CancelOutcome::Cancelled { .. }));

    // user, run_started, turn, assistant(toolcall), turn_completed, planned,
    // synthesized tool result, run_terminal = 8 revisions.
    let snapshot = wait_until_idle(&runtime, &sid, 8).await;
    assert_eq!(snapshot.current_revision, 8);

    let run = store.get_run(&accepted.run_id).await.unwrap().unwrap();
    assert_eq!(run.state, RunState::Cancelled);

    let chain = &snapshot.chains[&main_branch_id()];
    let synthesized = chain
        .iter()
        .find_map(|entry| match &entry.message {
            Message::ToolResult { results } => Some(results.clone()),
            _ => None,
        })
        .expect("synthesized tool result persisted");
    assert_eq!(synthesized.len(), 1);
    assert!(synthesized[0].is_error);

    // Cancelling again is domain-idempotent: already terminal.
    let again = runtime
        .cancel(CancelRequest {
            command_id: format!("cmd-{}", uuid::Uuid::new_v4()),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
        })
        .await
        .expect("second cancel");
    assert_eq!(
        again,
        CancelOutcome::AlreadyTerminal {
            state: RunState::Cancelled
        }
    );
}

#[tokio::test]
async fn get_run_returns_state() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("ok"));
    let accepted = runtime
        .start_run(start_request(provider, vec![]))
        .await
        .unwrap();
    wait_until_idle(&runtime, &sid, 6).await;

    let run = store.get_run(&accepted.run_id).await.unwrap().unwrap();
    assert_eq!(run.state, RunState::Completed);
    assert!(store
        .get_run(&kaynine_core::ids::RunId::from("missing"))
        .await
        .unwrap()
        .is_none());
}

#[tokio::test]
async fn snapshot_reflects_store_truth() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("ok"));
    runtime
        .start_run(start_request(provider, vec![]))
        .await
        .unwrap();
    let snapshot = wait_until_idle(&runtime, &sid, 6).await;

    let session = store.get_session(&sid).await.unwrap().unwrap();
    assert_eq!(snapshot.session, session);
    assert_eq!(snapshot.current_revision, session.current_revision);

    let branches = store.list_branches(&sid).await.unwrap();
    assert_eq!(snapshot.branches, branches);
    let store_chain = store
        .load_chain(branches[0].head_entry_id.as_ref().unwrap())
        .await
        .unwrap();
    assert_eq!(snapshot.chains[&main_branch_id()], store_chain);
    assert!(snapshot.unapplied_steers.is_empty());
    // RunStarted, TurnStarted, TextDelta, TurnCompleted, RunCompleted.
    assert_eq!(snapshot.last_run_seq, Some(5));
}

#[tokio::test]
async fn image_content_rejected_until_blob_upload_exists() {
    let (runtime, _store, _sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    let mut request = start_request(provider, vec![]);
    request.content.push(ContentBlock::Image {
        media_type: "image/png".to_string(),
        data: kaynine_core::message::BinaryRef {
            sha256: "abc".to_string(),
            byte_len: 3,
        },
    });
    let error = runtime.start_run(request).await.unwrap_err();
    assert_eq!(error, KaynineError::InvalidRequest);
}

#[tokio::test]
async fn subscription_first_item_is_snapshot_then_events() {
    let (runtime, _store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("hi there"));

    let mut subscription = runtime.watch_session(&sid).await.expect("watch");
    runtime
        .start_run(start_request(provider, vec![]))
        .await
        .expect("start_run accepted");

    let first = subscription.next().await.expect("first item");
    assert!(
        matches!(first, SubscriptionItem::Snapshot(_)),
        "first item must be a snapshot"
    );

    let mut saw_turn_started = false;
    let mut saw_text_delta = false;
    let mut last_seq = 0u64;
    loop {
        let item = tokio::time::timeout(Duration::from_secs(5), subscription.next())
            .await
            .expect("timed out waiting for subscription item")
            .expect("stream closed");
        match item {
            SubscriptionItem::Event(envelope) => {
                assert!(envelope.revision > 0);
                if let Some(seq) = envelope.run_seq {
                    assert!(seq > last_seq, "run_seq must strictly increase");
                    last_seq = seq;
                }
                match envelope.payload {
                    RealtimeEvent::TurnStarted { .. } => saw_turn_started = true,
                    RealtimeEvent::TextDelta { .. } => saw_text_delta = true,
                    RealtimeEvent::RunCompleted => break,
                    _ => {}
                }
            }
            _ => panic!("unexpected item before RunCompleted"),
        }
    }
    assert!(saw_turn_started);
    assert!(saw_text_delta);
}

#[tokio::test]
async fn slow_consumer_gets_resync_required() {
    let (runtime, _store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    let mut flood = vec![ProviderEvent::ResponseStarted];
    for _ in 0..300 {
        flood.push(ProviderEvent::TextDelta {
            block: 0,
            text: "x".to_string(),
        });
    }
    flood.push(ProviderEvent::ResponseCompleted {
        finish_reason: FinishReason::Stop,
    });
    provider.push_events(flood);

    let mut subscription = runtime.watch_session(&sid).await.expect("watch");
    runtime
        .start_run(start_request(provider, vec![]))
        .await
        .expect("start_run accepted");
    // Let the run finish while the consumer holds everything unread; 304
    // realtime events overflow the 256-slot broadcast buffer.
    wait_until_idle(&runtime, &sid, 6).await;

    let mut saw_resync = false;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        assert!(tokio::time::Instant::now() < deadline, "no resync observed");
        let item = subscription.next().await.expect("stream closed");
        match item {
            SubscriptionItem::Snapshot(_) if saw_resync => break,
            SubscriptionItem::ResyncRequired => saw_resync = true,
            SubscriptionItem::Snapshot(_) | SubscriptionItem::Event(_) => {}
        }
    }
    assert!(saw_resync);
}

#[tokio::test]
async fn update_session_changes_model_when_idle() {
    let (runtime, _store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("ok"));
    runtime
        .start_run(start_request(provider.clone(), vec![]))
        .await
        .unwrap();
    wait_until_idle(&runtime, &sid, 6).await;

    let record = runtime
        .update_session(UpdateSessionRequest {
            command_id: Some(format!("cmd-{}", uuid::Uuid::new_v4())),
            session_id: sid.clone(),
            default_model: Some(ModelId::from("model-2")),
            reasoning: Some(ReasoningLevel::High),
            metadata: Some(serde_json::json!({"theme": "dark"})),
        })
        .await
        .expect("update_session");
    assert_eq!(record.default_model, ModelId::from("model-2"));
    assert_eq!(record.metadata, serde_json::json!({"theme": "dark"}));
    assert_eq!(record.current_revision, 8); // 6 run events + ModelChanged + SessionMetadataUpdated

    // Active run blocks the update.
    provider.push_hang();
    runtime
        .start_run(start_request(provider, vec![]))
        .await
        .unwrap();
    let busy = runtime
        .update_session(UpdateSessionRequest {
            command_id: Some(format!("cmd-{}", uuid::Uuid::new_v4())),
            session_id: sid.clone(),
            default_model: Some(ModelId::from("model-3")),
            reasoning: None,
            metadata: None,
        })
        .await
        .unwrap_err();
    assert_eq!(busy, KaynineError::SessionBusy);
}

#[tokio::test]
async fn release_and_reacquire() {
    let (runtime, _store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("first"));
    runtime
        .start_run(start_request(provider.clone(), vec![]))
        .await
        .unwrap();
    wait_until_idle(&runtime, &sid, 6).await;

    assert_eq!(
        runtime.release_session(&sid).await.unwrap(),
        ReleaseOutcome::Released
    );

    // A fresh actor (new lease + recovery) serves the next run.
    provider.push_events(text_events("second"));
    runtime
        .start_run(start_request(provider.clone(), vec![]))
        .await
        .expect("start_run after release");
    wait_until_idle(&runtime, &sid, 12).await;

    // Release with an active run is refused.
    provider.push_hang();
    runtime
        .start_run(start_request(provider, vec![]))
        .await
        .unwrap();
    assert_eq!(
        runtime.release_session(&sid).await.unwrap(),
        ReleaseOutcome::RunAlreadyActive
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn uncooperative_tool_run_terminates_within_grace() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(tool_call_events(&[("call-1", "stuck", "{}")]));
    // The tool ignores cancellation and never returns; the loop must still
    // terminate within cancel_grace (30s) and commit an in-doubt result.
    let stuck = ScriptedTool::with_behavior("stuck", |_call, _ctx| {
        Box::pin(async {
            loop {
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
        })
    });

    let accepted = runtime
        .start_run(start_request(provider, vec![Arc::new(stuck)]))
        .await
        .expect("start_run accepted");

    tokio::time::sleep(Duration::from_millis(100)).await;
    let outcome = tokio::time::timeout(
        Duration::from_secs(40),
        runtime.cancel(CancelRequest {
            command_id: format!("cmd-{}", uuid::Uuid::new_v4()),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
        }),
    )
    .await
    .expect("cancel reply within 40s (grace + slack)")
    .expect("cancel");
    assert!(matches!(outcome, CancelOutcome::Cancelled { .. }));

    // The run must be fully settled once cancel replies.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    let snapshot = loop {
        let snapshot = runtime.get_snapshot(&sid).await.expect("snapshot");
        if snapshot.active_run.is_none() {
            break snapshot;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "active run never cleared after cancel"
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    };

    let run = store
        .get_run(&accepted.run_id)
        .await
        .unwrap()
        .expect("run row");
    assert_eq!(run.state, RunState::Cancelled);

    let chain = &snapshot.chains[&main_branch_id()];
    let synthesized = chain
        .iter()
        .find_map(|entry| match &entry.message {
            Message::ToolResult { results } => Some(results.clone()),
            _ => None,
        })
        .expect("in-doubt tool result persisted");
    assert!(synthesized[0].is_error);
    assert!(
        synthesized[0].text.contains("不得假定"),
        "result must be the in-doubt synthesis, got: {}",
        synthesized[0].text
    );
}

#[tokio::test]
async fn cancel_after_completion_reports_already_terminal() {
    let (runtime, _store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("done"));
    let accepted = runtime
        .start_run(start_request(provider, vec![]))
        .await
        .unwrap();
    wait_until_idle(&runtime, &sid, 6).await;

    let outcome = runtime
        .cancel(CancelRequest {
            command_id: format!("cmd-{}", uuid::Uuid::new_v4()),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
        })
        .await
        .expect("cancel after completion");
    assert_eq!(
        outcome,
        CancelOutcome::AlreadyTerminal {
            state: RunState::Completed
        }
    );
}

#[tokio::test]
async fn shutdown_cancels_active_and_releases() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_hang();
    let accepted = runtime
        .start_run(start_request(provider, vec![]))
        .await
        .expect("start_run accepted");

    let report = runtime
        .shutdown(Duration::from_secs(2))
        .await
        .expect("shutdown");
    assert!(report.cancelled_runs.contains(&accepted.run_id));
    assert!(report.interrupted_runs.is_empty());
    assert!(report.released_sessions >= 1);

    let run = store.get_run(&accepted.run_id).await.unwrap().unwrap();
    assert_eq!(run.state, RunState::Cancelled);

    // Lease was released: a brand-new runtime on the same store can acquire
    // the lease, recover, and serve commands.
    let runtime2 = AgentRuntime::new(store.clone());
    let snapshot = runtime2.get_snapshot(&sid).await.expect("snapshot after");
    // user, run_started, turn_started, run_terminal.
    assert_eq!(snapshot.current_revision, 4);
}

fn cmd() -> String {
    format!("cmd-{}", uuid::Uuid::new_v4())
}

async fn wait_run_completed(store: &Arc<SqliteStore>, run_id: &kaynine_core::ids::RunId) {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        let run = store.get_run(run_id).await.unwrap().expect("run row");
        if run.state != RunState::Running {
            if run.state != RunState::Completed {
                panic!("run failed: {:?}", run.failure);
            }
            return;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "run never completed"
        );
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
}

/// Reads subscription events until `pred` matches an event payload.
async fn next_event<F>(
    subscription: &mut kaynine_runtime::SessionSubscription,
    pred: F,
) -> RealtimeEvent
where
    F: Fn(&RealtimeEvent) -> bool,
{
    loop {
        let item = tokio::time::timeout(Duration::from_secs(5), subscription.next())
            .await
            .expect("timed out waiting for subscription item")
            .expect("stream closed");
        match item {
            SubscriptionItem::Event(envelope) => {
                if pred(&envelope.payload) {
                    return envelope.payload;
                }
            }
            SubscriptionItem::Snapshot(_) | SubscriptionItem::ResyncRequired => {}
        }
    }
}

#[tokio::test]
async fn interactive_approval_end_to_end() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(tool_call_events(&[("call-1", "echo", "{\"a\":1}")]));
    provider.push_events(text_events("done"));

    let mut subscription = runtime.watch_session(&sid).await.expect("watch");
    let mut request = start_request(provider, vec![Arc::new(ScriptedTool::echo("echo"))]);
    request.policy = Arc::new(ScriptedPolicy::ask());
    request.approval_timeout = Some(Duration::from_secs(5));
    let accepted = runtime
        .start_run(request)
        .await
        .expect("start_run accepted");

    // Persisted first, then broadcast: ApprovalPending reaches watchers.
    let pending = next_event(&mut subscription, |e| {
        matches!(e, RealtimeEvent::ApprovalPending { .. })
    })
    .await;
    let RealtimeEvent::ApprovalPending {
        call_id,
        deadline_unix,
    } = pending
    else {
        unreachable!()
    };
    assert_eq!(call_id, kaynine_core::ids::ToolCallId::from("call-1"));
    assert!(deadline_unix > 0);

    let outcome = runtime
        .resolve_approval(ApprovalResolution {
            command_id: cmd(),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
            call_id,
            approved: true,
        })
        .await
        .expect("resolve");
    assert_eq!(outcome, ApprovalOutcome::Delivered);

    let resolved = next_event(&mut subscription, |e| {
        matches!(e, RealtimeEvent::ApprovalResolved { .. })
    })
    .await;
    assert_eq!(
        resolved,
        RealtimeEvent::ApprovalResolved {
            call_id: kaynine_core::ids::ToolCallId::from("call-1"),
            approved: true,
        }
    );

    wait_run_completed(&store, &accepted.run_id).await;

    // The approved tool actually executed (Succeeded, not denied).
    let calls = store.list_tool_calls(&accepted.run_id).await.unwrap();
    assert_eq!(calls.len(), 1);
    assert_eq!(calls[0].state, ToolCallState::Succeeded);

    // A second resolve for the same call is no longer deliverable.
    let again = runtime
        .resolve_approval(ApprovalResolution {
            command_id: cmd(),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
            call_id: kaynine_core::ids::ToolCallId::from("call-1"),
            approved: true,
        })
        .await
        .expect("second resolve");
    assert_eq!(again, ApprovalOutcome::NotFound);
}

#[tokio::test]
async fn approval_denied_blocks_tool() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(tool_call_events(&[("call-1", "echo", "{}")]));
    provider.push_events(text_events("done"));

    let mut subscription = runtime.watch_session(&sid).await.expect("watch");
    let mut request = start_request(provider, vec![Arc::new(ScriptedTool::echo("echo"))]);
    request.policy = Arc::new(ScriptedPolicy::ask());
    request.approval_timeout = Some(Duration::from_secs(5));
    let accepted = runtime
        .start_run(request)
        .await
        .expect("start_run accepted");

    let pending = next_event(&mut subscription, |e| {
        matches!(e, RealtimeEvent::ApprovalPending { .. })
    })
    .await;
    let RealtimeEvent::ApprovalPending { call_id, .. } = pending else {
        unreachable!()
    };

    let outcome = runtime
        .resolve_approval(ApprovalResolution {
            command_id: cmd(),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
            call_id,
            approved: false,
        })
        .await
        .expect("resolve");
    assert_eq!(outcome, ApprovalOutcome::Delivered);

    // Run completes with a structured permission-denied tool result.
    wait_run_completed(&store, &accepted.run_id).await;
    let snapshot = runtime.get_snapshot(&sid).await.unwrap();
    let chain = &snapshot.chains[&main_branch_id()];
    let denied = chain
        .iter()
        .find_map(|entry| match &entry.message {
            Message::ToolResult { results } => Some(results.clone()),
            _ => None,
        })
        .expect("denied tool result persisted");
    assert!(denied[0].is_error);
    assert!(
        denied[0].text.contains("权限拒绝"),
        "got: {}",
        denied[0].text
    );

    let calls = store.list_tool_calls(&accepted.run_id).await.unwrap();
    assert_eq!(calls[0].state, ToolCallState::Failed);
}

#[tokio::test]
async fn approval_timeout_denies() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(tool_call_events(&[("call-1", "echo", "{}")]));
    provider.push_events(text_events("done"));

    let mut request = start_request(provider, vec![Arc::new(ScriptedTool::echo("echo"))]);
    request.policy = Arc::new(ScriptedPolicy::ask());
    request.approval_timeout = Some(Duration::from_millis(200));
    let accepted = runtime
        .start_run(request)
        .await
        .expect("start_run accepted");

    // Never resolve: the handler times out and denies (fail closed).
    wait_run_completed(&store, &accepted.run_id).await;
    // Read the chain straight from the store: a snapshot would clear the
    // actor's finished active-run entry that resolve_approval consults.
    let branches = store.list_branches(&sid).await.unwrap();
    let chain = store
        .load_chain(branches[0].head_entry_id.as_ref().unwrap())
        .await
        .unwrap();
    let denied = chain
        .iter()
        .find_map(|entry| match &entry.message {
            Message::ToolResult { results } => Some(results.clone()),
            _ => None,
        })
        .expect("timed-out tool result persisted");
    assert!(denied[0].is_error);
    assert!(
        denied[0].text.contains("审批超时"),
        "got: {}",
        denied[0].text
    );

    // A late resolve maps to Expired.
    tokio::time::sleep(Duration::from_millis(50)).await;
    let outcome = runtime
        .resolve_approval(ApprovalResolution {
            command_id: cmd(),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
            call_id: kaynine_core::ids::ToolCallId::from("call-1"),
            approved: true,
        })
        .await
        .expect("late resolve");
    assert_eq!(outcome, ApprovalOutcome::Expired);
}

#[tokio::test]
async fn steer_mid_run_injection() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    // Turn 1 plans a tool that takes a moment, giving steer a window to land
    // before the safe point after the tool batch is persisted.
    provider.push_events(tool_call_events(&[("call-1", "slow", "{}")]));
    provider.push_events(text_events("done"));

    let accepted = runtime
        .start_run(start_request(
            provider.clone(),
            vec![Arc::new(ScriptedTool::cancel_aware_sleep(
                "slow",
                // Long enough that the run stays in flight (and the steer
                // window stays open) even under heavy parallel test load.
                Duration::from_millis(1500),
            ))],
        ))
        .await
        .expect("start_run accepted");

    let accepted_steer = runtime
        .steer(SteerRequest {
            command_id: cmd(),
            session_id: sid.clone(),
            run_id: accepted.run_id.clone(),
            content: "补充: 先看缓存".to_string(),
        })
        .await
        .expect("steer accepted");
    assert!(!accepted_steer.steer_id.is_empty());
    assert!(accepted_steer.revision > 0);

    wait_run_completed(&store, &accepted.run_id).await;
    let _run = store.get_run(&accepted.run_id).await.unwrap().unwrap();

    // The steer became an Applied queued input and a chain User entry.
    let steers = store.list_steers(&sid).await.unwrap();
    assert_eq!(steers.len(), 1);
    assert_eq!(steers[0].content, "补充: 先看缓存");
    assert!(matches!(steers[0].state, SteerState::Applied { .. }));

    let snapshot = runtime.get_snapshot(&sid).await.unwrap();
    let chain = &snapshot.chains[&main_branch_id()];
    let injected = chain.iter().find(|entry| match &entry.message {
        Message::User { blocks } => blocks.iter().any(|b| match b {
            ContentBlock::Text { text } => text.contains("补充"),
            _ => false,
        }),
        _ => false,
    });
    assert!(injected.is_some(), "steer user entry missing from chain");
    assert!(snapshot.unapplied_steers.is_empty());

    // The second provider request contained the injected user message.
    let requests = provider.requests();
    assert!(requests.len() >= 2, "run should have made a second request");
    let second = &requests[1];
    let has_injection = second.messages.iter().any(|m| match m {
        kaynine_core::provider::ModelMessage::User { blocks } => blocks.iter().any(|b| match b {
            ContentBlock::Text { text } => text.contains("补充"),
            _ => false,
        }),
        _ => false,
    });
    assert!(has_injection, "injected steer not in second request");
}

#[tokio::test]
async fn steer_without_active_run_rejected() {
    let (runtime, _store, sid, _db) = setup().await;
    let error = runtime
        .steer(SteerRequest {
            command_id: cmd(),
            session_id: sid.clone(),
            run_id: kaynine_core::ids::RunId::from("run-missing"),
            content: "hello".to_string(),
        })
        .await
        .unwrap_err();
    assert_eq!(error, KaynineError::InvalidRequest);
}

#[tokio::test]
async fn start_run_with_composer_uses_rendered_system_prompt() {
    let (runtime, store, sid, _db) = setup().await;
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("好"));
    let composer = Arc::new(kaynine_core::prompt::PromptComposer::new(
        vec![Arc::new(kaynine_core::testing::StaticLayer::new(
            "assistant",
            100,
            Some("你是测试助手"),
        ))],
        Vec::new(),
    ));
    let mut request = start_request(provider.clone(), vec![]);
    request.prompt = Some(composer);
    let accepted = runtime.start_run(request).await.expect("start run");
    wait_until_idle(&runtime, &sid, accepted.revision).await;
    let run = store
        .get_run(&accepted.run_id)
        .await
        .unwrap()
        .expect("run record");
    assert_eq!(run.state, RunState::Completed);
    let requests = provider.requests();
    assert_eq!(requests.len(), 1);
    assert!(requests[0].system_prompt.contains("你是测试助手"));
}

// ---- compaction persistence + lazy restore (SPEC §10) ----
use kaynine_core::compaction::CompactionConfig;
use kaynine_core::provider::ModelMessage;
use kaynine_core::store::{ForkOutcome, ForkRequest};
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};

fn two_step_counter(first: u64, rest: u64) -> Arc<FakeTokenCounter> {
    let call = Arc::new(AtomicU64::new(0));
    let estimator_call = call.clone();
    Arc::new(FakeTokenCounter {
        estimator: Arc::new(move |_| {
            let n = estimator_call.fetch_add(1, AtomicOrdering::SeqCst);
            if n == 0 {
                first
            } else {
                rest
            }
        }),
        source: kaynine_core::provider::TokenMeasurementSource::Heuristic,
    })
}

fn compaction_config() -> CompactionConfig {
    CompactionConfig {
        trigger_ratio: 0.5,
        keep_recent_turns: 1,
        max_attempts: 3,
    }
}

fn content(text: &str) -> Vec<ContentBlock> {
    vec![ContentBlock::Text {
        text: text.to_string(),
    }]
}

/// Run 1 (plain text) then run 2 whose huge first token count forces a
/// compaction checkpoint covering run 1's turn.
async fn seed_with_checkpoint(runtime: &AgentRuntime, store: &Arc<SqliteStore>, _sid: &SessionId) {
    let provider1 = Arc::new(FakeProvider::new(caps()));
    provider1.push_events(text_events("答1"));
    let accepted1 = runtime
        .start_run(start_request(provider1, vec![]))
        .await
        .expect("run1 accepted");
    wait_run_completed(store, &accepted1.run_id).await;

    let provider2 = Arc::new(FakeProvider::new(caps()));
    provider2.push_events(text_events("摘要"));
    provider2.push_events(text_events("答2"));
    let mut request = start_request(provider2, vec![]);
    request.content = content("第二问");
    request.compaction = Some(compaction_config());
    request.token_counter = two_step_counter(500_000, 100);
    let accepted2 = runtime.start_run(request).await.expect("run2 accepted");
    wait_run_completed(store, &accepted2.run_id).await;
}

#[tokio::test]
async fn compaction_persists_checkpoint() {
    let (runtime, store, sid, _db) = setup().await;
    let revision_before = runtime.get_snapshot(&sid).await.unwrap().current_revision;
    seed_with_checkpoint(&runtime, &store, &sid).await;

    let summaries = store.list_summaries(&main_branch_id()).await.unwrap();
    assert_eq!(summaries.len(), 1);
    assert_eq!(summaries[0].text, "摘要");
    assert_eq!(summaries[0].branch_id, main_branch_id());
    assert!(!summaries[0].source_hash.is_empty());

    let snapshot = runtime.get_snapshot(&sid).await.unwrap();
    let chain = &snapshot.chains[&main_branch_id()];
    // Run 1's user + assistant entries were covered by the checkpoint.
    assert_eq!(summaries[0].covered_until_entry, chain[1].entry_id);
    let branch = snapshot
        .branches
        .iter()
        .find(|b| b.branch_id == main_branch_id())
        .unwrap();
    assert!(!branch.needs_compaction, "checkpoint clears the flag");
    assert!(snapshot.current_revision > revision_before);
}

#[tokio::test]
async fn summary_applied_on_next_run() {
    let (runtime, store, sid, _db) = setup().await;
    seed_with_checkpoint(&runtime, &store, &sid).await;

    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("答3"));
    let accepted = runtime
        .start_run(start_request(provider.clone(), vec![]))
        .await
        .expect("run3 accepted");
    wait_run_completed(&store, &accepted.run_id).await;

    let requests = provider.requests();
    assert!(!requests.is_empty());
    match &requests[0].messages[0] {
        ModelMessage::Summary { text } => assert_eq!(text, "摘要"),
        other => panic!("restored run should open with Summary, got {other:?}"),
    }
}

#[tokio::test]
async fn fork_invalidated_summary_not_reused() {
    let (runtime, store, sid, _db) = setup().await;
    seed_with_checkpoint(&runtime, &store, &sid).await;

    let snapshot = runtime.get_snapshot(&sid).await.unwrap();
    let first_entry = snapshot.chains[&main_branch_id()][0].entry_id.clone();
    let revision = snapshot.current_revision;
    runtime
        .release_session(&sid)
        .await
        .expect("release session");

    let owner = store
        .acquire_lease(&sid, "fork-test", 60)
        .await
        .expect("acquire lease");
    let fork_branch = BranchId::from("session-1_branch_fork1");
    let outcome = store
        .fork_branch(ForkRequest {
            command: None,
            session_id: sid.clone(),
            from_branch_id: main_branch_id(),
            from_entry_id: first_entry,
            new_branch_id: fork_branch.clone(),
            expected_revision: revision,
            owner: owner.clone(),
        })
        .await
        .expect("fork");
    assert!(
        matches!(outcome, ForkOutcome::Forked { ref branch, .. } if branch.needs_compaction),
        "fork before the covered prefix must invalidate the checkpoint"
    );
    store.release_lease(&sid, &owner).await.expect("release");

    // Run on the fork with compaction forced: the stale main-branch summary
    // must NOT replace the fork's own first turn.
    let provider = Arc::new(FakeProvider::new(caps()));
    provider.push_events(text_events("新摘要"));
    provider.push_events(text_events("答"));
    let mut request = start_request(provider.clone(), vec![]);
    request.branch_id = fork_branch.clone();
    request.content = content("fork-q");
    request.compaction = Some(compaction_config());
    request.token_counter = two_step_counter(500_000, 100);
    let accepted = runtime.start_run(request).await.expect("fork run accepted");
    wait_run_completed(&store, &accepted.run_id).await;

    let requests = provider.requests();
    assert_eq!(requests.len(), 2);
    let summary_request_saw_original = requests[0].messages.iter().any(|m| match m {
        ModelMessage::User { blocks } => blocks.iter().any(|b| match b {
            ContentBlock::Text { text } => text == "hello",
            _ => false,
        }),
        _ => false,
    });
    assert!(
        summary_request_saw_original,
        "fork run must summarize the original first turn, not reuse the stale summary"
    );
    match &requests[1].messages[0] {
        ModelMessage::Summary { text } => assert_eq!(text, "新摘要"),
        other => panic!("fork run main request should open with its own Summary, got {other:?}"),
    }

    let summaries = store.list_summaries(&fork_branch).await.unwrap();
    assert_eq!(summaries.len(), 1, "fork branch gets its own checkpoint");
    assert_eq!(summaries[0].text, "新摘要");
    assert_eq!(summaries[0].branch_id, fork_branch);
}